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

Author: jim19930609

CMakeLists.txt

@@ -43,6 +43,7 @@ option(WITH_ONEMKL      "Compile PaddlePaddle with oneMKL"              OFF)
 option(WITH_GPU         "Compile PaddlePaddle with NVIDIA GPU"          ${CUDA_FOUND})
 option(WITH_TENSORRT    "Compile PaddlePaddle with NVIDIA TensorRT"     OFF)
 option(WITH_XPU         "Compile PaddlePaddle with BAIDU KUNLUN XPU"    OFF)
+option(WITH_MLU    "Compile PaddlePaddle with CAMBRICON MLU"     OFF)
 option(WITH_WIN_DUMP_DBG "Compile with windows core dump debug mode"    OFF)
 option(WITH_ASCEND         "Compile PaddlePaddle with ASCEND"        OFF)
 option(WITH_ROCM        "Compile PaddlePaddle with ROCM platform"       OFF)
@@ -64,6 +65,9 @@ endif()
 if (WITH_GPU AND WITH_ROCM)
     message(FATAL_ERROR "Error when compile CUDA and ROCM at the same time")
 endif()
+if (WITH_GPU AND WITH_MLU)
+    message(FATAL_ERROR "Error when compile GPU and MLU at the same time")
+endif()
 
 if(WITH_GPU AND NOT APPLE)
     enable_language(CUDA)
@@ -302,6 +306,10 @@ if(WITH_GPU)
     endif()
 endif()
 
+if(WITH_MLU)
+    include(neuware)
+endif()
+
 if(WITH_ROCM)
     include(hip)
     include(miopen) # set miopen libraries, must before configure

cmake/configure.cmake

@@ -102,6 +102,11 @@ if(WITH_IPU)
     add_definitions(-DPADDLE_WITH_IPU)
 endif()
 
+if(WITH_MLU)
+    message(STATUS "Compile with MLU!")
+    add_definitions(-DPADDLE_WITH_MLU)
+endif()
+
 if(WITH_GPU)
     add_definitions(-DPADDLE_WITH_CUDA)
     add_definitions(-DEIGEN_USE_GPU)

cmake/external/concurrentqueue.cmake

@@ -0,0 +1,42 @@
+# Copyright (c) 2021 PaddlePaddle Authors. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+include(ExternalProject)
+
+set(CONCURRENTQUEUE_PROJECT "extern_concurrentqueue")
+set(CONCURRENTQUEUE_VER "v1.0.3")
+SET(CONCURRENTQUEUE_URL_MD5 118e5bb661b567634647312991e10222)
+set(CONCURRENTQUEUE_PREFIX_URL "https://github.com/cameron314/concurrentqueue/archive/refs/tags")
+set(CONCURRENTQUEUE_URL "${CONCURRENTQUEUE_PREFIX_URL}/${CONCURRENTQUEUE_VER}.tar.gz")
+
+MESSAGE(STATUS "CONCURRENTQUEUE_VERSION: ${CONCURRENTQUEUE_VER}, CONCURRENTQUEUE_URL: ${CONCURRENTQUEUE_URL}")
+
+set(CONCURRENTQUEUE_PREFIX_DIR ${THIRD_PARTY_PATH}/concurrentqueue)
+set(CONCURRENTQUEUE_SOURCE_DIR ${THIRD_PARTY_PATH}/concurrentqueue/src/)
+set(CONCURRENTQUEUE_INCLUDE_DIR "${CONCURRENTQUEUE_SOURCE_DIR}/extern_concurrentqueue")
+
+ExternalProject_Add(
+    ${CONCURRENTQUEUE_PROJECT}
+    ${EXTERNAL_PROJECT_LOG_ARGS}
+    URL                   ${CONCURRENTQUEUE_URL}
+    URL_MD5               ${CONCURRENTQUEUE_URL_MD5}
+    PREFIX                ${CONCURRENTQUEUE_PREFIX_DIR}
+    DOWNLOAD_NO_PROGRESS  1
+    CONFIGURE_COMMAND     ""
+    BUILD_COMMAND         ""
+    INSTALL_COMMAND       ""
+    UPDATE_COMMAND        ""
+    )
+
+include_directories(${CONCURRENTQUEUE_INCLUDE_DIR})

cmake/neuware.cmake

