Merge branch 'develop' into dropout_opt_clean_BcInT

Author: zhangboSJTU

cmake/operators.cmake

@@ -62,7 +62,6 @@ function(op_library TARGET)
   set(hip_cc_srcs)
   set(xpu_cc_srcs)
   set(xpu_kp_cc_srcs)
-  set(mlu_cc_srcs)
   set(cudnn_cu_cc_srcs)
   set(miopen_cu_cc_srcs)
   set(cudnn_cu_srcs)
@@ -307,9 +306,8 @@ function(op_library TARGET)
     # Unity Build relies on global option `WITH_UNITY_BUILD` and local option `UNITY`.
     if(WITH_UNITY_BUILD AND op_library_UNITY)
       # Combine the cc source files.
-      compose_unity_target_sources(
-        ${UNITY_TARGET} cc ${cc_srcs} ${mkldnn_cc_srcs} ${xpu_cc_srcs}
-        ${mlu_cc_srcs})
+      compose_unity_target_sources(${UNITY_TARGET} cc ${cc_srcs}
+                                   ${mkldnn_cc_srcs} ${xpu_cc_srcs})
       if(TARGET ${UNITY_TARGET})
         # If `UNITY_TARGET` exists, add source files to `UNITY_TARGET`.
         target_sources(${UNITY_TARGET} PRIVATE ${unity_target_cc_sources})
@@ -325,7 +323,7 @@ function(op_library TARGET)
     else()
       cc_library(
         ${TARGET}
-        SRCS ${cc_srcs} ${mkldnn_cc_srcs} ${xpu_cc_srcs} ${mlu_cc_srcs}
+        SRCS ${cc_srcs} ${mkldnn_cc_srcs} ${xpu_cc_srcs}
         DEPS ${op_library_DEPS} ${op_common_deps})
     endif()
   endif()
@@ -337,7 +335,6 @@ function(op_library TARGET)
   list(LENGTH mkldnn_cc_srcs mkldnn_cc_srcs_len)
   list(LENGTH xpu_cc_srcs xpu_cc_srcs_len)
   list(LENGTH miopen_cu_cc_srcs miopen_cu_cc_srcs_len)
-  list(LENGTH mlu_cc_srcs mlu_cc_srcs_len)
 
   # Define operators that don't need pybind here.
   foreach(
@@ -562,7 +559,6 @@ function(register_operators)
     "*_op.cc")
   string(REPLACE "_mkldnn" "" OPS "${OPS}")
   string(REPLACE "_xpu" "" OPS "${OPS}")
-  string(REPLACE "_mlu" "" OPS "${OPS}")
   string(REPLACE ".cc" "" OPS "${OPS}")
   list(REMOVE_DUPLICATES OPS)
   list(LENGTH register_operators_DEPS register_operators_DEPS_len)

cmake/phi_header.cmake

