Impl ResetAll and fix errors

Author: JiayiFeng

paddle/fluid/framework/reader.cc

@@ -26,7 +26,7 @@ void ReaderBase::ReadNext(std::vector<LoDTensor> *out) {
 
 void ReaderBase::InsertDecoratedReader(
     const std::shared_ptr<ReaderBase> &decorated_reader) {
-  std::lock_guard<std::mutex> guard(mu_));
+  std::lock_guard<std::mutex> guard(mu_);
   decorated_readers_.emplace_back(decorated_reader);
 }
 

paddle/fluid/framework/reader.h

@@ -104,7 +104,13 @@ class ReaderHolder {
   }
 
   void ResetAll() {
-    // TODO(fengjiayi): The interface of reseting all.
+    auto end_readers = reader_->GetEndPoints();
+    for (auto* reader : end_readers) {
+      reader->Shutdown();
+    }
+    for (auto* reader : end_readers) {
+      reader->Start();
+    }
   }
 
   void Shutdown() {

paddle/fluid/framework/reader_test.cc

@@ -21,13 +21,12 @@ class StubDecoratedReader : public paddle::framework::DecoratedReader {
   explicit StubDecoratedReader(const std::shared_ptr<ReaderBase> &reader)
       : DecoratedReader(reader) {}
 
-  void ReadNext(std::vector<paddle::framework::LoDTensor> *out) override {}
+  void ReadNextImpl(std::vector<paddle::framework::LoDTensor> *out) override {}
 };
 
 class StubRootReader : public paddle::framework::ReaderBase {
  public:
-  void ReadNext(std::vector<paddle::framework::LoDTensor> *out) override {}
-  void ReInit() override {}
+  void ReadNextImpl(std::vector<paddle::framework::LoDTensor> *out) override {}
 };
 
 TEST(READER, decorate_chain) {

python/paddle/fluid/layers/io.py

@@ -375,9 +375,6 @@ def open_recordio_file(filename,
     if pass_num > 1:
         main_prog_var = multi_pass(reader=main_prog_var, pass_num=pass_num)
 
-    if for_parallel:
-        main_prog_var = parallel(reader=main_prog_var)
-
     return monkey_patch_reader_methods(main_prog_var)