@@ -0,0 +1,22 @@
+if(NOT WITH_MLU)
+    return()
+endif()
+
+if(NOT ENV{NEUWARE_HOME})
+    set(NEUWARE_HOME "/usr/local/neuware")
+else()
+    set(NEUWARE_HOME $ENV{NEUWARE_HOME})
+endif()
+message(STATUS "NEUWARE_HOME: " ${NEUWARE_HOME})
+
+set(NEUWARE_INCLUDE_DIR ${NEUWARE_HOME}/include)
+set(NEUWARE_LIB_DIR ${NEUWARE_HOME}/lib64)
+
+INCLUDE_DIRECTORIES(${NEUWARE_INCLUDE_DIR})
+
+set(CNNL_LIB ${NEUWARE_LIB_DIR}/libcnnl.so)
+set(CNRT_LIB ${NEUWARE_LIB_DIR}/libcnrt.so)
+set(CNDRV_LIB ${NEUWARE_LIB_DIR}/libcndrv.so)
+
+generate_dummy_static_lib(LIB_NAME "neuware_lib" GENERATOR "neuware.cmake")
+TARGET_LINK_LIBRARIES(neuware_lib ${CNNL_LIB} ${CNRT_LIB} ${CNDRV_LIB})

cmake/operators.cmake

@@ -12,6 +12,7 @@ function(op_library TARGET)
     set(hip_cc_srcs)
     set(xpu_cc_srcs)
     set(npu_cc_srcs)
+    set(mlu_cc_srcs)
     set(cudnn_cu_cc_srcs)
     set(miopen_cu_cc_srcs)
     set(cudnn_cu_srcs)
@@ -24,6 +25,10 @@ function(op_library TARGET)
     if (WITH_ASCEND_CL)
       set(op_common_deps ${op_common_deps} npu_op_runner)
     endif()
+    if (WITH_MLU)
+      set(op_common_deps ${op_common_deps} mlu_baseop)
+    endif()
+
     # Option `UNITY` is used to specify that operator `TARGET` will compiles with Unity Build.
     set(options UNITY)
     set(oneValueArgs "")
@@ -98,6 +103,12 @@ function(op_library TARGET)
                 list(APPEND npu_cc_srcs ${NPU_FILE}.cc)
             endif()
         endif()
+        if(WITH_MLU)
+            string(REPLACE "_op" "_op_mlu" MLU_FILE "${TARGET}")
+            if (EXISTS ${CMAKE_CURRENT_SOURCE_DIR}/${MLU_FILE}.cc)
+                list(APPEND mlu_cc_srcs ${MLU_FILE}.cc)
+            endif()
+        endif()
     else()
         foreach(src ${op_library_SRCS})
             if(WITH_ROCM AND ${src} MATCHES ".*_cudnn_op.cu$")
@@ -122,6 +133,8 @@ function(op_library TARGET)
                 list(APPEND xpu_cc_srcs ${src})
             elseif(WITH_ASCEND_CL AND ${src} MATCHES ".*_op_npu.cc$")
                 list(APPEND npu_cc_srcs ${src})
+            elseif(WITH_MLU AND ${src} MATCHES ".*_op_mlu.cc$")
+                list(APPEND mlu_cc_srcs ${src})
             elseif(${src} MATCHES ".*\\.cc$")
                 list(APPEND cc_srcs ${src})
             else()
@@ -196,7 +209,7 @@ 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} ${npu_cc_srcs})
+            compose_unity_target_sources(${UNITY_TARGET} cc ${cc_srcs} ${mkldnn_cc_srcs} ${xpu_cc_srcs} ${npu_cc_srcs} ${mlu_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})
@@ -207,7 +220,7 @@ function(op_library TARGET)
             # Add alias library to handle dependencies.
             add_library(${TARGET} ALIAS ${UNITY_TARGET})
         else()