@@ -17,21 +17,24 @@ set(PADDLE_INFERENCE_INSTALL_DIR
 
 function(phi_header_path_compat TARGET_PATH)
   message(STATUS "phi header path compat processing: ${TARGET_PATH}")
-  file(GLOB HEADERS "${TARGET_PATH}/*" "*.h")
-  foreach(header ${HEADERS})
-    if(${header} MATCHES ".*.h$")
-      file(READ ${header} HEADER_CONTENT)
-      string(REPLACE "paddle/phi/" "paddle/include/experimental/phi/"
-                     HEADER_CONTENT "${HEADER_CONTENT}")
-      string(REPLACE "paddle/fluid/platform/"
-                     "paddle/include/experimental/phi/" HEADER_CONTENT
-                     "${HEADER_CONTENT}")
-      string(REPLACE "paddle/utils/" "paddle/include/experimental/utils/"
-                     HEADER_CONTENT "${HEADER_CONTENT}")
-      file(WRITE ${header} "${HEADER_CONTENT}")
-      message(STATUS "phi header path compat processing complete: ${header}")
-    endif()
-  endforeach()
+  string(FIND ${TARGET_PATH} "experimental" pos)
+  if(pos GREATER 1)
+    file(GLOB HEADERS "${TARGET_PATH}/*" "*.h")
+    foreach(header ${HEADERS})
+      if(${header} MATCHES ".*.h$")
+        file(READ ${header} HEADER_CONTENT)
+        string(REPLACE "paddle/phi/" "paddle/include/experimental/phi/"
+                       HEADER_CONTENT "${HEADER_CONTENT}")
+        string(REPLACE "paddle/fluid/platform/"
+                       "paddle/include/experimental/phi/" HEADER_CONTENT
+                       "${HEADER_CONTENT}")
+        string(REPLACE "paddle/utils/" "paddle/include/experimental/utils/"
+                       HEADER_CONTENT "${HEADER_CONTENT}")
+        file(WRITE ${header} "${HEADER_CONTENT}")
+        message(STATUS "phi header path compat processing complete: ${header}")
+      endif()
+    endforeach()
+  endif()
 endfunction()
 
 phi_header_path_compat(
@@ -48,7 +51,16 @@ phi_header_path_compat(
   ${PADDLE_INFERENCE_INSTALL_DIR}/paddle/include/experimental/phi/common)
 phi_header_path_compat(
   ${PADDLE_INFERENCE_INSTALL_DIR}/paddle/include/experimental/phi/core)
-phi_header_path_compat(${PADDLE_INFERENCE_INSTALL_DIR}/paddle/include/)
+
+# NOTE(liuyuanle): In inference lib, no need include paddle/utils/pybind.h, so we delete this.
+file(READ
+     ${PADDLE_INFERENCE_INSTALL_DIR}/paddle/include/experimental/extension.h
+     HEADER_CONTENT)
+string(REGEX REPLACE "#if !defined\\(PADDLE_ON_INFERENCE\\).*#endif" ""
+                     HEADER_CONTENT "${HEADER_CONTENT}")
+file(WRITE
+     ${PADDLE_INFERENCE_INSTALL_DIR}/paddle/include/experimental/extension.h
+     "${HEADER_CONTENT}")
 
 # In order to be compatible with the original behavior, the header file name needs to be changed
 file(RENAME

paddle/extension.h

@@ -16,8 +16,8 @@ limitations under the License. */
 
 // All paddle apis in C++ frontend
 #include "paddle/phi/api/all.h"
-// Python bindings for the C++ frontend (includes Python.h)
 #if !defined(PADDLE_ON_INFERENCE) && !defined(PADDLE_NO_PYTHON)
+// Python bindings for the C++ frontend (includes Python.h)
 #include "paddle/utils/pybind.h"
 #endif
 // For initialization of DeviceContextPool and MemoryMethod

paddle/fluid/distributed/fleet_executor/carrier.cc

@@ -17,6 +17,7 @@
 #include <algorithm>
 #include <vector>
 
+#include "gflags/gflags.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"
@@ -237,12 +238,10 @@ bool Carrier::Send(const InterceptorMessage& msg) {
     VLOG(3) << "Send a message from interceptor " << src_id
             << " to interceptor " << dst_id << ", which are in the same ranks.";
     return EnqueueInterceptorMessage(msg);
-  } else {
-    VLOG(3) << "Send a message from interceptor " << src_id
-            << " to interceptor " << dst_id
-            << ", which are in different ranks.";
-    return GlobalVal<MessageBus>::Get()->Send(dst_rank, msg);
   }
+  VLOG(3) << "Send a message from interceptor " << src_id << " to interceptor "
+          << dst_id << ", which are in different ranks.";
+  return GlobalVal<MessageBus>::Get()->Send(dst_rank, msg);
 }
 
 Interceptor* Carrier::SetInterceptor(int64_t interceptor_id,

paddle/fluid/distributed/fleet_executor/compute_interceptor.cc

@@ -158,25 +158,47 @@ void ComputeInterceptor::DecreaseBuff(int64_t down_id) {
 }
 
 bool ComputeInterceptor::IsInputReady() {
-  for (int64_t i = 0; i < node_->max_run_times(); ++i) {
+  std::map<int64_t, bool> scope_id_to_finish_flag;
+  if (!gen_step_to_scope_id_to_finish_flag_.empty()) {
+    scope_id_to_finish_flag =
+        gen_step_to_scope_id_to_finish_flag_.begin()->second;
+    VLOG(3) << "Is Input Ready in gen step "
+            << gen_step_to_scope_id_to_finish_flag_.begin()->first;
+  }
+  int64_t num_micro_step =
+      (num_micro_step_ == -1 ? node_->max_run_times() : num_micro_step_);
+  int64_t start_micro_step = (start_micro_step_ == -1 ? 0 : start_micro_step_);
+  for (int64_t i = start_micro_step; i < start_micro_step + num_micro_step;
+       ++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) {
-      for (auto iter : scope_id_to_finish_flag_) {
-        if (iter.first == i) {
-          break;
-        } else if (!iter.second) {
-          VLOG(3) << "The previous scope is not ready, waiting for the "
-                     "previous scope "
-                  << iter.first;
-          return false;
+      if (scope_id_to_finish_flag.empty()) {
+        cur_scope_id_ = i;
+        return true;
+      } else if (scope_id_to_finish_flag.find(i) !=
+                 scope_id_to_finish_flag.end()) {
+        for (auto iter : scope_id_to_finish_flag) {
+          if (iter.first == i) {
+            break;
+          } else if (!iter.second) {
+            VLOG(3) << "The previous scope is not ready, waiting for the "
+                       "previous scope "
+                    << iter.first << " in gen_step "
+                    << gen_step_to_scope_id_to_finish_flag_.begin()->first;
+            return false;
+          }
         }
+        cur_scope_id_ = i;
+        return true;
+      } else {
+        VLOG(3) << "Interceptor " << GetInterceptorId() << " in scope " << i
+                << " is larger than gen_step "
+                << gen_step_to_scope_id_to_finish_flag_.begin()->first;
       }
-      cur_scope_id_ = i;
-      return true;
     } else {
       VLOG(3) << "Interceptor " << GetInterceptorId() << " in scope " << i
               << "'s upstreams aren't all ready.";
@@ -203,6 +225,16 @@ bool ComputeInterceptor::CanWriteOutput() {
 }
 
 void ComputeInterceptor::SendDataReadyToDownStream() {
+  bool need_send_vars = !(node_->vars_to_dtype().empty());
+  InterceptorMessage ready_msg;
+  ready_msg.set_start_micro_step(start_micro_step_);
+  ready_msg.set_num_micro_step(num_micro_step_);
+  if (need_send_vars) {
+    ready_msg = PrepareVarsMsg();
+  } else {
+    ready_msg.set_message_type(DATA_IS_READY);
+    ready_msg.set_scope_idx(cur_scope_id_);
+  }
   for (auto& outs : out_buffs_) {
     auto down_id = outs.first;
     auto max_buff_size = outs.second.first;
@@ -221,13 +253,17 @@ void ComputeInterceptor::SendDataReadyToDownStream() {
     }
     outs.second.second = used_size;
 
-    InterceptorMessage ready_msg;
-    ready_msg.set_message_type(DATA_IS_READY);
-    ready_msg.set_scope_idx(cur_scope_id_);
-    VLOG(3) << "ComputeInterceptor " << interceptor_id_
-            << " Send data_is_ready msg to " << down_id
-            << " in scope: " << cur_scope_id_;
-    Send(down_id, ready_msg);
+    if (need_send_vars) {
+      VLOG(3) << "ComputeInterceptor " << interceptor_id_
+              << " Send data_with_vars msg to " << down_id
+              << " in scope: " << cur_scope_id_;
+      Send(down_id, ready_msg);
+    } else {
+      VLOG(3) << "ComputeInterceptor " << interceptor_id_
+              << " Send data_is_ready msg to " << down_id
+              << " in scope: " << cur_scope_id_;
+      Send(down_id, ready_msg);
+    }
   }
 }
 
@@ -289,13 +325,21 @@ void ComputeInterceptor::Run() {
 
     RunOps();
 
-    if (!scope_id_to_finish_flag_.empty()) {
+    if (!gen_step_to_scope_id_to_finish_flag_.empty()) {
+      auto iter = gen_step_to_scope_id_to_finish_flag_.begin();
+      VLOG(3) << "id=" << GetInterceptorId()
+              << " ComputeInterceptor running in scope " << cur_scope_id_
+              << " with gen_step " << iter->first;
+      auto& scope_id_to_finish_flag = iter->second;
       PADDLE_ENFORCE_NE(
-          scope_id_to_finish_flag_.find(cur_scope_id_),
-          scope_id_to_finish_flag_.end(),
+          scope_id_to_finish_flag.find(cur_scope_id_),
+          scope_id_to_finish_flag.end(),
           platform::errors::NotFound(
               "Can not find scope %ld in scope_id_to_finish", cur_scope_id_));
-      scope_id_to_finish_flag_.erase(cur_scope_id_);
+      scope_id_to_finish_flag.erase(cur_scope_id_);
+      if (scope_id_to_finish_flag.empty()) {
+        gen_step_to_scope_id_to_finish_flag_.erase(iter);
+      }
     }
 
     // send to downstream and increase buff used
@@ -310,6 +354,8 @@ void ComputeInterceptor::Compute(const InterceptorMessage& msg) {
     VLOG(3) << "Compute interceptor " << interceptor_id_
             << " receive data_is_ready " << msg.src_id() << " "
             << msg.scope_idx() << " ";
+    start_micro_step_ = msg.start_micro_step();
+    num_micro_step_ = msg.num_micro_step();
     IncreaseReady(msg.src_id(), msg.scope_idx());
     Run();
   } else if (msg.message_type() == DATA_IS_USELESS) {
@@ -327,10 +373,14 @@ void ComputeInterceptor::Compute(const InterceptorMessage& msg) {
     Run();
   } else if (msg.message_type() == START_LOOP) {
     VLOG(3) << "Compute interceptor " << interceptor_id_
-            << " receive start_loop " << msg.src_id() << " " << msg.scope_idx()
-            << " ";
+            << " receive start_loop " << msg.src_id() << " in scope "
+            << msg.scope_idx() << " with gen_step " << msg.gen_step();
+    start_micro_step_ = msg.start_micro_step();
+    num_micro_step_ = msg.num_micro_step();
     IncreaseReady(msg.src_id(), msg.scope_idx());
-    scope_id_to_finish_flag_.emplace(msg.scope_idx(), false);
+    int64_t gen_step = msg.gen_step();
+    gen_step_to_scope_id_to_finish_flag_[gen_step].emplace(msg.scope_idx(),
+                                                           false);
     Run();
   }
 }

paddle/fluid/distributed/fleet_executor/compute_interceptor.h

@@ -52,7 +52,10 @@ class ComputeInterceptor : public Interceptor {
 
   bool IsInputReady();
   bool CanWriteOutput();
-  std::map<int64_t, bool> scope_id_to_finish_flag_;
+  std::map<int64_t, std::map<int64_t, bool>>
+      gen_step_to_scope_id_to_finish_flag_;
+  int64_t start_micro_step_{-1};
+  int64_t num_micro_step_{-1};
 };
 
 }  // namespace distributed