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

Author: paulinagacek

cmake/experiments/cuda_module_loading_lazy.cmake

@@ -31,6 +31,11 @@ if(LINUX)
     message("cuda 11.7+ already support lazy module loading")
     return()
   endif()
+  if(${CUDA_VERSION} VERSION_LESS "11.2" AND ${CMAKE_CXX_COMPILER_VERSION}
+                                             VERSION_GREATER_EQUAL 12.0)
+    message("cuda less than 11.2 doesn't support gcc12")
+    return()
+  endif()
 
   message(
     "for cuda before 11.7, libcudart.so must be used for the lazy module loading trick to work, instead of libcudart_static.a"

cmake/external/gloo.cmake

@@ -31,6 +31,17 @@ set(GLOO_LIBRARIES
     "${GLOO_INSTALL_DIR}/lib/libgloo.a"
     CACHE FILEPATH "gloo library." FORCE)
 
+set(GLOO_PATCH_COMMAND "")
+if(WITH_GPU)
+  if(${CMAKE_CUDA_COMPILER_VERSION} LESS 12.0 AND ${CMAKE_CXX_COMPILER_VERSION}
+                                                  VERSION_GREATER 12.0)
+    file(TO_NATIVE_PATH ${PADDLE_SOURCE_DIR}/patches/gloo/device.cc.patch
+         native_dst)
+    set(GLOO_PATCH_COMMAND patch -d ${GLOO_SOURCE_DIR}/gloo/transport/tcp <
+                           ${native_dst})
+  endif()
+endif()
+
 include_directories(${GLOO_INCLUDE_DIR})
 
 if(WITH_ASCEND OR WITH_ASCEND_CL)
@@ -59,6 +70,7 @@ else()
     GIT_TAG ${GLOO_TAG}
     PREFIX "${GLOO_PREFIX_DIR}"
     UPDATE_COMMAND ""
+    PATCH_COMMAND ${GLOO_PATCH_COMMAND}
     CONFIGURE_COMMAND ""
     BUILD_COMMAND
       mkdir -p ${GLOO_SOURCE_DIR}/build && cd ${GLOO_SOURCE_DIR}/build && cmake

cmake/external/protobuf.cmake

@@ -250,6 +250,12 @@ function(build_protobuf TARGET_NAME BUILD_FOR_HOST)
   else()
     set(PROTOBUF_REPOSITORY ${GIT_URL}/protocolbuffers/protobuf.git)
     set(PROTOBUF_TAG 9f75c5aa851cd877fb0d93ccc31b8567a6706546)
+    if(WITH_GPU)
+      if(${CMAKE_CUDA_COMPILER_VERSION} LESS 12.0
+         AND ${CMAKE_CXX_COMPILER_VERSION} VERSION_GREATER 12.0)
+        set(PROTOBUF_TAG 2dc747c574b68a808ea4699d26942c8132fe2b09)
+      endif()
+    endif()
   endif()
   if(WITH_ARM_BRPC)
     set(ARM_PROTOBUF_URL
@@ -322,6 +328,12 @@ elseif(WITH_ARM_BRPC)
   set(PROTOBUF_VERSION 3.7.1-baidu-ee-common)
 else()
   set(PROTOBUF_VERSION 3.1.0)
+  if(WITH_GPU)
+    if(${CMAKE_CUDA_COMPILER_VERSION} LESS 12.0
+       AND ${CMAKE_CXX_COMPILER_VERSION} VERSION_GREATER 12.0)
+      set(PROTOBUF_VERSION 3.16.0)
+    endif()
+  endif()
 endif()
 
 if(NOT PROTOBUF_FOUND)

cmake/external/warpctc.cmake

@@ -25,6 +25,19 @@ set(WARPCTC_INSTALL_DIR ${THIRD_PARTY_PATH}/install/warpctc)
 set(WARPCTC_REPOSITORY ${GIT_URL}/baidu-research/warp-ctc.git)
 set(WARPCTC_TAG bdc2b4550453e0ef2d3b5190f9c6103a84eff184)
 
+set(WARPCTC_SOURCE_DIR ${THIRD_PARTY_PATH}/warpctc/src/extern_warpctc)
+set(WARPCTC_PATCH_COMMAND "")
+set(WARPCTC_CCBIN_OPTION "")
+if(NOT WIN32 AND WITH_GPU)
+  if(${CMAKE_CUDA_COMPILER_VERSION} LESS 12.0 AND ${CMAKE_CXX_COMPILER_VERSION}
+                                                  VERSION_GREATER 12.0)
+    file(TO_NATIVE_PATH
+         ${PADDLE_SOURCE_DIR}/patches/warpctc/CMakeLists.txt.patch native_src)
+    set(WARPCTC_PATCH_COMMAND patch -d ${WARPCTC_SOURCE_DIR} < ${native_src})
+    set(WARPCTC_CCBIN_OPTION -DCCBIN_COMPILER=${CCBIN_COMPILER})
+  endif()
+endif()
+
 set(WARPCTC_INCLUDE_DIR
     "${WARPCTC_INSTALL_DIR}/include"
     CACHE PATH "Warp-ctc Directory" FORCE)
@@ -112,7 +125,7 @@ else()
     GIT_TAG ${WARPCTC_TAG}
     PREFIX ${WARPCTC_PREFIX_DIR}
     UPDATE_COMMAND ""
-    PATCH_COMMAND ""
+    PATCH_COMMAND ${WARPCTC_PATCH_COMMAND}
     #BUILD_ALWAYS    1
     CMAKE_ARGS -DCMAKE_CXX_COMPILER=${CMAKE_CXX_COMPILER}
                -DCMAKE_C_COMPILER=${CMAKE_C_COMPILER}
@@ -132,7 +145,9 @@ else()
                -DBUILD_TESTS=OFF
                -DCMAKE_POSITION_INDEPENDENT_CODE=ON
                -DCMAKE_BUILD_TYPE=${THIRD_PARTY_BUILD_TYPE}
+               -DCUDA_TOOLKIT_ROOT_DIR=${CUDA_TOOLKIT_ROOT_DIR}
                ${EXTERNAL_OPTIONAL_ARGS}
+               ${WARPCTC_CCBIN_OPTION}
     CMAKE_CACHE_ARGS
       -DCMAKE_BUILD_TYPE:STRING=${THIRD_PARTY_BUILD_TYPE}
       -DCMAKE_POSITION_INDEPENDENT_CODE:BOOL=ON

paddle/fluid/distributed/fleet_executor/CMakeLists.txt

@@ -37,6 +37,7 @@ cc_library(
        compute_interceptor.cc
        amplifier_interceptor.cc
        cond_interceptor.cc
+       start_interceptor.cc
        source_interceptor.cc
        sink_interceptor.cc
        message_service.cc
@@ -69,6 +70,8 @@ if(WITH_DISTRIBUTE)
                                         ${DISTRIBUTE_COMPILE_FLAGS})
   set_source_files_properties(
     cond_interceptor.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
+  set_source_files_properties(
+    start_interceptor.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
   set_source_files_properties(
     source_interceptor.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
   set_source_files_properties(

paddle/fluid/distributed/fleet_executor/carrier.cc

@@ -36,6 +36,7 @@ USE_INTERCEPTOR(Compute);
 USE_INTERCEPTOR(Amplifier);
 USE_INTERCEPTOR(Sink);
 USE_INTERCEPTOR(Cond);
+USE_INTERCEPTOR(Start);
 
 void Carrier::Init(
     int64_t rank,

paddle/fluid/distributed/fleet_executor/compute_interceptor.cc

@@ -18,6 +18,7 @@
 #include "paddle/fluid/distributed/fleet_executor/task_node.h"
 #include "paddle/fluid/framework/executor_gc_helper.h"
 #include "paddle/fluid/framework/operator.h"
+#include "paddle/phi/core/errors.h"
 
 namespace paddle {
 namespace distributed {
@@ -33,23 +34,30 @@ void ComputeInterceptor::PrepareDeps() {
   auto& downstream = node_->downstream();
 
   for (auto up : upstream) {
-    in_readys_.emplace(up.first, std::make_pair(up.second, 0));
+    std::map<int64_t, int64_t> ready_size_map;
+    for (int64_t i = 0; i < node_->max_run_times(); ++i) {
+      ready_size_map.emplace(i, 0);
+    }
+    in_readys_.emplace(up.first, std::make_pair(up.second, ready_size_map));
   }
   for (auto down : downstream) {
     out_buffs_.emplace(down.first, std::make_pair(down.second, 0));
   }
 }
 
-void ComputeInterceptor::IncreaseReady(int64_t up_id) {
+void ComputeInterceptor::IncreaseReady(int64_t up_id, int64_t scope_id) {
   auto it = in_readys_.find(up_id);
   PADDLE_ENFORCE_NE(it,
                     in_readys_.end(),
                     platform::errors::NotFound(
                         "Cannot find upstream=%lld in in_readys.", up_id));
 
   auto max_ready_size = it->second.first;
-  auto ready_size = it->second.second;
-  ready_size += 1;
+  const auto& ready_scope_map = it->second.second;
+  int64_t ready_size = 0;
+  for (auto& scope_iter : ready_scope_map) {
+    ready_size += scope_iter.second;
+  }
   if (max_ready_size != INFINITE_BUFFER_SIZE) {
     PADDLE_ENFORCE_LE(
         ready_size,
@@ -61,7 +69,14 @@ void ComputeInterceptor::IncreaseReady(int64_t up_id) {
             ready_size,
             max_ready_size));
   }
-  it->second.second = ready_size;
+  PADDLE_ENFORCE_NE(
+      it->second.second.find(scope_id),
+      it->second.second.end(),
+      platform::errors::OutOfRange(
+          "Interceptor %lld can not find scope %lld in upstream ready map",
+          interceptor_id_,
+          scope_id));
+  it->second.second.at(scope_id) = ready_scope_map.at(scope_id) + 1;
 }
 
 void ComputeInterceptor::DecreaseBuff(int64_t down_id) {
@@ -83,16 +98,21 @@ void ComputeInterceptor::DecreaseBuff(int64_t down_id) {
 }
 
 bool ComputeInterceptor::IsInputReady() {
-  for (auto& ins : in_readys_) {
-    auto ready_size = ins.second.second;
-    // not ready, return false
-    if (ready_size == 0) {
-      VLOG(3) << "Interceptor " << GetInterceptorId()
+  for (int64_t i = 0; i < node_->max_run_times(); ++i) {
+    bool flag = true;
+    for (auto& ins : in_readys_) {
+      auto ready_size_map = ins.second.second;
+      flag = flag && (ready_size_map.at(i) != 0);
+    }
+    if (flag) {
+      cur_scope_id_ = i;
+      return true;
+    } else {
+      VLOG(3) << "Interceptor " << GetInterceptorId() << " in scope " << i
               << "'s upstreams aren't all ready.";
-      return false;
     }
   }
-  return true;
+  return false;
 }
 
 bool ComputeInterceptor::CanWriteOutput() {
@@ -144,7 +164,7 @@ void ComputeInterceptor::SendDataReadyToDownStream() {
 void ComputeInterceptor::ReplyCompletedToUpStream() {
   for (auto& ins : in_readys_) {
     auto up_id = ins.first;
-    auto ready_size = ins.second.second;
+    auto ready_size = ins.second.second.at(cur_scope_id_);
     ready_size -= 1;
     PADDLE_ENFORCE_GE(
         ready_size,
@@ -153,7 +173,7 @@ void ComputeInterceptor::ReplyCompletedToUpStream() {
             "upstream=%lld ready_size must >= 0, but now got %lld",
             up_id,
             ready_size));
-    ins.second.second = ready_size;
+    ins.second.second[cur_scope_id_] = ready_size;
 
     VLOG(3) << "ComputeInterceptor " << interceptor_id_
             << " Reply data_is_useless msg to " << up_id
@@ -187,11 +207,8 @@ void ComputeInterceptor::RunOps() {
 
 void ComputeInterceptor::Run() {
   while (IsInputReady() && CanWriteOutput()) {
-    VLOG(3) << "id=" << GetInterceptorId() << " ComputeInterceptor running";
-
-    // get the ready scope id from queue
-    cur_scope_id_ = ready_queue_.front();
-    ready_queue_.pop();
+    VLOG(0) << "id=" << GetInterceptorId()
+            << " ComputeInterceptor running in scope " << cur_scope_id_;
 
     RunOps();
 
@@ -204,10 +221,15 @@ void ComputeInterceptor::Run() {
 
 void ComputeInterceptor::Compute(const InterceptorMessage& msg) {
   if (msg.message_type() == DATA_IS_READY) {
-    IncreaseReady(msg.src_id());
-    ready_queue_.push(msg.scope_idx());
+    VLOG(3) << "Compute interceptor " << interceptor_id_
+            << " receive data_is_ready " << msg.src_id() << " "
+            << msg.scope_idx() << " ";
+    IncreaseReady(msg.src_id(), msg.scope_idx());
     Run();
   } else if (msg.message_type() == DATA_IS_USELESS) {
+    VLOG(3) << "Compute interceptor " << interceptor_id_
+            << " receive data_is_useless " << msg.src_id() << " "
+            << msg.scope_idx() << " ";
     DecreaseBuff(msg.src_id());
     Run();
   }

paddle/fluid/distributed/fleet_executor/compute_interceptor.h

@@ -32,25 +32,24 @@ class ComputeInterceptor : public Interceptor {
   virtual void RunOps();
   virtual void SendDataReadyToDownStream();
   virtual void ReplyCompletedToUpStream();
+  virtual void Compute(const InterceptorMessage& msg);
+  void Run();
+  void IncreaseReady(int64_t up_id, int64_t scope_id);
+  void DecreaseBuff(int64_t down_id);
 
-  std::queue<int64_t> ready_queue_;
   int64_t cur_scope_id_;
 
+  // upstream_id-->(max_ready_size, scope-->ready_size)
+  std::map<int64_t, std::pair<int64_t, std::map<int64_t, int64_t>>>
+      in_readys_{};
+  // downstream_id-->(max_buffer_size, used_size)
+  std::map<int64_t, std::pair<int64_t, int64_t>> out_buffs_{};
+
  private:
   void PrepareDeps();
 
-  void IncreaseReady(int64_t up_id);
-  void DecreaseBuff(int64_t down_id);
   bool IsInputReady();
   bool CanWriteOutput();
-
-  void Run();
-  void Compute(const InterceptorMessage& msg);
-
-  // upstream_id-->(max_ready_size, ready_size)
-  std::map<int64_t, std::pair<int64_t, int64_t>> in_readys_{};
-  // downstream_id-->(max_buffer_size, used_size)
-  std::map<int64_t, std::pair<int64_t, int64_t>> out_buffs_{};
 };
 
 }  // namespace distributed

paddle/fluid/distributed/fleet_executor/cond_interceptor.cc

@@ -98,8 +98,6 @@ void CondInterceptor::ReplyDataIsUseless(int64_t up_id) {
 }
 
 void CondInterceptor::Compute() {
-  cur_scope_id_ = ready_queue_.front();
-  ready_queue_.pop();
   bool cond = GetCondResult();
   VLOG(3) << "Cond interceptor get condition var " << node_->cond_var()
           << " with value " << cond;
@@ -109,14 +107,14 @@ void CondInterceptor::Compute() {
       SendDataReady(down_id);
     }
   } else {
-    VLOG(3) << "Finish loop in scope " << cur_scope_id_;
+    VLOG(0) << "Finish loop in scope " << cur_scope_id_;
     SendDataReady(stop_loop_id_);
   }
 }
 
 void CondInterceptor::Run(const InterceptorMessage& msg) {
   if (msg.message_type() == DATA_IS_READY) {
-    ready_queue_.push(msg.scope_idx());
+    cur_scope_id_ = msg.scope_idx();
     Compute();
   } else if (msg.message_type() == DATA_IS_USELESS) {
     if (node_->id_to_dep_type().at(msg.src_id()) == DependType::STOP_LOOP) {

paddle/fluid/distributed/fleet_executor/cond_interceptor.h

@@ -39,7 +39,6 @@ class CondInterceptor final : public Interceptor {
   void SendDataReady(int64_t down_id);
   void ReplyDataIsUseless(int64_t up_id);
 
-  std::queue<int64_t> ready_queue_;
   int64_t cur_scope_id_;
 
   std::set<int64_t> normal_in_id_;