-            cc_library(${TARGET} SRCS ${cc_srcs} ${mkldnn_cc_srcs} ${xpu_cc_srcs} ${npu_cc_srcs} DEPS ${op_library_DEPS}
+            cc_library(${TARGET} SRCS ${cc_srcs} ${mkldnn_cc_srcs} ${xpu_cc_srcs} ${npu_cc_srcs} ${mlu_cc_srcs} DEPS ${op_library_DEPS}
                 ${op_common_deps})
         endif()
     endif()
@@ -262,8 +275,10 @@ function(op_library TARGET)
     list(LENGTH xpu_cc_srcs xpu_cc_srcs_len)
     list(LENGTH miopen_cu_cc_srcs miopen_cu_cc_srcs_len)
     list(LENGTH npu_cc_srcs npu_cc_srcs_len)
+    list(LENGTH mlu_cc_srcs mlu_cc_srcs_len)
     if (${pybind_flag} EQUAL 0 AND ${mkldnn_cc_srcs_len} EQUAL 0 AND ${cu_srcs_len} EQUAL 0 AND ${cu_cc_srcs_len} EQUAL 0 AND
-        ${hip_srcs_len} EQUAL 0 AND ${hip_cc_srcs_len} EQUAL 0 AND ${miopen_cu_cc_srcs_len} EQUAL 0 AND ${xpu_cc_srcs_len} EQUAL 0 AND ${npu_cc_srcs_len} EQUAL 0)
+        ${hip_srcs_len} EQUAL 0 AND ${hip_cc_srcs_len} EQUAL 0 AND ${miopen_cu_cc_srcs_len} EQUAL 0 AND ${xpu_cc_srcs_len} EQUAL 0 AND
+        ${npu_cc_srcs_len} EQUAL 0 AND ${mlu_cc_srcs_len} EQUAL 0)
         file(APPEND ${pybind_file} "USE_CPU_ONLY_OP(${TARGET});\n")
         set(pybind_flag 1)
     endif()
@@ -322,6 +337,24 @@ function(op_library TARGET)
         endif()
         file(APPEND ${pybind_file} "USE_OP_DEVICE_KERNEL(${NPU_TARGET}, NPU);\n")
     endif()
+    if (WITH_MLU AND ${mlu_cc_srcs_len} GREATER 0)
+        file(READ ${ORIGINAL_TARGET}_mlu.cc TARGET_MLU_CONTENT)
+        # It is different from the logic above, becareful
+        string(REGEX MATCH "REGISTER_OP_MLU_KERNEL\\(.*" multi_mlu_register "${TARGET_MLU_CONTENT}")
+        # [ \t\r\n]* is used for blank characters
+        string(REGEX MATCH "REGISTER_OP_MLU_KERNEL\\([ \t\r\n]*[a-z0-9_]*," one_mlu_register "${multi_mlu_register}")
+
+        if (one_mlu_register STREQUAL "")
+            string(REPLACE "_op" "" MLU_TARGET "${TARGET}")
+        else ()
+            string(REPLACE "REGISTER_OP_MLU_KERNEL(" "" MLU_TARGET "${one_mlu_register}")
+            string(REPLACE "," "" MLU_TARGET "${MLU_TARGET}")
+            # [ \t\r\n]+ is used for blank characters.
+            # Here we use '+' instead of '*' since it is a REPLACE operation.
+            string(REGEX REPLACE "[ \t\r\n]+" "" MLU_TARGET "${MLU_TARGET}")
+        endif()
+        file(APPEND ${pybind_file} "USE_OP_DEVICE_KERNEL(${MLU_TARGET}, MLU);\n")
+    endif()
 
     # pybind USE_OP_DEVICE_KERNEL for MKLDNN
     if (WITH_MKLDNN AND ${mkldnn_cc_srcs_len} GREATER 0)
@@ -369,11 +402,11 @@ function(register_operators)
     set(multiValueArgs EXCLUDES DEPS)
     cmake_parse_arguments(register_operators "${options}" "${oneValueArgs}"
             "${multiValueArgs}" ${ARGN})
-
     file(GLOB OPS RELATIVE "${CMAKE_CURRENT_SOURCE_DIR}" "*_op.cc")
     string(REPLACE "_mkldnn" "" OPS "${OPS}")
     string(REPLACE "_xpu" "" OPS "${OPS}")
     string(REPLACE "_npu" "" 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/third_party.cmake

@@ -278,6 +278,11 @@ if(WITH_XPU)
     list(APPEND third_party_deps extern_xpu)
 endif(WITH_XPU)
 
+if(WITH_MLU)
+    include(external/concurrentqueue) # download, build, install concurrentqueue
+    list(APPEND third_party_deps extern_concurrentqueue)
+endif(WITH_MLU)
+
 if(WITH_PSLIB)
     include(external/pslib)          # download, build, install pslib
     list(APPEND third_party_deps extern_pslib)

paddle/fluid/distributed/fleet_executor/carrier.cc

@@ -240,13 +240,12 @@ void Carrier::CreateInterceptors() {
             task_node->run_at_offset(), task_node->run_per_steps()));
 
     std::unique_ptr<Interceptor> interceptor;
