提交 220db4f3 编写于 作者: Y Yancey1989

clean code

上级 cb8a24be
...@@ -118,7 +118,6 @@ std::unique_ptr<ir::Graph> BuildStrategy::Apply( ...@@ -118,7 +118,6 @@ std::unique_ptr<ir::Graph> BuildStrategy::Apply(
std::unique_ptr<ir::Graph> graph(new ir::Graph(main_program)); std::unique_ptr<ir::Graph> graph(new ir::Graph(main_program));
for (std::shared_ptr<ir::Pass> &pass : pass_builder_->AllPasses()) { for (std::shared_ptr<ir::Pass> &pass : pass_builder_->AllPasses()) {
VLOG(5) << "run pass: " << pass->Type();
if (pass->Type() == "multi_devices_pass") { if (pass->Type() == "multi_devices_pass") {
pass->Erase("places"); pass->Erase("places");
pass->SetNotOwned<const std::vector<platform::Place>>("places", &places); pass->SetNotOwned<const std::vector<platform::Place>>("places", &places);
......
...@@ -329,7 +329,6 @@ std::unique_ptr<ir::Graph> MultiDevSSAGraphBuilder::ApplyImpl( ...@@ -329,7 +329,6 @@ std::unique_ptr<ir::Graph> MultiDevSSAGraphBuilder::ApplyImpl(
std::unordered_map<std::string, int> sharded_var_device; std::unordered_map<std::string, int> sharded_var_device;
for (ir::Node *node : sorted_ops) { for (ir::Node *node : sorted_ops) {
VLOG(5) << "op name: " << node->Op()->Type();
if (boost::get<int>( if (boost::get<int>(
node->Op()->GetAttr(OpProtoAndCheckerMaker::OpRoleAttrName())) == node->Op()->GetAttr(OpProtoAndCheckerMaker::OpRoleAttrName())) ==
static_cast<int>(OpRole::kRPC)) { static_cast<int>(OpRole::kRPC)) {
...@@ -366,11 +365,9 @@ std::unique_ptr<ir::Graph> MultiDevSSAGraphBuilder::ApplyImpl( ...@@ -366,11 +365,9 @@ std::unique_ptr<ir::Graph> MultiDevSSAGraphBuilder::ApplyImpl(
// is true only for the op that scale the final scalar loss. // is true only for the op that scale the final scalar loss.
// It also assumes backward op will always follow the forward op in // It also assumes backward op will always follow the forward op in
// the block. // the block.
VLOG(5) << "this is loss scale op!";
is_forwarding = false; is_forwarding = false;
} else { } else {
int op_dev_id = GetOpDeviceID(result, node, sharded_var_device); int op_dev_id = GetOpDeviceID(result, node, sharded_var_device);
VLOG(5) << "on device id: " << op_dev_id;
if (op_dev_id != -1) { // This op only runs on one specific device. if (op_dev_id != -1) { // This op only runs on one specific device.
CreateComputationalOp(&result, node, op_dev_id); CreateComputationalOp(&result, node, op_dev_id);
for (ir::Node *n : node->outputs) { for (ir::Node *n : node->outputs) {
......
...@@ -20,8 +20,6 @@ limitations under the License. */ ...@@ -20,8 +20,6 @@ limitations under the License. */
#include <unordered_set> #include <unordered_set>
#include <vector> #include <vector>
#include "ThreadPool.h"
#include "paddle/fluid/framework/details/build_strategy.h" #include "paddle/fluid/framework/details/build_strategy.h"
#include "paddle/fluid/framework/details/execution_strategy.h" #include "paddle/fluid/framework/details/execution_strategy.h"
#include "paddle/fluid/framework/executor.h" #include "paddle/fluid/framework/executor.h"
......
...@@ -58,10 +58,7 @@ int64_t GetEagerDeletionThreshold() { ...@@ -58,10 +58,7 @@ int64_t GetEagerDeletionThreshold() {
(static_cast<int64_t>(1) << 30)); (static_cast<int64_t>(1) << 30));
} }
Scope::~Scope() { Scope::~Scope() { DropKids(); }
VLOG(5) << "~Scope()";
DropKids();
}
Scope& Scope::NewScope() const { Scope& Scope::NewScope() const {
SCOPE_LOCK_GUARD SCOPE_LOCK_GUARD
......
...@@ -48,18 +48,9 @@ void ThreadPool::Init() { ...@@ -48,18 +48,9 @@ void ThreadPool::Init() {
ThreadPool::ThreadPool(int num_threads) : running_(true) { ThreadPool::ThreadPool(int num_threads) : running_(true) {
threads_.resize(num_threads); threads_.resize(num_threads);
for (int i = 0; i < num_threads; ++i) { for (auto& thread : threads_) {
// for (auto& thread : threads_) {
// TODO(Yancey1989): binding the thread on the specify CPU number // TODO(Yancey1989): binding the thread on the specify CPU number
threads_[i].reset( thread.reset(new std::thread(std::bind(&ThreadPool::TaskLoop, this)));
new std::thread(std::bind(&ThreadPool::TaskLoop, this, i)));
/**
sched_param sch;
int policy;
pthread_getschedparam(threads_[i]->native_handle(), &policy, &sch);
if (pthread_setschedparam(threads_[i]->native_handle(), SCHED_FIFO, &sch)) {
VLOG(1) << "Failed to setschedparam: " << errno;
}**/
} }
} }
...@@ -77,7 +68,7 @@ ThreadPool::~ThreadPool() { ...@@ -77,7 +68,7 @@ ThreadPool::~ThreadPool() {
} }
} }
void ThreadPool::TaskLoop(int i) { void ThreadPool::TaskLoop() {
while (true) { while (true) {
Task task; Task task;
......
...@@ -99,7 +99,7 @@ class ThreadPool { ...@@ -99,7 +99,7 @@ class ThreadPool {
// The constructor starts threads to run TaskLoop, which retrieves // The constructor starts threads to run TaskLoop, which retrieves
// and runs tasks from the queue. // and runs tasks from the queue.
void TaskLoop(int i); void TaskLoop();
// Init is called by GetInstance. // Init is called by GetInstance.
static void Init(); static void Init();
......
...@@ -59,47 +59,3 @@ TEST(ThreadPool, ConcurrentRun) { ...@@ -59,47 +59,3 @@ TEST(ThreadPool, ConcurrentRun) {
} }
EXPECT_EQ(sum, ((n + 1) * n) / 2); EXPECT_EQ(sum, ((n + 1) * n) / 2);
} }
static int64_t GetTS() {
struct timeval tp;
gettimeofday(&tp, NULL);
return tp.tv_sec * 1000000 + tp.tv_usec;
}
void multi_call(std::function<void()> call) {
for (int i = 0; i < 500; ++i) {
call();
}
}
TEST(ThreadPool, PERFORMANCE) {
auto sum = [] {
int a = 0;
for (int i = 0; i < 1000; ++i) {
a += i;
}
};
// framework::ThreadPool *pool = new framework::ThreadPool(2);
int64_t start = GetTS();
for (int i = 0; i < 1000; ++i) {
// int64_t s = GetTS();
framework::Async(std::move(sum));
// pool->Run(std::move(sum));
// VLOG(5) << "push to pool spent : " << GetTS() - s << " (us).";
}
VLOG(5) << "pool spent: " << GetTS() - start << " (us).";
start = GetTS();
for (int i = 0; i < 1000; ++i) {
sum();
}
VLOG(5) << "sequence call spent: " << GetTS() - start << " (us).";
std::vector<std::thread> threads;
start = GetTS();
for (int i = 0; i < 2; ++i) {
std::thread t(multi_call, std::ref(sum));
threads.push_back(std::move(t));
}
for (auto& thread : threads) {
thread.join();
}
VLOG(5) << "two threads spent: " << GetTS() - start << " (us).";
}
...@@ -67,12 +67,9 @@ class BlockingQueue { ...@@ -67,12 +67,9 @@ class BlockingQueue {
} }
bool Receive(T* elem) { bool Receive(T* elem) {
VLOG(1) << "blocking queue::Receive ...";
std::unique_lock<std::mutex> lock(mutex_); std::unique_lock<std::mutex> lock(mutex_);
receive_cv_.wait(lock, [&] { return !queue_.empty() || closed_; }); receive_cv_.wait(lock, [&] { return !queue_.empty() || closed_; });
VLOG(1) << "queue_.empty()=" << queue_.empty();
if (!queue_.empty()) { if (!queue_.empty()) {
if (elem == nullptr) VLOG(1) << "elem is nullptr";
PADDLE_ENFORCE_NOT_NULL(elem); PADDLE_ENFORCE_NOT_NULL(elem);
*elem = queue_.front(); *elem = queue_.front();
if (LIKELY(!speed_test_mode_)) { if (LIKELY(!speed_test_mode_)) {
......
...@@ -82,13 +82,11 @@ void BufferedReader::StartImpl() { ...@@ -82,13 +82,11 @@ void BufferedReader::StartImpl() {
} }
void BufferedReader::ReadNextImpl(std::vector<framework::LoDTensor> *out) { void BufferedReader::ReadNextImpl(std::vector<framework::LoDTensor> *out) {
VLOG(1) << "ReadNextImpl start on place: " << place_;
if (position_.empty()) { if (position_.empty()) {
out->clear(); out->clear();
return; return;
} }
size_t i = position_.front().get(); size_t i = position_.front().get();
VLOG(1) << "position front: " << i;
position_.pop(); position_.pop();
if (i == -1UL) { if (i == -1UL) {
...@@ -105,7 +103,6 @@ void BufferedReader::ReadNextImpl(std::vector<framework::LoDTensor> *out) { ...@@ -105,7 +103,6 @@ void BufferedReader::ReadNextImpl(std::vector<framework::LoDTensor> *out) {
ReadAsync(prev_pos_); ReadAsync(prev_pos_);
} }
prev_pos_ = i; prev_pos_ = i;
VLOG(1) << "success ReadNextImpl";
} }
} // namespace reader } // namespace reader
......
...@@ -25,15 +25,9 @@ class CreateDoubleBufferReaderOp : public framework::OperatorBase { ...@@ -25,15 +25,9 @@ class CreateDoubleBufferReaderOp : public framework::OperatorBase {
private: private:
void RunImpl(const framework::Scope& scope, void RunImpl(const framework::Scope& scope,
const platform::Place& dev_place) const override { const platform::Place& dev_place) const override {
VLOG(1) << "find var in scope: " << &scope; auto* out = scope.Var(Output("Out"))
auto* out_var = scope.FindVar(Output("Out")); ->template GetMutable<framework::ReaderHolder>();
VLOG(1) << "var " << Output("Out") << " -> " << out_var;
auto* out = out_var->GetMutable<framework::ReaderHolder>();
// auto* out = scope.Var(Output("Out"))
// ->template GetMutable<framework::ReaderHolder>();
if (out->Get() != nullptr) { if (out->Get() != nullptr) {
VLOG(1) << Output("Out") << " is not nullptr.";
return; return;
} }
const auto& underlying_reader = scope.FindVar(Input("UnderlyingReader")) const auto& underlying_reader = scope.FindVar(Input("UnderlyingReader"))
...@@ -52,11 +46,8 @@ class CreateDoubleBufferReaderOp : public framework::OperatorBase { ...@@ -52,11 +46,8 @@ class CreateDoubleBufferReaderOp : public framework::OperatorBase {
sin >> num; sin >> num;
place = platform::CUDAPlace(static_cast<int>(num)); place = platform::CUDAPlace(static_cast<int>(num));
} }
VLOG(1) << "create buffered reader on " << place;
out->Reset(framework::MakeDecoratedReader<BufferedReader>(underlying_reader, out->Reset(framework::MakeDecoratedReader<BufferedReader>(underlying_reader,
place, 2)); place, 2));
VLOG(1) << "Reset Buffered Reader in var: "
<< scope.FindVar(Input("UnderlyingReader"));
} }
}; };
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册