提交 0e9f1e27 编写于 作者: F fengjiayi

Make ReaderBase thread safe and remove ThreadedReader

上级 0d2ccfbd
...@@ -18,11 +18,13 @@ namespace paddle { ...@@ -18,11 +18,13 @@ namespace paddle {
namespace framework { namespace framework {
void ReaderBase::ReadNext(std::vector<LoDTensor> *out) { void ReaderBase::ReadNext(std::vector<LoDTensor> *out) {
std::lock_guard<std::mutex> lock(mu_);
PADDLE_ENFORCE_EQ(status_, ReaderStatus::kRunning); PADDLE_ENFORCE_EQ(status_, ReaderStatus::kRunning);
ReadNextImpl(out); ReadNextImpl(out);
} }
void ReaderBase::Shutdown() { void ReaderBase::Shutdown() {
std::lock_guard<std::mutex> lock(mu_);
if (status_ != ReaderStatus::kStopped) { if (status_ != ReaderStatus::kStopped) {
ShutdownImpl(); ShutdownImpl();
status_ = ReaderStatus::kStopped; status_ = ReaderStatus::kStopped;
...@@ -30,6 +32,7 @@ void ReaderBase::Shutdown() { ...@@ -30,6 +32,7 @@ void ReaderBase::Shutdown() {
} }
void ReaderBase::Start() { void ReaderBase::Start() {
std::lock_guard<std::mutex> lock(mu_);
if (status_ != ReaderStatus::kRunning) { if (status_ != ReaderStatus::kRunning) {
StartImpl(); StartImpl();
status_ = ReaderStatus::kRunning; status_ = ReaderStatus::kRunning;
......
...@@ -44,6 +44,8 @@ class ReaderBase { ...@@ -44,6 +44,8 @@ class ReaderBase {
virtual void StartImpl() = 0; virtual void StartImpl() = 0;
ReaderStatus status_{kRunning}; ReaderStatus status_{kRunning};
mutable std::mutex mu_;
}; };
class DecoratedReader : public ReaderBase { class DecoratedReader : public ReaderBase {
......
...@@ -22,7 +22,6 @@ reader_library(create_batch_reader_op SRCS create_batch_reader_op.cc) ...@@ -22,7 +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_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_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_multi_pass_reader_op SRCS create_multi_pass_reader_op.cc)
reader_library(create_threaded_reader_op SRCS create_threaded_reader_op.cc)
reader_library(create_custom_reader_op SRCS create_custom_reader_op.cc) reader_library(create_custom_reader_op SRCS create_custom_reader_op.cc)
reader_library(create_py_reader_op SRCS create_py_reader_op.cc) reader_library(create_py_reader_op SRCS create_py_reader_op.cc)
......
// Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#include "paddle/fluid/operators/detail/safe_ref.h"
#include "paddle/fluid/operators/reader/reader_op_registry.h"
namespace paddle {
namespace operators {
namespace reader {
class ThreadedReader : public framework::DecoratedReader {
public:
explicit ThreadedReader(const std::shared_ptr<ReaderBase>& reader)
: DecoratedReader(reader) {}
void ReadNextImpl(std::vector<framework::LoDTensor>* out) override {
std::lock_guard<std::mutex> lock(mutex_);
reader_->ReadNext(out);
}
private:
void ShutdownImpl() override {
std::lock_guard<std::mutex> lock(mutex_);
reader_->Shutdown();
}
void StartImpl() override {
std::lock_guard<std::mutex> lock(mutex_);
reader_->Start();
}
std::mutex mutex_;
};
class CreateThreadedReaderOp : public framework::OperatorBase {
public:
using framework::OperatorBase::OperatorBase;
private:
void RunImpl(const framework::Scope& scope,
const platform::Place& dev_place) const override {
auto* out = detail::Ref(scope.FindVar(Output("Out")))
.GetMutable<framework::ReaderHolder>();
if (out->Get() != nullptr) {
return;
}
const auto& underlying_reader = scope.FindVar(Input("UnderlyingReader"))
->Get<framework::ReaderHolder>();
out->Reset(new ThreadedReader(underlying_reader.Get()));
}
};
class CreateThreadedReaderOpMaker : public DecoratedReaderMakerBase {
protected:
void Apply() override {
AddComment(R"DOC(
CreateThreadedReader Operator
This operator creates a threaded reader. A threaded reader's
'ReadNext()' can be invoked by several threads at the same
time.
)DOC");
}
};
} // namespace reader
} // namespace operators
} // namespace paddle
namespace reader = paddle::operators::reader;
REGISTER_DECORATED_READER_OPERATOR(create_threaded_reader,
reader::CreateThreadedReaderOp,
reader::CreateThreadedReaderOpMaker);
...@@ -529,9 +529,6 @@ def open_files(filenames, ...@@ -529,9 +529,6 @@ def open_files(filenames,
main_prog_reader = multi_pass( main_prog_reader = multi_pass(
reader=main_prog_reader, pass_num=pass_num) reader=main_prog_reader, pass_num=pass_num)
if for_parallel:
main_prog_reader = parallel(reader=main_prog_reader)
return monkey_patch_reader_methods(main_prog_reader) return monkey_patch_reader_methods(main_prog_reader)
...@@ -647,11 +644,6 @@ def multi_pass(reader, pass_num): ...@@ -647,11 +644,6 @@ def multi_pass(reader, pass_num):
'create_multi_pass_reader', reader, {'pass_num': int(pass_num)}) 'create_multi_pass_reader', reader, {'pass_num': int(pass_num)})
def parallel(reader):
return __create_shared_decorated_reader__('create_threaded_reader', reader,
{})
def read_file(reader): def read_file(reader):
""" """
Execute the given reader and get data via it. Execute the given reader and get data via it.
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册