-    if (task_node->type().empty()) {
-      // TODO(wangxi): delete this in future
-      interceptor.reset(new Interceptor(interceptor_id, task_node));
-    } else {
-      interceptor = InterceptorFactory::Create(task_node->type(),
-                                               interceptor_id, task_node);
-    }
+    PADDLE_ENFORCE_NE(task_node->type().empty(), true,
+                      platform::errors::NotFound(
+                          "Cannot found type for task node with id %lld",
+                          task_node->task_id()));
+    interceptor = InterceptorFactory::Create(task_node->type(), interceptor_id,
+                                             task_node);
     interceptor->SetPlace(place_);
     interceptor->SetMiniBatchScope(minibatch_scope_);
     interceptor->SetMicroBatchScope(microbatch_scopes_);

paddle/fluid/distributed/fleet_executor/fleet_executor.cc

@@ -48,32 +48,29 @@ void FleetExecutor::Init(
     const framework::ProgramDesc& program_desc, framework::Scope* scope,
     const platform::Place& place, const std::vector<TaskNode*>& task_nodes,
     const std::unordered_map<int64_t, int64_t>& task_id_to_rank) {
-  if (task_nodes.size() == 0) {
-    LOG(INFO) << "fleet executor will use c++ side scheduler construction.";
-    runtime_graph_ = std::make_shared<RuntimeGraph>(program_desc, exe_desc_);
-  } else {
-    LOG(INFO) << "fleet executor has been set dependency on python side.";
-    // TODO(fleet_exe devs): the unused_vars should be got from run time graph
-    std::vector<std::unique_ptr<framework::OperatorBase>> ops;
-    for (auto task_node : task_nodes) {
-      for (auto op : task_node->ops()) {
-        ops.emplace_back(std::unique_ptr<framework::OperatorBase>(op));
-      }
-    }
-    auto unused_vars = framework::GetUnusedVars(program_desc.Block(0), ops, {});
-    runtime_graph_ = std::make_shared<RuntimeGraph>();
-    std::unordered_map<int64_t, TaskNode*> interceptor_id_to_task;
-    for (auto task_node : task_nodes) {
-      task_node->SetUnusedVars(unused_vars);
-      int64_t interceptor_id = task_node->task_id();
-      interceptor_id_to_task.emplace(interceptor_id, task_node);
-    }
-    runtime_graph_->SetInterceptorIdToRank(task_id_to_rank);
-    runtime_graph_->SetInterceptorIdToNode(interceptor_id_to_task);
-    for (auto& unique_op : ops) {
-      unique_op.release();
+  PADDLE_ENFORCE_GT(task_nodes.size(), 0,
+                    platform::errors::InvalidArgument(
+                        "Fleet executor is inited with empty task node"));
+  // TODO(fleet_exe devs): the unused_vars should be got from run time graph
+  std::vector<std::unique_ptr<framework::OperatorBase>> ops;
+  for (auto task_node : task_nodes) {
+    for (auto op : task_node->ops()) {
+      ops.emplace_back(std::unique_ptr<framework::OperatorBase>(op));
     }
   }
+  auto unused_vars = framework::GetUnusedVars(program_desc.Block(0), ops, {});
+  runtime_graph_ = std::make_shared<RuntimeGraph>();
+  std::unordered_map<int64_t, TaskNode*> interceptor_id_to_task;
+  for (auto task_node : task_nodes) {
+    task_node->SetUnusedVars(unused_vars);
+    int64_t interceptor_id = task_node->task_id();
+    interceptor_id_to_task.emplace(interceptor_id, task_node);
+  }
+  runtime_graph_->SetInterceptorIdToRank(task_id_to_rank);
+  runtime_graph_->SetInterceptorIdToNode(interceptor_id_to_task);
+  for (auto& unique_op : ops) {
+    unique_op.release();
+  }
   root_scope_ = scope;
   place_ = place;
   PADDLE_ENFORCE_NOT_NULL(root_scope_, platform::errors::InvalidArgument(

paddle/fluid/distributed/fleet_executor/fleet_executor_desc.proto

@@ -23,9 +23,5 @@ message RankInfo {
 message FleetExecutorDesc {
   optional int64 cur_rank = 1 [ default = 0 ]; // Rank id of current processor
   repeated RankInfo cluster_info = 2;
-  optional int32 dp_degree = 3 [ default = 1 ];
-  optional int32 mp_degree = 4 [ default = 1 ];
-  optional int32 pp_degree = 5 [ default = 1 ];
-  optional int64 num_micro_batches = 6 [ default = 1 ];
-  optional int64 num_slots = 7 [ default = 1 ];
+  optional int64 num_micro_batches = 3 [ default = 1 ];
 }