Change code

paddle/fluid/operators/reader/buffered_reader.cc

@@ -28,15 +28,15 @@ BufferedReader::BufferedReader(
       buffer_size_(buffer_size) {
   cpu_buffer_.resize(buffer_size);
   gpu_buffer_.resize(buffer_size);
-  AppendFutureToBatchSize();
+  ReadTillBufferFullAsync();
 }
-void BufferedReader::AppendFutureToBatchSize() {
+void BufferedReader::ReadTillBufferFullAsync() {
   PADDLE_ENFORCE_EQ(position_.size(), 0U);
   for (size_t i = 0; i < buffer_size_; ++i) {
-    AppendFuture(i);
+    ReadAsync(i);
   }
 }
-void BufferedReader::AppendFuture(size_t i) {
+void BufferedReader::ReadAsync(size_t i) {
   position_.emplace(thread_pool_.enqueue([this, i]() -> size_t {
     TensorVec &cpu = cpu_buffer_[i];
     reader_->ReadNext(&cpu);
@@ -50,6 +50,7 @@ void BufferedReader::AppendFuture(size_t i) {
       gpu.resize(cpu.size());
       for (size_t i = 0; i < cpu.size(); ++i) {
         framework::TensorCopySync(cpu[i], place_, &gpu[i]);
+        gpu[i].set_lod(cpu[i].lod());
       }
     }
     return i;
@@ -60,10 +61,11 @@ void BufferedReader::ShutdownImpl() {
   while (!position_.empty()) {
     position_.pop();
   }
+  prev_pos_ = -1UL;
 }
 void BufferedReader::StartImpl() {
   reader_->Start();
-  AppendFutureToBatchSize();
+  ReadTillBufferFullAsync();
 }
 void BufferedReader::ReadNextImpl(std::vector<framework::LoDTensor> *out) {
   if (position_.empty()) {
@@ -79,7 +81,14 @@ void BufferedReader::ReadNextImpl(std::vector<framework::LoDTensor> *out) {
   }
 
   *out = platform::is_gpu_place(place_) ? gpu_buffer_[i] : cpu_buffer_[i];
-  AppendFuture(i);
+
+  // Do not push current position into ReadAsync. Push the previous position
+  // Since all computation in fluid are async, change the data of
+  // current position may cause data error.
+  if (prev_pos_ != -1Ul) {
+    ReadAsync(prev_pos_);
+  }
+  prev_pos_ = i;
 }
 
 }  // namespace reader

paddle/fluid/operators/reader/buffered_reader.h

@@ -35,9 +35,9 @@ class BufferedReader : public framework::DecoratedReader {
   ~BufferedReader() override;
 
  private:
-  void AppendFutureToBatchSize();
+  void ReadTillBufferFullAsync();
 
-  void AppendFuture(size_t i);
+  void ReadAsync(size_t i);
 
  protected:
   void ShutdownImpl() override;
@@ -50,8 +50,15 @@ class BufferedReader : public framework::DecoratedReader {
   const size_t buffer_size_;
 
   std::queue<std::future<size_t>> position_;
+
+  // The buffer for reading data.
+  // NOTE: the simplest way to implement buffered reader is do not use any
+  // buffer, just async read and create futures as buffer size. However, to
+  // malloc Tensor every time is extremely slow. Here we store all data in
+  // buffers and prevent alloc every time.
   std::vector<TensorVec> cpu_buffer_;
   std::vector<TensorVec> gpu_buffer_;
+  size_t prev_pos_{-1UL};
 };
 
 }  // namespace reader

python/paddle/fluid/tests/unittests/test_py_reader_push_pop.py

@@ -45,12 +45,12 @@ class TestPyReader(unittest.TestCase):
             ) else fluid.CPUPlace()
             executor = fluid.Executor(place)
 
-            data_file, feed_queue = fluid.layers.py_reader(
+            data_file = fluid.layers.py_reader(
                 capacity=self.capacity,
                 dtypes=self.dtypes,
                 lod_levels=self.lod_levels,
                 shapes=self.shapes)
-
+            feed_queue = data_file.queue
             read_out_data = fluid.layers.read_file(data_file)
             self.inputs = []
 

python/paddle/fluid/tests/unittests/test_py_reader_using_executor.py

@@ -52,11 +52,12 @@ def simple_fc_net(in_size,
                   batch_size,
                   queue_capacity,
                   use_double_buffer=False):
-    reader, feed_queue = fluid.layers.py_reader(
+    reader = fluid.layers.py_reader(
         capacity=queue_capacity,
         shapes=[[-1, in_size], [-1, 1]],
         lod_levels=[0, 0],
         dtypes=['float32', 'int64'])
+    feed_queue = reader.queue
     reader = fluid.layers.batch(reader, batch_size=batch_size)
     if use_double_buffer:
         reader = fluid.layers.double_buffer(reader)