未验证 提交 9171aaa0 编写于 作者: W WangXi 提交者: GitHub

[fleet_executor] remove SetCreatingFlag (#38539)

上级 48f061fb
...@@ -73,7 +73,6 @@ class Carrier final { ...@@ -73,7 +73,6 @@ class Carrier final {
Interceptor* SetInterceptor(int64_t interceptor_id, Interceptor* SetInterceptor(int64_t interceptor_id,
std::unique_ptr<Interceptor>); std::unique_ptr<Interceptor>);
void SetCreatingFlag(bool flag) {}
void SetMsgBus(const std::shared_ptr<MessageBus>& msg_bus) { void SetMsgBus(const std::shared_ptr<MessageBus>& msg_bus) {
msg_bus_ = msg_bus; msg_bus_ = msg_bus;
} }
......
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#include <vector> #include <vector>
#include "paddle/fluid/framework/blocking_queue.h" #include "paddle/fluid/framework/blocking_queue.h"
#include "paddle/fluid/platform/macros.h"
namespace paddle { namespace paddle {
namespace distributed { namespace distributed {
...@@ -66,6 +67,8 @@ class TaskLoop { ...@@ -66,6 +67,8 @@ class TaskLoop {
} }
private: private:
DISABLE_COPY_AND_ASSIGN(TaskLoop);
void AbortNotInLoopThread(); void AbortNotInLoopThread();
static thread_local TaskLoop* thread_local_loop_; static thread_local TaskLoop* thread_local_loop_;
......
...@@ -18,6 +18,8 @@ ...@@ -18,6 +18,8 @@
#include <mutex> #include <mutex>
#include <thread> #include <thread>
#include "paddle/fluid/platform/macros.h"
namespace paddle { namespace paddle {
namespace distributed { namespace distributed {
...@@ -31,6 +33,8 @@ class TaskLoopThread { ...@@ -31,6 +33,8 @@ class TaskLoopThread {
TaskLoop* StartLoop(); TaskLoop* StartLoop();
private: private:
DISABLE_COPY_AND_ASSIGN(TaskLoopThread);
void Loop(); void Loop();
bool start_; bool start_;
......
...@@ -17,6 +17,8 @@ ...@@ -17,6 +17,8 @@
#include <memory> #include <memory>
#include <vector> #include <vector>
#include "paddle/fluid/platform/macros.h"
namespace paddle { namespace paddle {
namespace distributed { namespace distributed {
...@@ -37,6 +39,8 @@ class TaskLoopThreadPool { ...@@ -37,6 +39,8 @@ class TaskLoopThreadPool {
std::vector<TaskLoop*> GetAllLoops(); std::vector<TaskLoop*> GetAllLoops();
private: private:
DISABLE_COPY_AND_ASSIGN(TaskLoopThreadPool);
bool start_; bool start_;
int thread_num_; int thread_num_;
std::vector<std::unique_ptr<TaskLoopThread>> threads_; std::vector<std::unique_ptr<TaskLoopThread>> threads_;
......
...@@ -84,8 +84,6 @@ TEST(ComputeInterceptor, Compute) { ...@@ -84,8 +84,6 @@ TEST(ComputeInterceptor, Compute) {
a->SetPlace(place); a->SetPlace(place);
a->SetMicroBatchScope(scopes); a->SetMicroBatchScope(scopes);
carrier.SetCreatingFlag(false);
// start // start
InterceptorMessage msg; InterceptorMessage msg;
msg.set_message_type(DATA_IS_READY); msg.set_message_type(DATA_IS_READY);
......
...@@ -69,8 +69,6 @@ TEST(ComputeInterceptor, Compute) { ...@@ -69,8 +69,6 @@ TEST(ComputeInterceptor, Compute) {
carrier.SetInterceptor(1, InterceptorFactory::Create("Compute", 1, node_b)); carrier.SetInterceptor(1, InterceptorFactory::Create("Compute", 1, node_b));
carrier.SetInterceptor(2, InterceptorFactory::Create("Compute", 2, node_c)); carrier.SetInterceptor(2, InterceptorFactory::Create("Compute", 2, node_c));
carrier.SetCreatingFlag(false);
InterceptorMessage msg; InterceptorMessage msg;
msg.set_message_type(DATA_IS_READY); msg.set_message_type(DATA_IS_READY);
// test run three times // test run three times
......
...@@ -68,7 +68,6 @@ TEST(InterceptorTest, PingPong) { ...@@ -68,7 +68,6 @@ TEST(InterceptorTest, PingPong) {
0, InterceptorFactory::Create("PingPong", 0, nullptr)); 0, InterceptorFactory::Create("PingPong", 0, nullptr));
carrier.SetInterceptor(1, std::make_unique<PingPongInterceptor>(1, nullptr)); carrier.SetInterceptor(1, std::make_unique<PingPongInterceptor>(1, nullptr));
carrier.SetCreatingFlag(false);
InterceptorMessage msg; InterceptorMessage msg;
a->Send(1, msg); a->Send(1, msg);
......
...@@ -113,7 +113,6 @@ TEST(InterceptorTest, PingPong) { ...@@ -113,7 +113,6 @@ TEST(InterceptorTest, PingPong) {
if (pid == 0) { if (pid == 0) {
Carrier* carrier = Carrier* carrier =
FleetExecutor::CreateCarrier(0, interceptor_id_to_rank); FleetExecutor::CreateCarrier(0, interceptor_id_to_rank);
carrier->SetCreatingFlag(false);
auto msg_bus = std::make_shared<MessageBus>(); auto msg_bus = std::make_shared<MessageBus>();
carrier->SetMsgBus(msg_bus); carrier->SetMsgBus(msg_bus);
// NOTE: need Init msg_bus after carrier SetMsgBus // NOTE: need Init msg_bus after carrier SetMsgBus
...@@ -128,7 +127,6 @@ TEST(InterceptorTest, PingPong) { ...@@ -128,7 +127,6 @@ TEST(InterceptorTest, PingPong) {
} else { } else {
Carrier* carrier = Carrier* carrier =
FleetExecutor::CreateCarrier(1, interceptor_id_to_rank); FleetExecutor::CreateCarrier(1, interceptor_id_to_rank);
carrier->SetCreatingFlag(false);
auto msg_bus = std::make_shared<MessageBus>(); auto msg_bus = std::make_shared<MessageBus>();
carrier->SetMsgBus(msg_bus); carrier->SetMsgBus(msg_bus);
msg_bus->Init(1, {{0, ip0}, {1, ip1}}, ip1); msg_bus->Init(1, {{0, ip0}, {1, ip1}}, ip1);
......
...@@ -80,8 +80,6 @@ TEST(AmplifierInterceptor, Amplifier) { ...@@ -80,8 +80,6 @@ TEST(AmplifierInterceptor, Amplifier) {
carrier.SetInterceptor(4, InterceptorFactory::Create("Amplifier", 4, node_e)); carrier.SetInterceptor(4, InterceptorFactory::Create("Amplifier", 4, node_e));
carrier.SetInterceptor(5, InterceptorFactory::Create("Compute", 5, node_f)); carrier.SetInterceptor(5, InterceptorFactory::Create("Compute", 5, node_f));
carrier.SetCreatingFlag(false);
// start // start
InterceptorMessage msg; InterceptorMessage msg;
msg.set_message_type(DATA_IS_READY); msg.set_message_type(DATA_IS_READY);
......
...@@ -96,8 +96,6 @@ TEST(AmplifierInterceptor, Amplifier) { ...@@ -96,8 +96,6 @@ TEST(AmplifierInterceptor, Amplifier) {
carrier.SetInterceptor(2, InterceptorFactory::Create("Compute", 2, node_c)); carrier.SetInterceptor(2, InterceptorFactory::Create("Compute", 2, node_c));
carrier.SetInterceptor(3, InterceptorFactory::Create("Amplifier", 3, node_d)); carrier.SetInterceptor(3, InterceptorFactory::Create("Amplifier", 3, node_d));
carrier.SetCreatingFlag(false);
// start // start
InterceptorMessage msg; InterceptorMessage msg;
msg.set_message_type(DATA_IS_READY); msg.set_message_type(DATA_IS_READY);
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册