Merge branch 'develop' of https://github.com/PaddlePaddle/Paddle into pten_matmul_grad

Author: zyfncg

paddle/fluid/distributed/fleet_executor/CMakeLists.txt

@@ -19,6 +19,9 @@ cc_library(fleet_executor SRCS fleet_executor.cc carrier.cc task_node.cc runtime
 
 if(WITH_DISTRIBUTE)
  set(DISTRIBUTE_COMPILE_FLAGS "-Wno-non-virtual-dtor -Wno-error=non-virtual-dtor -Wno-error=delete-non-virtual-dtor")
+ if (CMAKE_CXX_COMPILER_VERSION VERSION_GREATER 7.0)
+ set(DISTRIBUTE_COMPILE_FLAGS "${DISTRIBUTE_COMPILE_FLAGS} -faligned-new")
+ endif()
  set_source_files_properties(interceptor.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
  set_source_files_properties(compute_interceptor.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
  set_source_files_properties(amplifier_interceptor.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})

paddle/fluid/distributed/fleet_executor/carrier.cc

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 #include "paddle/fluid/distributed/fleet_executor/carrier.h"
-#include "paddle/fluid/distributed/fleet_executor/global_map.h"
+#include "paddle/fluid/distributed/fleet_executor/global.h"
 #include "paddle/fluid/distributed/fleet_executor/interceptor.h"
 #include "paddle/fluid/distributed/fleet_executor/interceptor_message_service.h"
 #include "paddle/fluid/distributed/fleet_executor/message_bus.h"
@@ -71,17 +71,13 @@ Carrier::~Carrier() { VLOG(3) << "Carrier's destructor."; }
 
 bool Carrier::EnqueueInterceptorMessage(
  const InterceptorMessage& interceptor_message) {
- if (interceptor_message.ctrl_message()) {
- VLOG(3) << "Receiving control message from rank "
- << interceptor_message.src_id() << " to rank "
- << interceptor_message.dst_id();
- // for barrier
- msg_bus_->IncreaseBarrierCount();
- } else {
- int64_t dst_id = interceptor_message.dst_id();
- Interceptor* dst_interceptor = GetInterceptor(dst_id);
- dst_interceptor->EnqueueRemoteInterceptorMessage(interceptor_message);
- }
+ PADDLE_ENFORCE_EQ(
+ interceptor_message.ctrl_message(), false,
+ platform::errors::Fatal(
+ "Control message should be only send inter rank using message bus."));
+ int64_t dst_id = interceptor_message.dst_id();
+ Interceptor* dst_interceptor = GetInterceptor(dst_id);
+ dst_interceptor->EnqueueRemoteInterceptorMessage(interceptor_message);
  return true;
 }
 
@@ -106,11 +102,6 @@ void Carrier::WakeUp() {
 }
 
 void Carrier::Start() {
- PADDLE_ENFORCE_EQ(msg_bus_->IsInit(), true,
- platform::errors::PreconditionNotMet(
- "Using message bus since it has not been initialized. "
- "Please invoke MessageBus::Init() before using it or "
- "neccessary components are not ready."));
  PADDLE_ENFORCE_EQ(is_init_, true, platform::errors::PreconditionNotMet(
  "Using carrier before initialized."));
  for (int64_t id : source_interceptor_ids_) {
@@ -154,19 +145,10 @@ bool Carrier::Send(const InterceptorMessage& msg) {
  << " to interceptor " << dst_id << ", which are in the same ranks.";
  return EnqueueInterceptorMessage(msg);
  } else {
- PADDLE_ENFORCE_NOT_NULL(
- msg_bus_.get(),
- platform::errors::Unavailable("Message bus is released accidently"));
- PADDLE_ENFORCE_EQ(
- msg_bus_->IsInit(), true,
- platform::errors::PreconditionNotMet(
- "Using message bus since it has not been initialized. "
- "Please invoke MessageBus::Init() before using it or "
- "neccessary components are not ready."));
  VLOG(3) << "Send a message from interceptor " << src_id
  << " to interceptor " << dst_id
  << ", which are in different ranks.";
- return msg_bus_->Send(dst_rank, msg);
+ return GlobalVal<MessageBus>::Get()->Send(dst_rank, msg);
  }
 }
 

paddle/fluid/distributed/fleet_executor/carrier.h

@@ -73,10 +73,6 @@ class Carrier final {
  Interceptor* SetInterceptor(int64_t interceptor_id,
  std::unique_ptr<Interceptor>);
 
- void SetMsgBus(const std::shared_ptr<MessageBus>& msg_bus) {
- msg_bus_ = msg_bus;
- }
-
  void Start();
 
  bool IsInit() const;
@@ -107,7 +103,6 @@ class Carrier final {
  framework::Scope* minibatch_scope_;
  paddle::platform::Place place_;
  paddle::platform::DeviceContext* dev_ctx_{nullptr};
- std::shared_ptr<MessageBus> msg_bus_;
  int64_t rank_;
  std::string carrier_id_;
  std::unordered_map<int64_t, TaskNode*> interceptor_id_to_node_;

paddle/fluid/distributed/fleet_executor/fleet_executor.cc

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 #include "paddle/fluid/distributed/fleet_executor/fleet_executor.h"
-#include "paddle/fluid/distributed/fleet_executor/global_map.h"
+#include "paddle/fluid/distributed/fleet_executor/global.h"
 #include "paddle/fluid/distributed/fleet_executor/message_bus.h"
 #include "paddle/fluid/distributed/fleet_executor/runtime_graph.h"
 #include "paddle/fluid/distributed/fleet_executor/task_node.h"
@@ -32,6 +32,9 @@ FleetExecutor::FleetExecutor(const std::string& exe_desc_str) {
  bool parse_flag = exe_desc_.ParseFromString(exe_desc_str);
  PADDLE_ENFORCE(parse_flag, platform::errors::PreconditionNotMet(
  "Error occurs while parsing string to proto"));
+ // Message bus will be created and inited only once
+ GlobalVal<MessageBus>::Create();
+ InitMessageBus();
 }
 
 FleetExecutor::~FleetExecutor() {
@@ -81,21 +84,16 @@ void FleetExecutor::Init(
  CopyParameters(i, program_desc);
  }
  VLOG(5) << runtime_graph_->DebugString();
- msg_bus_ = std::make_shared<MessageBus>();
  Carrier* carrier =
  GlobalMap<std::string, Carrier>::Create(carrier_id, carrier_id);
  carrier_ids_.insert(carrier_id);
- GlobalVal<std::string>::Set(carrier_id);
- // TODO(liyurui): Maybe message bus should be created only once
+ // Set current running carrier
+ GlobalVal<std::string>::Set(new std::string(carrier_id));
  InitCarrier(carrier);
- InitMessageBus();
-
- // Wait for all message bus connected.
- msg_bus_->Barrier();
+ GlobalVal<MessageBus>::Get()->Barrier();
 }
 
 void FleetExecutor::InitCarrier(Carrier* carrier) {
- carrier->SetMsgBus(msg_bus_);
  carrier->Init(exe_desc_.cur_rank(), runtime_graph_->interceptor_id_to_rank(),
  runtime_graph_->interceptor_id_to_node(), root_scope_,
  minibatch_scope_, microbatch_scopes_, place_);
@@ -131,14 +129,18 @@ void FleetExecutor::InitMessageBus() {
  VLOG(3) << "The number of ranks are "
  << (rank_to_addr.size() == 0 ? 1 : rank_to_addr.size()) << ".";
  VLOG(5) << ss.str();
- if (!msg_bus_->IsInit()) {
- msg_bus_->Init(cur_rank, rank_to_addr, addr);
- }
+ GlobalVal<MessageBus>::Get()->Init(cur_rank, rank_to_addr, addr);
 }
 
 void FleetExecutor::Run(const std::string& carrier_id) {
- GlobalMap<std::string, Carrier>::Get(carrier_id)->Start();
- GlobalVal<std::string>::Set(carrier_id);
+ Carrier* carrier = GlobalMap<std::string, Carrier>::Get(carrier_id);
+ // Set current running carrier
+ if (*GlobalVal<std::string>::Get() != carrier_id) {
+ GlobalVal<std::string>::Set(new std::string(carrier_id));
+ // TODO(liyurui): Move barrier to service
+ GlobalVal<MessageBus>::Get()->Barrier();
+ }
+ carrier->Start();
  for (auto* micro_scop : microbatch_scopes_) {
  // By default, we should delete all kid scopes after run executor because
  // some operators may create local scope when running, such as while_op.

paddle/fluid/distributed/fleet_executor/fleet_executor.h

@@ -55,9 +55,6 @@ class FleetExecutor final {
  framework::Scope* minibatch_scope_;
  platform::Place place_;
  std::vector<framework::Scope*> microbatch_scopes_;
- // The carriers under FleetExecutor will share message bus,
- // using shared_ptr to manage lifetime and condition race.
- std::shared_ptr<MessageBus> msg_bus_;
  std::unordered_set<std::string> carrier_ids_;
 };
 

paddle/fluid/distributed/fleet_executor/global_map.h renamed to paddle/fluid/distributed/fleet_executor/global.h

@@ -14,24 +14,41 @@
 
 #pragma once
 
+#include "paddle/fluid/platform/enforce.h"
+
 namespace paddle {
 namespace distributed {
 
-// TODO(liyurui): Change this file to global.h
 template <typename T>
 class GlobalVal final {
  public:
- static T Get() { return *GetPtr(); }
- static T Set(T val) {
- auto* ptr = GetPtr();
- *ptr = val;
- return val;
+ static T* Get() {
+ T* ptr = GetPPtr()->get();
+ PADDLE_ENFORCE_NOT_NULL(
+ ptr, platform::errors::NotFound("This value is not global value."));
+ return ptr;
+ }
+ template <typename... Args>
+ static T* Create(Args&&... args) {
+ auto* ptr = GetPPtr();
+ PADDLE_ENFORCE_EQ(ptr->get(), nullptr,
+ platform::errors::AlreadyExists(
+ "This value is already a global value."));
+ T* item = new T(std::forward<Args>(args)...);
+ ptr->reset(item);
+ return item;
+ }
+
+ static T* Set(T* new_item) {
+ auto* ptr = GetPPtr();
+ ptr->reset(new_item);
+ return ptr->get();
  }
 
  private:
- static T* GetPtr() {
- static T value;
- return &value;
+ static std::unique_ptr<T>* GetPPtr() {
+ static std::unique_ptr<T> ptr;
+ return &ptr;
  }
 };
 

paddle/fluid/distributed/fleet_executor/interceptor_message_service.cc

@@ -15,8 +15,8 @@
  !defined(PADDLE_WITH_ASCEND_CL)
 #include "paddle/fluid/distributed/fleet_executor/interceptor_message_service.h"
 #include "brpc/server.h"
-#include "paddle/fluid/distributed/fleet_executor/carrier.h"
-#include "paddle/fluid/distributed/fleet_executor/global_map.h"
+#include "paddle/fluid/distributed/fleet_executor/global.h"
+#include "paddle/fluid/distributed/fleet_executor/message_bus.h"
 
 namespace paddle {
 namespace distributed {
@@ -29,9 +29,7 @@ void InterceptorMessageServiceImpl::InterceptorMessageService(
  VLOG(3) << "Interceptor Message Service receives a message from interceptor "
  << request->src_id() << " to interceptor " << request->dst_id()
  << ", with the message: " << request->message_type();
- const auto& carrier_id = GlobalVal<std::string>::Get();
- bool flag = GlobalMap<std::string, Carrier>::Get(carrier_id)
- ->EnqueueInterceptorMessage(*request);
+ bool flag = GlobalVal<MessageBus>::Get()->DispatchMsgToCarrier(*request);
  response->set_rst(flag);
 }
 

paddle/fluid/distributed/fleet_executor/message_bus.cc

@@ -17,6 +17,8 @@
 #include <set>
 #include <thread>
 
+#include "paddle/fluid/distributed/fleet_executor/carrier.h"
+#include "paddle/fluid/distributed/fleet_executor/global.h"
 #include "paddle/fluid/distributed/fleet_executor/message_bus.h"
 #include "paddle/fluid/platform/gen_comm_id_helper.h"
 
@@ -81,6 +83,10 @@ const std::string& MessageBus::GetAddr(int64_t rank) const {
 
 bool MessageBus::Send(int64_t dst_rank,
  const InterceptorMessage& interceptor_message) {
+ PADDLE_ENFORCE_EQ(
+ IsInit(), true,
+ platform::errors::PreconditionNotMet(
+ "Using message bus since it has not been initialized."));
 #if defined(PADDLE_WITH_DISTRIBUTE) && defined(PADDLE_WITH_PSCORE) && \
  !defined(PADDLE_WITH_ASCEND_CL)
  int retry_time = 0; // message bus will retry sending for 10 times
@@ -155,6 +161,22 @@ void MessageBus::Barrier() {
  }
 }
 
+bool MessageBus::DispatchMsgToCarrier(
+ const InterceptorMessage& interceptor_message) {
+ if (interceptor_message.ctrl_message()) {
+ VLOG(3) << "Receiving control message from rank "
+ << interceptor_message.src_id() << " to rank "
+ << interceptor_message.dst_id();
+ // for barrier
+ IncreaseBarrierCount();
+ return true;
+ } else {
+ const std::string& carrier_id = *GlobalVal<std::string>::Get();
+ return GlobalMap<std::string, Carrier>::Get(carrier_id)
+ ->EnqueueInterceptorMessage(interceptor_message);
+ }
+}
+
 void MessageBus::ListenPort() {
  if (addr_ == "") {
  LOG(INFO) << "No need listen to port since training on single card.";

paddle/fluid/distributed/fleet_executor/message_bus.h

@@ -54,6 +54,7 @@ class MessageBus final {
 
  void IncreaseBarrierCount();
  void Barrier();
+ bool DispatchMsgToCarrier(const InterceptorMessage& interceptor_message);
 
  private:
  DISABLE_COPY_AND_ASSIGN(MessageBus);

paddle/fluid/distributed/fleet_executor/test/compute_interceptor_run_op_test.cc

@@ -18,7 +18,7 @@ limitations under the License. */
 #include "gtest/gtest.h"
 
 #include "paddle/fluid/distributed/fleet_executor/carrier.h"
-#include "paddle/fluid/distributed/fleet_executor/global_map.h"
+#include "paddle/fluid/distributed/fleet_executor/global.h"
 #include "paddle/fluid/distributed/fleet_executor/interceptor.h"
 #include "paddle/fluid/distributed/fleet_executor/message_bus.h"
 #include "paddle/fluid/distributed/fleet_executor/task_node.h"
@@ -67,9 +67,8 @@ TEST(ComputeInterceptor, Compute) {
  GlobalMap<std::string, Carrier>::Create(carrier_id, carrier_id);
  carrier->Init(0, {{0, 0}, {1, 0}});
 
- auto msg_bus = std::make_shared<MessageBus>();
+ MessageBus* msg_bus = GlobalVal<MessageBus>::Create();
  msg_bus->Init(0, {{0, "127.0.0.0:0"}}, "");
- carrier->SetMsgBus(msg_bus);
 
  // FIXME: don't delete, otherwise interceptor will use undefined node
  TaskNode* node_a =