Merge pull request #12273 from JiayiFeng/update_py_reader

Some enhancement on readers

paddle/fluid/operators/reader/buffered_reader.cc

@@ -18,7 +18,14 @@
 namespace paddle {
 namespace operators {
 namespace reader {
-BufferedReader::~BufferedReader() { reader_->Shutdown(); }
+BufferedReader::~BufferedReader() {
+  reader_->Shutdown();
+  while (!position_.empty()) {
+    position_.front().wait();
+    position_.pop();
+  }
+}
+
 BufferedReader::BufferedReader(
     const std::shared_ptr<framework::ReaderBase> &reader,
     const platform::Place &place, size_t buffer_size)
@@ -30,12 +37,14 @@ BufferedReader::BufferedReader(
   gpu_buffer_.resize(buffer_size);
   ReadTillBufferFullAsync();
 }
+
 void BufferedReader::ReadTillBufferFullAsync() {
   PADDLE_ENFORCE_EQ(position_.size(), 0U);
   for (size_t i = 0; i < buffer_size_; ++i) {
     ReadAsync(i);
   }
 }
+
 void BufferedReader::ReadAsync(size_t i) {
   position_.emplace(thread_pool_.enqueue([this, i]() -> size_t {
     TensorVec &cpu = cpu_buffer_[i];
@@ -56,6 +65,7 @@ void BufferedReader::ReadAsync(size_t i) {
     return i;
   }));
 }
+
 void BufferedReader::ShutdownImpl() {
   reader_->Shutdown();
   while (!position_.empty()) {
@@ -63,10 +73,12 @@ void BufferedReader::ShutdownImpl() {
   }
   prev_pos_ = -1UL;
 }
+
 void BufferedReader::StartImpl() {
   reader_->Start();
   ReadTillBufferFullAsync();
 }
+
 void BufferedReader::ReadNextImpl(std::vector<framework::LoDTensor> *out) {
   if (position_.empty()) {
     out->clear();

python/paddle/fluid/layers/io.py

@@ -579,6 +579,7 @@ def py_reader(capacity,
             feed_queue.close()
 
         reader.thread = threading.Thread(target=__provider_thread__)
+        reader.thread.daemon = True
         reader.thread.start()
 
     def __set_tensor_provider__(func):

python/paddle/fluid/tests/demo/pyreader.py

@@ -25,7 +25,8 @@ def network(is_train):
         capacity=10,
         shapes=((-1, 784), (-1, 1)),
         dtypes=('float32', 'int64'),
-        name="train_reader" if is_train else "test_reader")
+        name="train_reader" if is_train else "test_reader",
+        use_double_buffer=True)
     img, label = fluid.layers.read_file(reader)
 
     hidden = img
@@ -56,14 +57,16 @@ def main():
         with fluid.unique_name.guard():
             test_loss, test_reader = network(False)
 
-    fluid.Executor(fluid.CUDAPlace(0)).run(startup_prog)
-    fluid.Executor(fluid.CUDAPlace(0)).run(test_startup)
+    use_cuda = fluid.core.is_compiled_with_cuda()
+    place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
+    fluid.Executor(place).run(startup_prog)
+    fluid.Executor(place).run(test_startup)
 
     trainer = fluid.ParallelExecutor(
-        use_cuda=True, loss_name=loss.name, main_program=train_prog)
+        use_cuda=use_cuda, loss_name=loss.name, main_program=train_prog)
 
     tester = fluid.ParallelExecutor(
-        use_cuda=True, share_vars_from=trainer, main_program=test_prog)
+        use_cuda=use_cuda, share_vars_from=trainer, main_program=test_prog)
 
     train_reader.decorate_paddle_reader(
         paddle.v2.reader.shuffle(