Complete threaded reader

paddle/fluid/operators/reader/CMakeLists.txt

@@ -22,5 +22,6 @@ reader_library(create_batch_reader_op SRCS create_batch_reader_op.cc)
 reader_library(create_recordio_file_reader_op SRCS create_recordio_file_reader_op.cc)
 reader_library(create_double_buffer_reader_op SRCS create_double_buffer_reader_op.cc)
 reader_library(create_multi_pass_reader_op SRCS create_multi_pass_reader_op.cc)
+reader_library(create_threaded_reader_op SRCS create_threaded_reader_op.cc)
 # Export local libraries to parent
 set(READER_LIBRARY ${LOCAL_READER_LIBS} PARENT_SCOPE)

paddle/fluid/operators/reader/create_threaded_reader_op.cc

@@ -57,7 +57,15 @@ class ThreadedReader : public framework::DecoratedReader {
     return !threda_buffer.empty();
   }
 
-  void ReInit() override;
+  void ReInit() override {
+    if (!unsafe_mode_) {
+      PADDLE_THROW(
+          "ThreadedReader::ReInit() is disabled when 'unsafe_mode' is false.");
+    }
+    VLOG(5) << "ThreadedReader::ReInit() is invoked! It might be buggy in "
+               "multi-thread environment.";
+    reader_->ReInit();
+  }
 
   ~ThreadedReader() {
     for (auto& p : thread_buffers_) {
@@ -123,3 +131,8 @@ class CreateThreadedReaderOpMaker : public DecoratedReaderMakerBase {
 }  // namespace reader
 }  // namespace operators
 }  // namespace paddle
+
+namespace reader = paddle::operators::reader;
+REGISTER_FILE_READER_OPERATOR(create_threaded_reader,
+                              reader::CreateThreadedReaderOp,
+                              reader::CreateThreadedReaderOpMaker);