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

cmake/inference_lib.cmake

@@ -181,6 +181,13 @@ IF(WITH_XPU)
         DSTS ${dst_dir} ${dst_dir})
 ENDIF()
 
+IF(WITH_IPU)
+    set(dst_dir "${PADDLE_INFERENCE_INSTALL_DIR}/third_party/install/ipu")
+    copy(inference_lib_dist
+        SRCS ${CMAKE_BINARY_DIR}/paddle/fluid/platform/device/ipu/libpaddle_ipu.so
+        DSTS ${dst_dir})
+ENDIF()
+
 # CMakeCache Info
 copy(inference_lib_dist
         SRCS ${CMAKE_CURRENT_BINARY_DIR}/CMakeCache.txt
@@ -189,6 +196,7 @@ copy(inference_lib_dist
 copy_part_of_thrid_party(inference_lib_dist ${PADDLE_INFERENCE_INSTALL_DIR})
 
 set(src_dir "${PADDLE_SOURCE_DIR}/paddle/fluid")
+
 if(WIN32)
     if(WITH_STATIC_LIB)
         set(paddle_inference_lib $<TARGET_FILE_DIR:paddle_inference>/libpaddle_inference.lib
@@ -304,7 +312,7 @@ copy(fluid_lib_dist
         )
 
 set(module "platform")
-set(platform_lib_deps profiler_proto error_codes_proto)
+set(platform_lib_deps profiler_proto errors)
 if(WITH_GPU)
   set(platform_lib_deps ${platform_lib_deps} external_error_proto)
 endif(WITH_GPU)
@@ -317,7 +325,7 @@ copy(fluid_lib_dist
 
 set(module "string")
 copy(fluid_lib_dist
-        SRCS ${src_dir}/${module}/*.h ${src_dir}/${module}/tinyformat/*.h
+        SRCS ${PADDLE_SOURCE_DIR}/paddle/utils/${module}/*.h ${PADDLE_SOURCE_DIR}/paddle/utils/${module}/tinyformat/*.h 
         DSTS ${dst_dir}/${module} ${dst_dir}/${module}/tinyformat
         )
 

cmake/pten.cmake

@@ -243,3 +243,29 @@ function(register_kernels)
         endif()
     endforeach()
 endfunction()
+
+function(append_op_util_declare TARGET)
+    file(READ ${CMAKE_CURRENT_SOURCE_DIR}/${TARGET} target_content)
+    string(REGEX MATCH "(PT_REGISTER_API_NAME|PT_REGISTER_ARG_MAPPING_FN)\\([ \t\r\n]*[a-z0-9_]*" util_registrar "${target_content}")
+    string(REPLACE "PT_REGISTER_ARG_MAPPING_FN" "PT_DECLARE_ARG_MAPPING_FN" util_declare "${util_registrar}")
+    string(REPLACE "PT_REGISTER_API_NAME" "PT_REGISTER_API_NAME" util_declare "${util_declare}")
+    string(APPEND util_declare ");")
+    file(APPEND ${op_utils_header} "${util_declare}")
+endfunction()
+
+function(register_op_utils TARGET_NAME)
+    set(utils_srcs)
+    set(options "")
+    set(oneValueArgs "")
+    set(multiValueArgs EXCLUDES DEPS)
+    cmake_parse_arguments(register_op_utils "${options}" "${oneValueArgs}"
+        "${multiValueArgs}" ${ARGN})
+
+    file(GLOB SIGNATURES RELATIVE "${CMAKE_CURRENT_SOURCE_DIR}" "*_sig.cc")
+    foreach(target ${SIGNATURES})
+        append_op_util_declare(${target})
+        list(APPEND utils_srcs ${CMAKE_CURRENT_SOURCE_DIR}/${target})
+    endforeach()
+
+    cc_library(${TARGET_NAME} SRCS ${utils_srcs} DEPS ${register_op_utils_DEPS})
+endfunction()

paddle/CMakeLists.txt

+add_subdirectory(utils)
 add_subdirectory(scripts)
 add_subdirectory(testing)
 set(PYTHON_TESTS_DIR ${PADDLE_BINARY_DIR}/python/paddle/fluid/tests CACHE INTERNAL "python tests directory")

paddle/fluid/CMakeLists.txt

@@ -4,7 +4,6 @@ add_subdirectory(distributed)
 add_subdirectory(framework)
 add_subdirectory(imperative)
 add_subdirectory(operators)
-add_subdirectory(string)
 add_subdirectory(pybind)
 add_subdirectory(eager)
 # NOTE: please add subdirectory inference at last.

paddle/fluid/distributed/CMakeLists.txt

@@ -13,17 +13,7 @@ if (CMAKE_CXX_COMPILER_VERSION VERSION_GREATER 7.0)
 endif()
 
 add_subdirectory(common)
-add_subdirectory(service)
-add_subdirectory(table)
+add_subdirectory(ps)
 add_subdirectory(test)
 add_subdirectory(index_dataset)
 add_subdirectory(fleet_executor)
-
-get_property(RPC_DEPS GLOBAL PROPERTY RPC_DEPS)
-
-set_source_files_properties(fleet.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
-cc_library(fleet
-        SRCS fleet.cc
-        DEPS framework_proto ps_framework_proto ps_service variable_helper scope op_registry fs shell ${RPC_DEPS})
-
-target_link_libraries(fleet z)

paddle/fluid/distributed/fleet_executor/dist_model.cc

@@ -13,11 +13,13 @@
 // limitations under the License.
 
 #include <glog/logging.h>
+#include <chrono>  // NOLINT
 
 #include "paddle/fluid/distributed/fleet_executor/dist_model.h"
 #include "paddle/fluid/distributed/fleet_executor/fleet_executor.h"
 #include "paddle/fluid/distributed/fleet_executor/task_node.h"
 #include "paddle/fluid/framework/block_desc.h"
+#include "paddle/fluid/framework/feed_fetch_method.h"
 #include "paddle/fluid/framework/naive_executor.h"
 #include "paddle/fluid/framework/op_proto_maker.h"
 #include "paddle/fluid/framework/program_desc.h"
@@ -37,10 +39,110 @@ bool IsPersistable(const framework::VarDesc *var) {
   }
   return false;
 }
+
+bool LoadDataFromDistModelTensor(const DistModelTensor &input_data,
+                                 framework::LoDTensor *input_tensor,
+                                 const platform::Place &place) {
+  VLOG(3) << "Loading data from DistModelTensor for " << input_data.name;
+  framework::DDim dims = framework::make_ddim(input_data.shape);
+  void *input_tensor_ptr;
+  if (input_data.dtype == DistModelDataType::INT64) {
+    input_tensor_ptr = input_tensor->mutable_data<int64_t>(dims, place);
+  } else if (input_data.dtype == DistModelDataType::FLOAT32) {
+    input_tensor_ptr = input_tensor->mutable_data<float>(dims, place);
+  } else if (input_data.dtype == DistModelDataType::INT32) {
+    input_tensor_ptr = input_tensor->mutable_data<int32_t>(dims, place);
+  } else {
+    // Q(fleet exe dev): for input/output, should we support fp16
+    LOG(ERROR) << "unsupported feed type " << input_data.dtype;
+    return false;
+  }
+
+  PADDLE_ENFORCE_NOT_NULL(
+      input_tensor_ptr,
+      paddle::platform::errors::Fatal(
+          "LoDTensor creation failed. DistModel loaded data failed."));
+  PADDLE_ENFORCE_NOT_NULL(input_data.data.data(),
+                          paddle::platform::errors::InvalidArgument(
+                              "DistModelTensor contains no data."));
+
+  if (platform::is_cpu_place(place)) {
+    VLOG(3) << "Loading data for CPU.";
+    std::memcpy(static_cast<void *>(input_tensor_ptr), input_data.data.data(),
+                input_data.data.length());
+  } else if (platform::is_gpu_place(place)) {
+    VLOG(3) << "Loading data for GPU.";
+#if defined(PADDLE_WITH_CUDA) || defined(PADDLE_WITH_HIP)
+    platform::DeviceContextPool &pool = platform::DeviceContextPool::Instance();
+    auto *dev_ctx =
+        dynamic_cast<const platform::CUDADeviceContext *>(pool.Get(place));
+    auto gpu_place = place;
+    memory::Copy(gpu_place, static_cast<void *>(input_tensor_ptr),
+                 platform::CPUPlace(), input_data.data.data(),
+                 input_data.data.length(), dev_ctx->stream());
+#else
+    PADDLE_THROW(paddle::platform::errors::Fatal(
+        "Paddle wasn't compiled with CUDA, but place is GPU."));
+#endif
+  } else {
+    PADDLE_THROW(paddle::platform::errors::InvalidArgument(
+        "DistModel only supports CPU and GPU."));
+  }
+
+  framework::LoD dst_lod;
+  for (auto &src_lod : input_data.lod) {
+    dst_lod.emplace_back(src_lod);
+  }
+  input_tensor->set_lod(dst_lod);
+  return true;
+}
+
+std::string DistModelDTypeToString(DistModelDataType dtype) {
+  switch (dtype) {
+    case DistModelDataType::FLOAT32:
+      return "float32";
+    case DistModelDataType::FLOAT16:
+      return "float16";
+    case DistModelDataType::INT64:
+      return "int64";
+    case DistModelDataType::INT32:
+      return "int32";
+    case DistModelDataType::INT8:
+      return "int8";
+  }
+  return "NOT SUPPORT DTYPE";
+}
+
+bool IsPPFirstStage(const DistModelConfig &config) {
+  return config.local_rank - config.mp_degree < 0;
+}
+
+bool IsPPLastStage(const DistModelConfig &config) {
+  return config.local_rank + config.mp_degree >= config.nranks;
+}
+
+class DistModelTimer {
+ public:
+  void tic() { tic_time = std::chrono::high_resolution_clock::now(); }
+  double toc() {
+    std::chrono::high_resolution_clock::time_point toc_time =
+        std::chrono::high_resolution_clock::now();
+    std::chrono::duration<double> time_elapse =
+        std::chrono::duration_cast<std::chrono::duration<double>>(toc_time -
+                                                                  tic_time);
+    double time_elapse_in_ms =
+        static_cast<double>(time_elapse.count()) * 1000.0;
+    return time_elapse_in_ms;
+  }
+
+ private:
+  std::chrono::high_resolution_clock::time_point tic_time;
+};
+
 }  // namespace
 
 bool DistModel::Init() {
-  /* TODO(fleet exe dev): implement this funct */
+  carrier_id_ = "inference";
   bool init_method = (!config_.model_dir.empty() || config_.program_desc);
   PADDLE_ENFORCE_EQ(init_method, true,
                     platform::errors::InvalidArgument(
@@ -127,10 +229,9 @@ bool DistModel::CommInit() {
     InsertCommOp("mp_comm_id", mp_group_nranks, mp_group_rank, peer_endpoints,
                  comm_init_block, config_.mp_ring_id);
   }
-  if (config_.pp_degree) {
-    // NOTE: the last pp stage doesn't need init pp comm
+  if (config_.pp_degree > 1) {
     VLOG(3) << "Init comm group for pp.";
-    if (config_.local_rank - config_.mp_degree >= 0) {
+    if (!IsPPFirstStage(config_)) {
       PADDLE_ENFORCE_EQ(config_.pp_upstream_ring_id >= 0, true,
                         platform::errors::InvalidArgument(
                             "pp upstream ring id must be provided for "
@@ -143,7 +244,7 @@ bool DistModel::CommInit() {
                    comm_init_block, config_.pp_upstream_ring_id);
     }
 
-    if (config_.local_rank + config_.mp_degree < config_.nranks) {
+    if (!IsPPLastStage(config_)) {
       PADDLE_ENFORCE_EQ(config_.pp_downstream_ring_id >= 0, true,
                         platform::errors::InvalidArgument(
                             "pp downstream ring id must be provided for "
@@ -326,7 +427,7 @@ bool DistModel::PrepareFleetExe() {
     id_to_rank.insert({i, i});
   }
   fleet_exe.reset(new FleetExecutor(executor_desc_));
-  fleet_exe->Init("inference", *(program_.get()), scope_.get(), place_, 1,
+  fleet_exe->Init(carrier_id_, *(program_.get()), scope_.get(), place_, 1,
                   {task_node_.get()}, id_to_rank);
   return true;
 }
@@ -340,8 +441,27 @@ bool DistModel::PrepareFeedAndFetch() {
         feeds_.resize(idx + 1);
       }
       feeds_[idx] = op;
-      feed_names_[op->Output("Out")[0]] = idx;
-      idx_to_feeds_[idx] = op->Output("Out")[0];
+      std::string var_name = op->Output("Out")[0];
+      feed_names_[var_name] = idx;
+      idx_to_feeds_[idx] = var_name;
+      framework::VarDesc *real_var = program_->Block(0).FindVar(var_name);
+      if (!real_var) {
+        LOG(ERROR)
+            << "The output of feed ops [" << var_name
+            << "] cannot be found in the program. Check the inference program.";
+        return false;
+      }
+      if (real_var->GetDataType() == framework::proto::VarType::FP32) {
+        feeds_to_dtype_.insert({var_name, DistModelDataType::FLOAT32});
+      } else if (real_var->GetDataType() == framework::proto::VarType::INT32) {
+        feeds_to_dtype_.insert({var_name, DistModelDataType::INT32});
+      } else if (real_var->GetDataType() == framework::proto::VarType::INT64) {
+        feeds_to_dtype_.insert({var_name, DistModelDataType::INT64});
+      } else {
+        LOG(ERROR) << "Don't support feed var dtype for: "
+                   << real_var->GetDataType();
+        return false;
+      }
     } else if (op->Type() == "fetch") {
       VLOG(3) << "fetch op with fetch var: " << op->Input("X")[0];
       int idx = BOOST_GET_CONST(int, op->GetAttr("col"));
@@ -349,15 +469,170 @@ bool DistModel::PrepareFeedAndFetch() {
         fetches_.resize(idx + 1);
       }
       fetches_[idx] = op;
-      id_to_fetches_[idx] = op->Input("X")[0];
+      idx_to_fetches_[idx] = op->Input("X")[0];
     }
   }
+
+  if (config_.pp_degree == 1) {
+    if (feeds_.size() == 0) {
+      LOG(ERROR) << "No feed ops in the inf program, please check the program.";
+      return false;
+    }
+    if (fetches_.size() == 0) {
+      LOG(ERROR) << "No fetch op in the inf program, please check the program.";
+      return false;
+    }
+  } else {
+    if (IsPPFirstStage(config_)) {
+      if (feeds_.size() == 0) {
+        LOG(ERROR) << "Feed ops are needed for the first pp stage.";
+        return false;
+      }
+    } else {
+      if (feeds_.size() > 0) {
+        LOG(WARNING) << "Feed op is found in the non-first stage of pp.";
+      } else {
+        LOG(INFO) << "No feed ops in non-first pp stage.";
+      }
+    }
+    if (IsPPLastStage(config_)) {
+      if (fetches_.size() == 0) {
+        LOG(WARNING) << "No fetch op was found in the last pp stage. Make sure "
+                        "the result has been sent to frist pp stage.";
+      }
+    } else {
+      if (fetches_.size() > 0) {
+        LOG(WARNING) << "Fetch op is found in the non-last stage of pp.";
+      } else {
+        LOG(INFO) << "No fetch op in non-last pp stage.";
+      }
+    }
+  }
+  return true;
+}
+
+bool DistModel::FeedData(const std::vector<DistModelTensor> &input_data,
+                         framework::Scope *scope) {
+  VLOG(3) << "DistModel is feeding data.";
+  if (input_data.size() != feeds_.size()) {
+    LOG(ERROR) << "Should provide " << feeds_.size() << " feeds, but got "
+               << input_data.size() << " data.";
+    return false;
+  }
+  feed_tensors_.resize(feeds_.size());
+  for (size_t i = 0; i < input_data.size(); ++i) {
+    // feed each data separately
+    framework::LoDTensor *input_tensor = &(feed_tensors_[i]);
+    if (!LoadDataFromDistModelTensor(input_data[i], input_tensor, place_)) {
+      LOG(ERROR) << "Fail to load data from tensor " << input_data[i].name;
+      return false;
+    }
+    std::string target_name = input_data[i].name;
+    if (feed_names_.find(target_name) == feed_names_.end()) {
+      LOG(ERROR) << "The input name [" << target_name
+                 << "] cannot be found in the program."
+                 << " DistModel loads data failed.";
+      return false;
+    }
+    if (input_data[i].dtype != feeds_to_dtype_[target_name]) {
+      LOG(ERROR) << "Feed var [" << target_name << "] expected dtype is: "
+                 << DistModelDTypeToString(feeds_to_dtype_[target_name])
+                 << ". But received dtype is: "
+                 << DistModelDTypeToString(input_data[i].dtype) << ".";
+      return false;
+    }
+    int feed_idx = feed_names_[target_name];
+    framework::SetFeedVariable(scope, *input_tensor, "feed", feed_idx);
+  }
   return true;
 }
 
-void DistModel::Run(const std::vector<DistModelTensor> &input_data,
+bool DistModel::FetchResults(std::vector<DistModelTensor> *output_data,
+                             framework::Scope *scope) {
+  VLOG(3) << "DistModel is fetch results.";
+  output_data->resize(fetches_.size());
+  for (size_t i = 0; i < fetches_.size(); ++i) {
+    int idx = BOOST_GET_CONST(int, fetches_[i]->GetAttr("col"));
+    VLOG(3) << "Fetching data for [" << idx_to_fetches_[idx] << "]";
+    PADDLE_ENFORCE_EQ(
+        static_cast<size_t>(idx), i,
+        platform::errors::InvalidArgument(
+            "Fetch op's col attr(%d) should be equal to the index(%d)", idx,
+            i));
+    framework::FetchType &fetch_var =
+        framework::GetFetchVariable(*scope, "fetch", idx);
+    auto &fetch = BOOST_GET(framework::LoDTensor, fetch_var);
+    auto type = fetch.type();
+    auto output = &(output_data->at(i));
+    output->name = idx_to_fetches_[idx];
+    bool rst = false;
+    if (type == framework::proto::VarType::FP32) {
+      rst = FetchResult<float>(fetch, output);
+      output->dtype = DistModelDataType::FLOAT32;
+    } else if (type == framework::proto::VarType::INT64) {
+      rst = FetchResult<int64_t>(fetch, output);
+      output->dtype = DistModelDataType::INT64;
+    } else if (type == framework::proto::VarType::INT32) {
+      rst = FetchResult<int32_t>(fetch, output);
+      output->dtype = DistModelDataType::INT32;
+    } else {
+      LOG(ERROR) << "DistModel meets unknown fetch data type. DistModel only "
+                    "supports float32, int64 and int32 fetch type for now.";
+    }
+    if (!rst) {
+      LOG(ERROR) << "DistModel fails to fetch result " << idx_to_fetches_[idx];
+      return false;
+    }
+  }
+  return true;
+}
+
+template <typename T>
+bool DistModel::FetchResult(const framework::LoDTensor &fetch,
+                            DistModelTensor *output_data) {
+  auto shape = framework::vectorize(fetch.dims());
+  output_data->shape.assign(shape.begin(), shape.end());
+  const T *data = fetch.data<T>();
+  int64_t num_elems = fetch.numel();
+  output_data->data.Resize(num_elems * sizeof(T));
+  // The output of fetch op is always on the cpu, no need switch on place
+  memcpy(output_data->data.data(), data, num_elems * sizeof(T));
+  output_data->lod.clear();
+  for (auto &level : fetch.lod()) {
+    output_data->lod.emplace_back(level.begin(), level.end());
+  }
+  return true;
+}
+
+bool DistModel::Run(const std::vector<DistModelTensor> &input_data,
                     std::vector<DistModelTensor> *output_data) {
-  /* TODO(fleet exe dev): implement this funct */
+  // TODO(fleet exe dev): support pipeline inf mode
+  VLOG(3) << "DistModel run for once.";
+
+  DistModelTimer timer;
+  timer.tic();
+
+  if (!FeedData(input_data, scope_.get())) {
+    LOG(ERROR) << "DistModel failed at feeding data.";
+    return false;
+  }
+  double feed_elapse = timer.toc();
+  VLOG(3) << "Finish loading data, cost " << feed_elapse << "ms.";
+
+  fleet_exe->Run(carrier_id_);
+  double fleet_exe_elapse = timer.toc();
+  VLOG(3) << "Finish FleetExe running, cost " << fleet_exe_elapse - feed_elapse
+          << "ms.";
+
+  if (!FetchResults(output_data, scope_.get())) {
+    LOG(ERROR) << "DistModel failed at fetching result.";
+    return false;
+  }
+  double fetch_elapse = timer.toc();
+  VLOG(3) << "Finish fetching data, cost " << fetch_elapse - fleet_exe_elapse
+          << "ms.";
+  VLOG(3) << "DistModel finish inf, cost " << fetch_elapse << "ms";
+  return true;
 }
 
 }  // namespace distributed

paddle/fluid/distributed/fleet_executor/dist_model.h

@@ -19,6 +19,7 @@
 
 #include "paddle/fluid/distributed/fleet_executor/dist_model_tensor_wrapper.h"
 #include "paddle/fluid/distributed/fleet_executor/fleet_executor_desc.pb.h"
+#include "paddle/fluid/framework/lod_tensor.h"
 #include "paddle/fluid/framework/tensor.h"
 #include "paddle/fluid/platform/macros.h"
 #include "paddle/fluid/platform/place.h"
@@ -57,7 +58,7 @@ class DistModel {
  public:
   explicit DistModel(const DistModelConfig& config) : config_(config) {}
   bool Init();
-  void Run(const std::vector<DistModelTensor>& input_data,
+  bool Run(const std::vector<DistModelTensor>& input_data,
            std::vector<DistModelTensor>* output_data);
   ~DistModel() = default;
 
@@ -75,12 +76,22 @@ class DistModel {
   void InsertCommOp(std::string tmp_var_name, int nranks, int rank,
                     const std::vector<std::string>& peer_endpoints,
                     framework::BlockDesc* block, int ring_id);
+  bool FeedData(const std::vector<DistModelTensor>& input_data,
+                framework::Scope* scope);
+  bool FetchResults(std::vector<DistModelTensor>* output_data,
+                    framework::Scope* scope);
+  template <typename T>
+  bool FetchResult(const framework::LoDTensor& fetch,
+                   DistModelTensor* output_data);
 
+  std::string carrier_id_;
+  std::vector<framework::LoDTensor> feed_tensors_;
   std::vector<framework::OpDesc*> feeds_;
   std::map<std::string, int64_t> feed_names_;
   std::map<int64_t, std::string> idx_to_feeds_;
+  std::map<std::string, DistModelDataType> feeds_to_dtype_;
   std::vector<framework::OpDesc*> fetches_;
-  std::map<int64_t, std::string> id_to_fetches_;
+  std::map<int64_t, std::string> idx_to_fetches_;
   DistModelConfig config_;
   FleetExecutorDesc executor_desc_;
   std::shared_ptr<FleetExecutor> fleet_exe;

paddle/fluid/distributed/fleet_executor/dist_model_tensor_wrapper.h

@@ -62,7 +62,7 @@ class DistModelDataBuf {
   void Free();
   void* data_{nullptr};
   size_t length_{0};
-  bool memory_owned_{false};
+  bool memory_owned_{true};
 };
 
 struct DistModelTensor {

paddle/fluid/distributed/ps/CMakeLists.txt

+set_property(GLOBAL PROPERTY RPC_DEPS sendrecv_rpc ${BRPC_DEPS} string_helper)
+add_subdirectory(table)
+add_subdirectory(service)
+add_subdirectory(wrapper)

paddle/fluid/distributed/service/CMakeLists.txt → paddle/fluid/distributed/ps/service/CMakeLists.txt

@@ -8,12 +8,12 @@ brpc_library(sendrecv_rpc SRCS
         PROTO sendrecv.proto
         DEPS ${BRPC_DEPS} )
 
-set_property(GLOBAL PROPERTY RPC_DEPS sendrecv_rpc ${BRPC_DEPS} string_helper)
+#set_property(GLOBAL PROPERTY RPC_DEPS sendrecv_rpc ${BRPC_DEPS} string_helper)
 
 get_property(RPC_DEPS GLOBAL PROPERTY RPC_DEPS)
 
-set_source_files_properties(communicator.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
-set_source_files_properties(service.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
+set_source_files_properties(communicator/communicator.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
+set_source_files_properties(ps_service/service.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
 set_source_files_properties(brpc_ps_server.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
 set_source_files_properties(brpc_ps_client.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
 set_source_files_properties(ps_local_client.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
@@ -36,11 +36,13 @@ ps_local_client.cc DEPS boost eigen3 table brpc_utils simple_threadpool ${RPC_DE
 cc_library(client SRCS ps_client.cc DEPS downpour_client boost ${RPC_DEPS})
 cc_library(server SRCS server.cc DEPS downpour_server boost ${RPC_DEPS})
 
-cc_library(communicator SRCS communicator.cc DEPS scope client boost table math_function selected_rows_functor ${RPC_DEPS})
-cc_library(ps_service SRCS service.cc DEPS communicator client server boost ${RPC_DEPS})
+cc_library(communicator SRCS communicator/communicator.cc DEPS scope client boost table math_function selected_rows_functor ${RPC_DEPS})
+cc_library(ps_service SRCS ps_service/service.cc DEPS communicator client server boost ${RPC_DEPS})
 
 cc_library(heter_server SRCS heter_server.cc DEPS brpc_utils ${COMMON_DEPS} ${RPC_DEPS})
 cc_library(heter_client SRCS heter_client.cc DEPS brpc_utils ${COMMON_DEPS} ${RPC_DEPS})
 
-set_source_files_properties(graph_py_service.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
-cc_library(graph_py_service SRCS graph_py_service.cc DEPS ps_service)
+set_source_files_properties(ps_service/graph_py_service.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
+cc_library(graph_py_service SRCS ps_service/graph_py_service.cc DEPS ps_service)
+
+#add_subdirectory(communicator)

paddle/fluid/distributed/service/brpc_ps_client.cc → paddle/fluid/distributed/ps/service/brpc_ps_client.cc

@@ -16,7 +16,7 @@
 #include <sstream>
 #include <string>
 
-#include "paddle/fluid/distributed/service/brpc_ps_client.h"
+#include "paddle/fluid/distributed/ps/service/brpc_ps_client.h"
 #include "paddle/fluid/framework/archive.h"
 
 static const int max_port = 65535;

paddle/fluid/distributed/service/brpc_ps_client.h → paddle/fluid/distributed/ps/service/brpc_ps_client.h

@@ -22,8 +22,8 @@
 #include "brpc/channel.h"
 #include "brpc/controller.h"
 #include "brpc/server.h"
-#include "paddle/fluid/distributed/service/brpc_utils.h"
-#include "paddle/fluid/distributed/service/ps_client.h"
+#include "paddle/fluid/distributed/ps/service/brpc_utils.h"
+#include "paddle/fluid/distributed/ps/service/ps_client.h"
 #include "paddle/fluid/framework/channel.h"
 #include "paddle/fluid/framework/lod_tensor.h"
 #include "paddle/fluid/framework/scope.h"

paddle/fluid/distributed/service/brpc_ps_server.cc → paddle/fluid/distributed/ps/service/brpc_ps_server.cc

@@ -12,12 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/service/brpc_ps_server.h"
+#include "paddle/fluid/distributed/ps/service/brpc_ps_server.h"
 #include <thread>  // NOLINT
 #include "butil/object_pool.h"
 #include "paddle/fluid/distributed/common/cost_timer.h"
-#include "paddle/fluid/distributed/table/depends/sparse_utils.h"
-#include "paddle/fluid/distributed/table/table.h"
+#include "paddle/fluid/distributed/ps/table/depends/sparse_utils.h"
+#include "paddle/fluid/distributed/ps/table/table.h"
 #include "paddle/fluid/framework/archive.h"
 #include "paddle/fluid/platform/profiler.h"
 

paddle/fluid/distributed/service/brpc_ps_server.h → paddle/fluid/distributed/ps/service/brpc_ps_server.h

@@ -17,8 +17,8 @@
 #include "brpc/channel.h"
 #include "brpc/controller.h"
 #include "brpc/server.h"
-#include "paddle/fluid/distributed/service/brpc_utils.h"
-#include "paddle/fluid/distributed/service/server.h"
+#include "paddle/fluid/distributed/ps/service/brpc_utils.h"
+#include "paddle/fluid/distributed/ps/service/server.h"
 
 namespace brpc {
 class Controller;

paddle/fluid/distributed/service/brpc_utils.cc → paddle/fluid/distributed/ps/service/brpc_utils.cc

@@ -12,7 +12,7 @@ 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 "paddle/fluid/distributed/service/brpc_utils.h"
+#include "paddle/fluid/distributed/ps/service/brpc_utils.h"
 #include <arpa/inet.h>
 #include <netdb.h>
 #include "paddle/fluid/platform/enforce.h"
@@ -76,7 +76,7 @@ void SerializeToMultiVarMsgAndIOBuf(
 
     if (var->IsType<framework::LoDTensor>()) {
       SerializeLodTensor(var, ctx, send_var_msg, &temp_iobuf);
-    } else if (var->IsType<framework::SelectedRows>()) {
+    } else if (var->IsType<pten::SelectedRows>()) {
       SerializeSelectedRows(var, ctx, send_var_msg, &temp_iobuf);
     }
     iobuf->append(temp_iobuf);
@@ -127,7 +127,7 @@ void SerializeLodTensor(framework::Variable* var,
 void SerializeSelectedRows(framework::Variable* var,
                            const platform::DeviceContext& ctx, VarMsg* var_msg,
                            butil::IOBuf* iobuf) {
-  framework::SelectedRows* slr = var->GetMutable<framework::SelectedRows>();
+  pten::SelectedRows* slr = var->GetMutable<pten::SelectedRows>();
   auto* tensor = slr->mutable_value();
   auto* rows = slr->mutable_rows();
 
@@ -255,7 +255,7 @@ void DeserializeSelectedRows(
     butil::IOBufBytesIterator& io_buffer_itr,  // NOLINT
     const platform::DeviceContext& ctx) {
   const auto place = ctx.GetPlace();
-  auto* slr = var->GetMutable<framework::SelectedRows>();
+  auto* slr = var->GetMutable<pten::SelectedRows>();
   framework::Tensor* tensor = slr->mutable_value();
   slr->set_height(msg.slr_height());
   std::vector<int64_t> tmp_rows(msg.dims()[0]);

paddle/fluid/distributed/service/brpc_utils.h → paddle/fluid/distributed/ps/service/brpc_utils.h

@@ -20,7 +20,7 @@ limitations under the License. */
 #include <vector>
 
 #include "brpc/channel.h"
-#include "paddle/fluid/distributed/service/sendrecv.pb.h"
+#include "paddle/fluid/distributed/ps/service/sendrecv.pb.h"
 #include "paddle/fluid/framework/data_type.h"
 #include "paddle/fluid/framework/lod_tensor.h"
 #include "paddle/fluid/framework/scope.h"

paddle/fluid/distributed/ps/service/communicator/CMakeLists.txt

+
+
+get_property(RPC_DEPS GLOBAL PROPERTY RPC_DEPS)
+
+set_source_files_properties(communicator.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
+
+
+cc_library(communicator SRCS communicator.cc DEPS scope client boost table math_function selected_rows_functor ${RPC_DEPS})

paddle/fluid/distributed/service/communicator.cc → paddle/fluid/distributed/ps/service/communicator/communicator.cc

@@ -12,12 +12,12 @@ 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 "paddle/fluid/distributed/service/communicator.h"
+#include "paddle/fluid/distributed/ps/service/communicator/communicator.h"
 
 #include <google/protobuf/text_format.h>
 
 #include "gflags/gflags.h"
-#include "paddle/fluid/distributed/service/brpc_ps_client.h"
+#include "paddle/fluid/distributed/ps/service/brpc_ps_client.h"
 #include "paddle/fluid/platform/profiler.h"
 #include "paddle/fluid/string/string_helper.h"
 
@@ -28,7 +28,7 @@ namespace paddle {
 namespace distributed {
 
 using framework::LoDTensor;
-using framework::SelectedRows;
+using pten::SelectedRows;
 
 const uint32_t MAX_FEASIGN_NUM = 1024 * 100 * 100;
 
@@ -293,7 +293,7 @@ void Communicator::RpcSendSparse(const std::string &var_name, int table_id,
   std::vector<float *> push_g_vec;
 
   auto *send_var = scope.FindVar(var_name);
-  auto *tensor = send_var->GetMutable<SelectedRows>();
+  auto *tensor = send_var->GetMutable<pten::SelectedRows>();
   auto dim = tensor->value().dims()[1];
   std::transform(tensor->rows().begin(), tensor->rows().end(),
                  std::back_inserter(sparse_push_keys),
@@ -1012,10 +1012,10 @@ void GeoCommunicator::Send(const std::vector<std::string> &var_names,
 
   auto *var = scope.FindVar(table_name);
 
-  PADDLE_ENFORCE_EQ(var->IsType<framework::SelectedRows>(), true,
+  PADDLE_ENFORCE_EQ(var->IsType<pten::SelectedRows>(), true,
                     platform::errors::InvalidArgument(
                         "Only need to send Sparse Grad in Geo mode."));
-  auto &rows = var->Get<framework::SelectedRows>().rows();
+  auto &rows = var->Get<pten::SelectedRows>().rows();
 
   // insert ids which has not been record
   for (size_t j = 0; j < rows.size(); j++) {
@@ -1290,7 +1290,7 @@ void GeoCommunicator::SendSparse(const std::string &varname,
   auto cpu_ctx = paddle::platform::CPUDeviceContext();
 
   auto *var_delta = delta_scope_->Var(varname);
-  auto *t_delta = var_delta->GetMutable<framework::SelectedRows>();
+  auto *t_delta = var_delta->GetMutable<pten::SelectedRows>();
   auto *var_t_value = t_delta->mutable_value();
   var_t_value->Resize({static_cast<int64_t>(sparse_ids.size()), dims1});
   auto *t_value = var_t_value->mutable_data<float>(cpu_ctx.GetPlace());

paddle/fluid/distributed/service/communicator.h → paddle/fluid/distributed/ps/service/communicator/communicator.h

@@ -29,7 +29,7 @@ limitations under the License. */
 #include <vector>
 
 #include "gflags/gflags.h"
-#include "paddle/fluid/distributed/communicator_common.h"
+#include "paddle/fluid/distributed/ps/service/communicator/communicator_common.h"
 #include "paddle/fluid/framework/scope.h"
 #include "paddle/fluid/framework/variable.h"
 #include "paddle/fluid/framework/variable_helper.h"
@@ -41,7 +41,7 @@ limitations under the License. */
 #include "paddle/fluid/platform/place.h"
 #include "paddle/fluid/string/split.h"
 
-#include "paddle/fluid/distributed/service/ps_client.h"
+#include "paddle/fluid/distributed/ps/service/ps_client.h"
 
 namespace paddle {
 namespace distributed {
@@ -193,15 +193,15 @@ inline void MergeVars(const std::string &var_name,
       result.device(*cpu_ctx.eigen_device()) =
           result / static_cast<T>(vars.size());
     }
-  } else if (var0->IsType<framework::SelectedRows>()) {
-    auto &slr0 = var0->Get<framework::SelectedRows>();
-    auto *out_slr = out_var->GetMutable<framework::SelectedRows>();
+  } else if (var0->IsType<pten::SelectedRows>()) {
+    auto &slr0 = var0->Get<pten::SelectedRows>();
+    auto *out_slr = out_var->GetMutable<pten::SelectedRows>();
     out_slr->mutable_rows()->clear();
     out_slr->mutable_value()->mutable_data<T>({{}}, cpu_place);
-    std::vector<const paddle::framework::SelectedRows *> inputs;
+    std::vector<const pten::SelectedRows *> inputs;
     inputs.reserve(vars.size());
     for (auto &var : vars) {
-      inputs.push_back(&var->Get<framework::SelectedRows>());
+      inputs.push_back(&var->Get<pten::SelectedRows>());
     }
     auto dev_ctx = paddle::platform::CPUDeviceContext();
     if (merge_add) {

paddle/fluid/distributed/service/env.cc → paddle/fluid/distributed/ps/service/env.cc

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/service/env.h"
+#include "paddle/fluid/distributed/ps/service/env.h"
 
 namespace paddle {
 namespace distributed {}  // namespace distributed

paddle/fluid/distributed/service/graph_brpc_client.cc → paddle/fluid/distributed/ps/service/graph_brpc_client.cc

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/service/graph_brpc_client.h"
+#include "paddle/fluid/distributed/ps/service/graph_brpc_client.h"
 #include <algorithm>
 #include <memory>
 #include <sstream>
@@ -20,8 +20,8 @@
 #include <utility>
 #include <vector>
 #include "Eigen/Dense"
-#include "paddle/fluid/distributed/service/brpc_ps_client.h"
-#include "paddle/fluid/distributed/table/table.h"
+#include "paddle/fluid/distributed/ps/service/brpc_ps_client.h"
+#include "paddle/fluid/distributed/ps/table/table.h"
 #include "paddle/fluid/framework/archive.h"
 #include "paddle/fluid/string/string_helper.h"
 namespace paddle {

paddle/fluid/distributed/service/graph_brpc_client.h → paddle/fluid/distributed/ps/service/graph_brpc_client.h

@@ -24,10 +24,10 @@
 #include "brpc/channel.h"
 #include "brpc/controller.h"
 #include "brpc/server.h"
-#include "paddle/fluid/distributed/service/brpc_ps_client.h"
-#include "paddle/fluid/distributed/service/graph_brpc_server.h"
-#include "paddle/fluid/distributed/service/ps_client.h"
-#include "paddle/fluid/distributed/table/table.h"
+#include "paddle/fluid/distributed/ps/service/brpc_ps_client.h"
+#include "paddle/fluid/distributed/ps/service/graph_brpc_server.h"
+#include "paddle/fluid/distributed/ps/service/ps_client.h"
+#include "paddle/fluid/distributed/ps/table/table.h"
 #include "paddle/fluid/framework/lod_tensor.h"
 #include "paddle/fluid/framework/scope.h"
 #include "paddle/fluid/framework/tensor_util.h"

paddle/fluid/distributed/service/graph_brpc_server.cc → paddle/fluid/distributed/ps/service/graph_brpc_server.cc

@@ -12,14 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/service/graph_brpc_server.h"
-#include "paddle/fluid/distributed/service/brpc_ps_server.h"
+#include "paddle/fluid/distributed/ps/service/graph_brpc_server.h"
+#include "paddle/fluid/distributed/ps/service/brpc_ps_server.h"
 
 #include <thread>  // NOLINT
 #include <utility>
 #include "butil/endpoint.h"
 #include "iomanip"
-#include "paddle/fluid/distributed/service/brpc_ps_client.h"
+#include "paddle/fluid/distributed/ps/service/brpc_ps_client.h"
 #include "paddle/fluid/framework/archive.h"
 #include "paddle/fluid/platform/profiler.h"
 namespace paddle {

paddle/fluid/distributed/service/graph_brpc_server.h → paddle/fluid/distributed/ps/service/graph_brpc_server.h

@@ -20,10 +20,10 @@
 
 #include <memory>
 #include <vector>
-#include "paddle/fluid/distributed/service/brpc_ps_server.h"
-#include "paddle/fluid/distributed/service/server.h"
-#include "paddle/fluid/distributed/table/common_graph_table.h"
-#include "paddle/fluid/distributed/table/table.h"
+#include "paddle/fluid/distributed/ps/service/brpc_ps_server.h"
+#include "paddle/fluid/distributed/ps/service/server.h"
+#include "paddle/fluid/distributed/ps/table/common_graph_table.h"
+#include "paddle/fluid/distributed/ps/table/table.h"
 namespace paddle {
 namespace distributed {
 class GraphBrpcServer : public PSServer {

paddle/fluid/distributed/service/heter_client.cc → paddle/fluid/distributed/ps/service/heter_client.cc

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/service/heter_client.h"
+#include "paddle/fluid/distributed/ps/service/heter_client.h"
 #include "paddle/fluid/platform/profiler.h"
 #include "paddle/fluid/string/split.h"
 

paddle/fluid/distributed/service/heter_client.h → paddle/fluid/distributed/ps/service/heter_client.h

@@ -25,9 +25,9 @@ limitations under the License. */
 #include "brpc/channel.h"
 #include "brpc/controller.h"
 #include "brpc/server.h"
-#include "paddle/fluid/distributed/service/brpc_ps_client.h"
-#include "paddle/fluid/distributed/service/brpc_utils.h"
-#include "paddle/fluid/distributed/service/sendrecv.pb.h"
+#include "paddle/fluid/distributed/ps/service/brpc_ps_client.h"
+#include "paddle/fluid/distributed/ps/service/brpc_utils.h"
+#include "paddle/fluid/distributed/ps/service/sendrecv.pb.h"
 #include "paddle/fluid/framework/scope.h"
 #include "paddle/fluid/framework/tensor.h"
 #include "paddle/fluid/framework/variable_helper.h"

paddle/fluid/distributed/service/heter_server.cc → paddle/fluid/distributed/ps/service/heter_server.cc

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/service/heter_server.h"
+#include "paddle/fluid/distributed/ps/service/heter_server.h"
 #include "paddle/fluid/string/split.h"
 
 namespace paddle {

paddle/fluid/distributed/service/heter_server.h → paddle/fluid/distributed/ps/service/heter_server.h

@@ -25,8 +25,8 @@ limitations under the License. */
 #include "brpc/channel.h"
 #include "brpc/controller.h"
 #include "brpc/server.h"
-#include "paddle/fluid/distributed/service/brpc_utils.h"
-#include "paddle/fluid/distributed/service/sendrecv.pb.h"
+#include "paddle/fluid/distributed/ps/service/brpc_utils.h"
+#include "paddle/fluid/distributed/ps/service/sendrecv.pb.h"
 #include "paddle/fluid/framework/blocking_queue.h"
 #include "paddle/fluid/framework/executor.h"
 #include "paddle/fluid/framework/program_desc.h"

paddle/fluid/distributed/service/ps_client.cc → paddle/fluid/distributed/ps/service/ps_client.cc

@@ -12,12 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/service/ps_client.h"
+#include "paddle/fluid/distributed/ps/service/ps_client.h"
 #include "glog/logging.h"
-#include "paddle/fluid/distributed/service/brpc_ps_client.h"
-#include "paddle/fluid/distributed/service/graph_brpc_client.h"
-#include "paddle/fluid/distributed/service/ps_local_client.h"
-#include "paddle/fluid/distributed/table/table.h"
+#include "paddle/fluid/distributed/ps/service/brpc_ps_client.h"
+#include "paddle/fluid/distributed/ps/service/graph_brpc_client.h"
+#include "paddle/fluid/distributed/ps/service/ps_local_client.h"
+#include "paddle/fluid/distributed/ps/table/table.h"
 
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/service/ps_client.h → paddle/fluid/distributed/ps/service/ps_client.h

@@ -22,10 +22,10 @@
 #include <vector>
 #include "paddle/fluid/distributed/common/cost_timer.h"
 #include "paddle/fluid/distributed/ps.pb.h"
-#include "paddle/fluid/distributed/service/env.h"
-#include "paddle/fluid/distributed/service/sendrecv.pb.h"
-#include "paddle/fluid/distributed/table/accessor.h"
-#include "paddle/fluid/distributed/table/graph/graph_node.h"
+#include "paddle/fluid/distributed/ps/service/env.h"
+#include "paddle/fluid/distributed/ps/service/sendrecv.pb.h"
+#include "paddle/fluid/distributed/ps/table/accessor.h"
+#include "paddle/fluid/distributed/ps/table/graph/graph_node.h"
 #include "paddle/fluid/platform/timer.h"
 
 namespace paddle {

paddle/fluid/distributed/service/ps_local_client.cc → paddle/fluid/distributed/ps/service/ps_local_client.cc

@@ -12,8 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/service/ps_local_client.h"
-#include "paddle/fluid/distributed/table/table.h"
+#include "paddle/fluid/distributed/ps/service/ps_local_client.h"
+#include "paddle/fluid/distributed/ps/table/table.h"
 
 //#define pslib_debug_dense_compress
 

paddle/fluid/distributed/service/ps_local_client.h → paddle/fluid/distributed/ps/service/ps_local_client.h

@@ -15,7 +15,7 @@
 #include "brpc/channel.h"
 #include "brpc/controller.h"
 #include "brpc/server.h"
-#include "paddle/fluid/distributed/service/ps_client.h"
+#include "paddle/fluid/distributed/ps/service/ps_client.h"
 
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/service/ps_local_server.h → paddle/fluid/distributed/ps/service/ps_local_server.h

@@ -16,7 +16,7 @@
 
 #include <memory>
 #include <vector>
-#include "paddle/fluid/distributed/service/server.h"
+#include "paddle/fluid/distributed/ps/service/server.h"
 
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/service/graph_py_service.cc → paddle/fluid/distributed/ps/service/ps_service/graph_py_service.cc

@@ -12,11 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/service/graph_py_service.h"
+#include "paddle/fluid/distributed/ps/service/ps_service/graph_py_service.h"
 #include <thread>  // NOLINT
 #include "butil/endpoint.h"
 #include "iomanip"
-#include "paddle/fluid/distributed/table/table.h"
+#include "paddle/fluid/distributed/ps/table/table.h"
 #include "paddle/fluid/framework/archive.h"
 #include "paddle/fluid/platform/profiler.h"
 namespace paddle {

paddle/fluid/distributed/service/graph_py_service.h → paddle/fluid/distributed/ps/service/ps_service/graph_py_service.h

@@ -32,11 +32,11 @@
 #include "paddle/fluid/framework/variable.h"
 
 #include "paddle/fluid/distributed/ps.pb.h"
-#include "paddle/fluid/distributed/service/env.h"
-#include "paddle/fluid/distributed/service/graph_brpc_client.h"
-#include "paddle/fluid/distributed/service/graph_brpc_server.h"
-#include "paddle/fluid/distributed/service/sendrecv.pb.h"
-#include "paddle/fluid/distributed/service/service.h"
+#include "paddle/fluid/distributed/ps/service/env.h"
+#include "paddle/fluid/distributed/ps/service/graph_brpc_client.h"
+#include "paddle/fluid/distributed/ps/service/graph_brpc_server.h"
+#include "paddle/fluid/distributed/ps/service/ps_service/service.h"
+#include "paddle/fluid/distributed/ps/service/sendrecv.pb.h"
 #include "paddle/fluid/framework/program_desc.h"
 #include "paddle/fluid/operators/math/math_function.h"
 #include "paddle/fluid/platform/place.h"

paddle/fluid/distributed/service/service.cc → paddle/fluid/distributed/ps/service/ps_service/service.cc

@@ -12,13 +12,13 @@
    See the License for the specific language governing permissions and
    limitations under the License. */
 
-#include "paddle/fluid/distributed/service/service.h"
+#include "paddle/fluid/distributed/ps/service/ps_service/service.h"
 
 #include <fcntl.h>
 #include <google/protobuf/io/zero_copy_stream_impl.h>
 #include <google/protobuf/text_format.h>
 #include <iostream>
-#include "paddle/fluid/distributed/service/communicator.h"
+#include "paddle/fluid/distributed/ps/service/communicator/communicator.h"
 #include "paddle/fluid/string/string_helper.h"
 
 using namespace std;  // NOLINT

paddle/fluid/distributed/service/service.h → paddle/fluid/distributed/ps/service/ps_service/service.h

@@ -20,9 +20,9 @@ limitations under the License. */
 #include <vector>
 
 #include "paddle/fluid/distributed/ps.pb.h"
-#include "paddle/fluid/distributed/service/ps_client.h"
-#include "paddle/fluid/distributed/service/sendrecv.pb.h"
-#include "paddle/fluid/distributed/service/server.h"
+#include "paddle/fluid/distributed/ps/service/ps_client.h"
+#include "paddle/fluid/distributed/ps/service/sendrecv.pb.h"
+#include "paddle/fluid/distributed/ps/service/server.h"
 
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/service/server.cc → paddle/fluid/distributed/ps/service/server.cc

@@ -12,13 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/service/server.h"
+#include "paddle/fluid/distributed/ps/service/server.h"
 
 #include "glog/logging.h"
-#include "paddle/fluid/distributed/service/brpc_ps_server.h"
-#include "paddle/fluid/distributed/service/graph_brpc_server.h"
-#include "paddle/fluid/distributed/service/ps_local_server.h"
-#include "paddle/fluid/distributed/table/table.h"
+#include "paddle/fluid/distributed/ps/service/brpc_ps_server.h"
+#include "paddle/fluid/distributed/ps/service/graph_brpc_server.h"
+#include "paddle/fluid/distributed/ps/service/ps_local_server.h"
+#include "paddle/fluid/distributed/ps/table/table.h"
 
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/service/server.h → paddle/fluid/distributed/ps/service/server.h

@@ -24,8 +24,8 @@
 #include "google/protobuf/service.h"
 #include "paddle/fluid/distributed/common/registerer.h"
 #include "paddle/fluid/distributed/ps.pb.h"
-#include "paddle/fluid/distributed/service/env.h"
-#include "paddle/fluid/distributed/service/sendrecv.pb.h"
+#include "paddle/fluid/distributed/ps/service/env.h"
+#include "paddle/fluid/distributed/ps/service/sendrecv.pb.h"
 #include "paddle/fluid/framework/channel.h"
 #include "paddle/fluid/framework/scope.h"
 #include "paddle/fluid/platform/device_context.h"

paddle/fluid/distributed/table/barrier_table.cc → paddle/fluid/distributed/ps/table/barrier_table.cc

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/table/common_table.h"
+#include "paddle/fluid/distributed/ps/table/common_table.h"
 
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/table/common_dense_table.cc → paddle/fluid/distributed/ps/table/common_dense_table.cc

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/table/common_dense_table.h"
+#include "paddle/fluid/distributed/ps/table/common_dense_table.h"
 
 #include "paddle/fluid/platform/enforce.h"
 

paddle/fluid/distributed/table/common_dense_table.h → paddle/fluid/distributed/ps/table/common_dense_table.h

@@ -19,10 +19,10 @@
 #include <pthread.h>
 #include <string>
 #include "Eigen/Dense"
-#include "paddle/fluid/distributed/table/accessor.h"
-#include "paddle/fluid/distributed/table/common_table.h"
-#include "paddle/fluid/distributed/table/depends/dense.h"
-#include "paddle/fluid/distributed/table/depends/initializers.h"
+#include "paddle/fluid/distributed/ps/table/accessor.h"
+#include "paddle/fluid/distributed/ps/table/common_table.h"
+#include "paddle/fluid/distributed/ps/table/depends/dense.h"
+#include "paddle/fluid/distributed/ps/table/depends/initializers.h"
 #include "paddle/fluid/string/string_helper.h"
 
 namespace paddle {

paddle/fluid/distributed/table/common_graph_table.cc → paddle/fluid/distributed/ps/table/common_graph_table.cc

@@ -12,14 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/table/common_graph_table.h"
+#include "paddle/fluid/distributed/ps/table/common_graph_table.h"
 #include <time.h>
 #include <algorithm>
 #include <chrono>
 #include <set>
 #include <sstream>
 #include "paddle/fluid/distributed/common/utils.h"
-#include "paddle/fluid/distributed/table/graph/graph_node.h"
+#include "paddle/fluid/distributed/ps/table/graph/graph_node.h"
 #include "paddle/fluid/framework/generator.h"
 #include "paddle/fluid/string/printf.h"
 #include "paddle/fluid/string/string_helper.h"

paddle/fluid/distributed/table/common_graph_table.h → paddle/fluid/distributed/ps/table/common_graph_table.h

@@ -36,11 +36,12 @@
 #include <unordered_set>
 #include <utility>
 #include <vector>
-#include "paddle/fluid/distributed/table/accessor.h"
-#include "paddle/fluid/distributed/table/common_table.h"
-#include "paddle/fluid/distributed/table/graph/graph_node.h"
-#include "paddle/fluid/framework/rw_lock.h"
+#include "paddle/fluid/distributed/ps/table/accessor.h"
+#include "paddle/fluid/distributed/ps/table/common_table.h"
+#include "paddle/fluid/distributed/ps/table/graph/graph_node.h"
 #include "paddle/fluid/string/string_helper.h"
+#include "paddle/pten/core/utils/rw_lock.h"
+
 namespace paddle {
 namespace distributed {
 class GraphShard {

paddle/fluid/distributed/table/common_sparse_table.cc → paddle/fluid/distributed/ps/table/common_sparse_table.cc

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/table/common_sparse_table.h"
+#include "paddle/fluid/distributed/ps/table/common_sparse_table.h"
 #include <sstream>
 
 #include "glog/logging.h"

paddle/fluid/distributed/table/common_sparse_table.h → paddle/fluid/distributed/ps/table/common_sparse_table.h

@@ -24,13 +24,13 @@
 #include <utility>
 #include <vector>
 #include "Eigen/Dense"
-#include "paddle/fluid/distributed/table/accessor.h"
-#include "paddle/fluid/distributed/table/common_table.h"
-#include "paddle/fluid/distributed/table/depends/initializers.h"
-#include "paddle/fluid/distributed/table/depends/large_scale_kv.h"
-#include "paddle/fluid/distributed/table/depends/sparse.h"
-#include "paddle/fluid/framework/rw_lock.h"
+#include "paddle/fluid/distributed/ps/table/accessor.h"
+#include "paddle/fluid/distributed/ps/table/common_table.h"
+#include "paddle/fluid/distributed/ps/table/depends/initializers.h"
+#include "paddle/fluid/distributed/ps/table/depends/large_scale_kv.h"
+#include "paddle/fluid/distributed/ps/table/depends/sparse.h"
 #include "paddle/fluid/string/string_helper.h"
+#include "paddle/pten/core/utils/rw_lock.h"
 
 #define PSERVER_SAVE_SUFFIX ".shard"
 
@@ -110,7 +110,7 @@ struct Meta {
 
 class CommonSparseTable : public SparseTable {
  public:
-  CommonSparseTable() { rwlock_.reset(new framework::RWLock); }
+  CommonSparseTable() { rwlock_.reset(new pten::RWLock); }
   virtual ~CommonSparseTable() {}
 
   // unused method begin
@@ -193,7 +193,7 @@ class CommonSparseTable : public SparseTable {
   std::shared_ptr<SparseOptimizer> optimizer_;
   std::vector<std::shared_ptr<ValueBlock>> shard_values_;
   std::unordered_map<uint64_t, ReservoirValue<float>> pull_reservoir_;
-  std::unique_ptr<framework::RWLock> rwlock_{nullptr};
+  std::unique_ptr<pten::RWLock> rwlock_{nullptr};
 };
 
 }  // namespace distributed

paddle/fluid/distributed/table/common_table.h → paddle/fluid/distributed/ps/table/common_table.h

@@ -19,7 +19,7 @@
 #include <mutex>               // NOLINT
 #include <set>
 
-#include "paddle/fluid/distributed/table/table.h"
+#include "paddle/fluid/distributed/ps/table/table.h"
 
 #include "paddle/fluid/distributed/common/utils.h"
 

paddle/fluid/distributed/table/ctr_accessor.cc → paddle/fluid/distributed/ps/table/ctr_accessor.cc

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/table/ctr_accessor.h"
+#include "paddle/fluid/distributed/ps/table/ctr_accessor.h"
 #include <gflags/gflags.h>
 #include "glog/logging.h"
 #include "paddle/fluid/string/string_helper.h"

paddle/fluid/distributed/table/ctr_accessor.h → paddle/fluid/distributed/ps/table/ctr_accessor.h

@@ -18,8 +18,8 @@
 #include <vector>
 #include "paddle/fluid/distributed/common/registerer.h"
 #include "paddle/fluid/distributed/ps.pb.h"
-#include "paddle/fluid/distributed/table/accessor.h"
-#include "paddle/fluid/distributed/table/sparse_sgd_rule.h"
+#include "paddle/fluid/distributed/ps/table/accessor.h"
+#include "paddle/fluid/distributed/ps/table/sparse_sgd_rule.h"
 
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/table/depends/large_scale_kv.h → paddle/fluid/distributed/ps/table/depends/large_scale_kv.h

@@ -28,11 +28,10 @@
 
 #include "butil/object_pool.h"
 #include "paddle/fluid/distributed/common/utils.h"
-#include "paddle/fluid/distributed/table/depends/initializers.h"
-#include "paddle/fluid/distributed/thirdparty/round_robin.h"
+#include "paddle/fluid/distributed/ps/table/depends/initializers.h"
+#include "paddle/fluid/distributed/ps/thirdparty/round_robin.h"
 #include "paddle/fluid/framework/generator.h"
 #include "paddle/fluid/framework/lod_tensor.h"
-#include "paddle/fluid/framework/rw_lock.h"
 #include "paddle/fluid/framework/selected_rows_utils.h"
 #include "paddle/fluid/framework/tensor.h"
 #include "paddle/fluid/framework/threadpool.h"
@@ -43,6 +42,7 @@
 #include "paddle/fluid/string/printf.h"
 #include "paddle/fluid/string/string_helper.h"
 #include "paddle/pten/backends/dynload/port.h"
+#include "paddle/pten/core/utils/rw_lock.h"
 
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/table/depends/sparse.h → paddle/fluid/distributed/ps/table/depends/sparse.h

@@ -24,7 +24,7 @@
 #include "gflags/gflags.h"
 
 #include "paddle/fluid/distributed/common/utils.h"
-#include "paddle/fluid/distributed/table/depends/large_scale_kv.h"
+#include "paddle/fluid/distributed/ps/table/depends/large_scale_kv.h"
 
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/table/graph/graph_edge.cc → paddle/fluid/distributed/ps/table/graph/graph_edge.cc

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/table/graph/graph_edge.h"
+#include "paddle/fluid/distributed/ps/table/graph/graph_edge.h"
 #include <cstring>
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/table/graph/graph_node.cc → paddle/fluid/distributed/ps/table/graph/graph_node.cc

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/table/graph/graph_node.h"
+#include "paddle/fluid/distributed/ps/table/graph/graph_node.h"
 #include <cstring>
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/table/graph/graph_node.h → paddle/fluid/distributed/ps/table/graph/graph_node.h

@@ -18,7 +18,7 @@
 #include <memory>
 #include <sstream>
 #include <vector>
-#include "paddle/fluid/distributed/table/graph/graph_weighted_sampler.h"
+#include "paddle/fluid/distributed/ps/table/graph/graph_weighted_sampler.h"
 namespace paddle {
 namespace distributed {
 

paddle/fluid/distributed/table/graph/graph_weighted_sampler.cc → paddle/fluid/distributed/ps/table/graph/graph_weighted_sampler.cc

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/table/graph/graph_weighted_sampler.h"
+#include "paddle/fluid/distributed/ps/table/graph/graph_weighted_sampler.h"
 #include <iostream>
 #include <memory>
 #include <unordered_map>

paddle/fluid/distributed/table/graph/graph_weighted_sampler.h → paddle/fluid/distributed/ps/table/graph/graph_weighted_sampler.h

@@ -18,7 +18,7 @@
 #include <random>
 #include <unordered_map>
 #include <vector>
-#include "paddle/fluid/distributed/table/graph/graph_edge.h"
+#include "paddle/fluid/distributed/ps/table/graph/graph_edge.h"
 namespace paddle {
 namespace distributed {
 

paddle/fluid/distributed/table/memory_sparse_table.cc → paddle/fluid/distributed/ps/table/memory_sparse_table.cc

@@ -16,7 +16,7 @@
 #include <sstream>
 
 #include "paddle/fluid/distributed/common/cost_timer.h"
-#include "paddle/fluid/distributed/table/memory_sparse_table.h"
+#include "paddle/fluid/distributed/ps/table/memory_sparse_table.h"
 #include "paddle/fluid/framework/io/fs.h"
 
 #include "boost/lexical_cast.hpp"

paddle/fluid/distributed/table/memory_sparse_table.h → paddle/fluid/distributed/ps/table/memory_sparse_table.h

@@ -24,9 +24,9 @@
 #include <utility>
 #include <vector>
 #include "Eigen/Dense"
-#include "paddle/fluid/distributed/table/accessor.h"
-#include "paddle/fluid/distributed/table/common_table.h"
-#include "paddle/fluid/distributed/table/depends/feature_value.h"
+#include "paddle/fluid/distributed/ps/table/accessor.h"
+#include "paddle/fluid/distributed/ps/table/common_table.h"
+#include "paddle/fluid/distributed/ps/table/depends/feature_value.h"
 #include "paddle/fluid/string/string_helper.h"
 
 #define PSERVER_SAVE_SUFFIX ".shard"

paddle/fluid/distributed/table/sparse_geo_table.cc → paddle/fluid/distributed/ps/table/sparse_geo_table.cc

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/table/sparse_geo_table.h"
+#include "paddle/fluid/distributed/ps/table/sparse_geo_table.h"
 
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/table/sparse_geo_table.h → paddle/fluid/distributed/ps/table/sparse_geo_table.h

@@ -24,15 +24,15 @@
 #include <vector>
 
 #include "Eigen/Dense"
-#include "paddle/fluid/distributed/table/accessor.h"
-#include "paddle/fluid/distributed/table/common_sparse_table.h"
-#include "paddle/fluid/distributed/table/common_table.h"
-#include "paddle/fluid/distributed/table/depends/geo_recorder.h"
-#include "paddle/fluid/distributed/table/depends/initializers.h"
-#include "paddle/fluid/distributed/table/depends/large_scale_kv.h"
-#include "paddle/fluid/distributed/table/depends/sparse.h"
-#include "paddle/fluid/framework/rw_lock.h"
+#include "paddle/fluid/distributed/ps/table/accessor.h"
+#include "paddle/fluid/distributed/ps/table/common_sparse_table.h"
+#include "paddle/fluid/distributed/ps/table/common_table.h"
+#include "paddle/fluid/distributed/ps/table/depends/geo_recorder.h"
+#include "paddle/fluid/distributed/ps/table/depends/initializers.h"
+#include "paddle/fluid/distributed/ps/table/depends/large_scale_kv.h"
+#include "paddle/fluid/distributed/ps/table/depends/sparse.h"
 #include "paddle/fluid/string/string_helper.h"
+#include "paddle/pten/core/utils/rw_lock.h"
 
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/table/sparse_sgd_rule.cc → paddle/fluid/distributed/ps/table/sparse_sgd_rule.cc

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/table/sparse_sgd_rule.h"
+#include "paddle/fluid/distributed/ps/table/sparse_sgd_rule.h"
 #include <gflags/gflags.h>
 #include "glog/logging.h"
 

paddle/fluid/distributed/table/ssd_sparse_table.cc → paddle/fluid/distributed/ps/table/ssd_sparse_table.cc

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 #ifdef PADDLE_WITH_HETERPS
-#include "paddle/fluid/distributed/table/ssd_sparse_table.h"
+#include "paddle/fluid/distributed/ps/table/ssd_sparse_table.h"
 
 DEFINE_string(rocksdb_path, "database", "path of sparse table rocksdb file");
 

paddle/fluid/distributed/table/ssd_sparse_table.h → paddle/fluid/distributed/ps/table/ssd_sparse_table.h

@@ -13,8 +13,8 @@
 // limitations under the License.
 
 #pragma once
-#include "paddle/fluid/distributed/table/common_sparse_table.h"
-#include "paddle/fluid/distributed/table/depends/rocksdb_warpper.h"
+#include "paddle/fluid/distributed/ps/table/common_sparse_table.h"
+#include "paddle/fluid/distributed/ps/table/depends/rocksdb_warpper.h"
 #ifdef PADDLE_WITH_HETERPS
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/table/table.cc → paddle/fluid/distributed/ps/table/table.cc

@@ -12,22 +12,22 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/table/table.h"
+#include "paddle/fluid/distributed/ps/table/table.h"
 
 #include "glog/logging.h"
 #include "paddle/fluid/distributed/common/registerer.h"
 
-#include "paddle/fluid/distributed/table/common_dense_table.h"
-#include "paddle/fluid/distributed/table/common_graph_table.h"
-#include "paddle/fluid/distributed/table/common_sparse_table.h"
-#include "paddle/fluid/distributed/table/sparse_geo_table.h"
+#include "paddle/fluid/distributed/ps/table/common_dense_table.h"
+#include "paddle/fluid/distributed/ps/table/common_graph_table.h"
+#include "paddle/fluid/distributed/ps/table/common_sparse_table.h"
+#include "paddle/fluid/distributed/ps/table/sparse_geo_table.h"
 #ifdef PADDLE_WITH_HETERPS
-#include "paddle/fluid/distributed/table/ssd_sparse_table.h"
+#include "paddle/fluid/distributed/ps/table/ssd_sparse_table.h"
 #endif
-#include "paddle/fluid/distributed/table/ctr_accessor.h"
-#include "paddle/fluid/distributed/table/memory_sparse_table.h"
-#include "paddle/fluid/distributed/table/tensor_accessor.h"
-#include "paddle/fluid/distributed/table/tensor_table.h"
+#include "paddle/fluid/distributed/ps/table/ctr_accessor.h"
+#include "paddle/fluid/distributed/ps/table/memory_sparse_table.h"
+#include "paddle/fluid/distributed/ps/table/tensor_accessor.h"
+#include "paddle/fluid/distributed/ps/table/tensor_table.h"
 
 namespace paddle {
 namespace distributed {
@@ -83,9 +83,10 @@ int32_t Table::initialize_accessor() {
 
   LOG(INFO) << "accessor initializing: table_id: " << _config.table_id()
             << ", accessor_name: " << _config.accessor().accessor_class();
-  auto *accessor = CREATE_PSCORE_CLASS(
-      ValueAccessor,
-      _config.accessor().accessor_class()) if (accessor == NULL) {
+  auto *accessor =
+      CREATE_PSCORE_CLASS(ValueAccessor, _config.accessor().accessor_class());
+
+  if (accessor == NULL) {
     LOG(ERROR) << "accessor is unregisteg, table_id:" << _config.table_id()
                << ", accessor_name:" << _config.accessor().accessor_class();
     return -1;

paddle/fluid/distributed/table/table.h → paddle/fluid/distributed/ps/table/table.h

@@ -21,9 +21,9 @@
 #include <string>
 #include <utility>
 #include "paddle/fluid/distributed/common/afs_warpper.h"
-#include "paddle/fluid/distributed/table/accessor.h"
-#include "paddle/fluid/distributed/table/depends/sparse_utils.h"
-#include "paddle/fluid/distributed/table/graph/graph_node.h"
+#include "paddle/fluid/distributed/ps/table/accessor.h"
+#include "paddle/fluid/distributed/ps/table/depends/sparse_utils.h"
+#include "paddle/fluid/distributed/ps/table/graph/graph_node.h"
 #include "paddle/fluid/framework/program_desc.h"
 #include "paddle/fluid/framework/scope.h"
 #include "paddle/fluid/platform/device_context.h"

paddle/fluid/distributed/table/tensor_accessor.cc → paddle/fluid/distributed/ps/table/tensor_accessor.cc

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/table/tensor_accessor.h"
+#include "paddle/fluid/distributed/ps/table/tensor_accessor.h"
 #include "Eigen/Dense"
 
 namespace paddle {

paddle/fluid/distributed/table/tensor_accessor.h → paddle/fluid/distributed/ps/table/tensor_accessor.h

@@ -20,7 +20,7 @@
 
 #include "paddle/fluid/distributed/common/registerer.h"
 #include "paddle/fluid/distributed/ps.pb.h"
-#include "paddle/fluid/distributed/table/accessor.h"
+#include "paddle/fluid/distributed/ps/table/accessor.h"
 
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/table/tensor_table.cc → paddle/fluid/distributed/ps/table/tensor_table.cc

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/distributed/table/tensor_table.h"
+#include "paddle/fluid/distributed/ps/table/tensor_table.h"
 
 DECLARE_double(eager_delete_tensor_gb);
 namespace paddle {

paddle/fluid/distributed/table/tensor_table.h → paddle/fluid/distributed/ps/table/tensor_table.h

@@ -24,7 +24,7 @@
 #include <vector>
 
 #include "paddle/fluid/distributed/common/utils.h"
-#include "paddle/fluid/distributed/table/table.h"
+#include "paddle/fluid/distributed/ps/table/table.h"
 #include "paddle/fluid/framework/executor.h"
 #include "paddle/fluid/framework/program_desc.h"
 #include "paddle/fluid/platform/device_context.h"

paddle/fluid/distributed/ps/wrapper/CMakeLists.txt

+
+get_property(RPC_DEPS GLOBAL PROPERTY RPC_DEPS)
+
+set_source_files_properties(fleet.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
+cc_library(fleet
+        SRCS fleet.cc
+        DEPS framework_proto ps_framework_proto ps_service variable_helper scope op_registry fs shell ${RPC_DEPS})
+
+target_link_libraries(fleet z)

paddle/fluid/distributed/fleet.cc → paddle/fluid/distributed/ps/wrapper/fleet.cc

@@ -12,9 +12,9 @@ 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 "paddle/fluid/distributed/fleet.h"
-#include "paddle/fluid/distributed/service/communicator.h"
-#include "paddle/fluid/distributed/table/table.h"
+#include "paddle/fluid/distributed/ps/wrapper/fleet.h"
+#include "paddle/fluid/distributed/ps/service/communicator/communicator.h"
+#include "paddle/fluid/distributed/ps/table/table.h"
 
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/fleet.h → paddle/fluid/distributed/ps/wrapper/fleet.h

@@ -23,8 +23,8 @@ limitations under the License. */
 #include <unordered_map>
 #include <vector>
 
-#include "paddle/fluid/distributed/communicator_common.h"
-#include "paddle/fluid/distributed/service/service.h"
+#include "paddle/fluid/distributed/ps/service/communicator/communicator_common.h"
+#include "paddle/fluid/distributed/ps/service/ps_service/service.h"
 #include "paddle/fluid/framework/archive.h"
 #include "paddle/fluid/framework/io/fs.h"
 #include "paddle/fluid/framework/io/shell.h"
@@ -49,7 +49,7 @@ class PSCore;
 
 using framework::LoDTensor;
 using framework::Scope;
-using framework::SelectedRows;
+using pten::SelectedRows;
 using framework::Variable;
 
 using RpcCtxMap = std::unordered_map<std::string, CommContext>;

paddle/fluid/distributed/test/barrier_table_test.cc

@@ -17,8 +17,8 @@ limitations under the License. */
 #include <vector>
 #include "gtest/gtest.h"
 #include "paddle/fluid/distributed/ps.pb.h"
-#include "paddle/fluid/distributed/table/common_table.h"
-#include "paddle/fluid/distributed/table/table.h"
+#include "paddle/fluid/distributed/ps/table/common_table.h"
+#include "paddle/fluid/distributed/ps/table/table.h"
 
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/test/brpc_service_dense_sgd_test.cc

@@ -17,8 +17,8 @@ limitations under the License. */
 #include <thread>  // NOLINT
 
 #include "gtest/gtest.h"
-#include "paddle/fluid/distributed/service/brpc_ps_client.h"
-#include "paddle/fluid/distributed/service/brpc_ps_server.h"
+#include "paddle/fluid/distributed/ps/service/brpc_ps_client.h"
+#include "paddle/fluid/distributed/ps/service/brpc_ps_server.h"
 #include "paddle/fluid/framework/program_desc.h"
 #include "paddle/fluid/framework/scope.h"
 #include "paddle/fluid/operators/math/math_function.h"

paddle/fluid/distributed/test/brpc_service_sparse_sgd_test.cc

@@ -18,9 +18,9 @@ limitations under the License. */
 
 #include "gtest/gtest.h"
 #include "paddle/fluid/distributed/ps.pb.h"
-#include "paddle/fluid/distributed/service/brpc_ps_client.h"
-#include "paddle/fluid/distributed/service/brpc_ps_server.h"
-#include "paddle/fluid/distributed/service/env.h"
+#include "paddle/fluid/distributed/ps/service/brpc_ps_client.h"
+#include "paddle/fluid/distributed/ps/service/brpc_ps_server.h"
+#include "paddle/fluid/distributed/ps/service/env.h"
 #include "paddle/fluid/framework/program_desc.h"
 #include "paddle/fluid/operators/math/math_function.h"
 #include "paddle/fluid/platform/place.h"

paddle/fluid/distributed/test/brpc_utils_test.cc

@@ -16,7 +16,7 @@ limitations under the License. */
 
 #include "gtest/gtest.h"
 
-#include "paddle/fluid/distributed/service/brpc_utils.h"
+#include "paddle/fluid/distributed/ps/service/brpc_utils.h"
 #include "paddle/fluid/operators/math/math_function.h"
 
 namespace paddle {
@@ -56,7 +56,7 @@ void CreateVarsOnScope(framework::Scope* scope, platform::Place* place,
 
   // var 3
   framework::Variable* var3 = scope->Var("x3");
-  auto* slr = var3->GetMutable<framework::SelectedRows>();
+  auto* slr = var3->GetMutable<pten::SelectedRows>();
   slr->set_height(564);
   auto* tensor3 = slr->mutable_value();
   auto* rows = slr->mutable_rows();
@@ -111,7 +111,7 @@ void RunMultiVarMsg(platform::Place place) {
 
   // check var3
   framework::Variable* var3 = scope_recv.FindVar("x3");
-  auto* slr = var3->GetMutable<framework::SelectedRows>();
+  auto* slr = var3->GetMutable<pten::SelectedRows>();
   EXPECT_EQ(slr->rows().size(), 564);
   for (int i = 0; i < 564; ++i) {
     EXPECT_EQ(slr->rows()[i], i);

paddle/fluid/distributed/test/ctr_accessor_test.cc

@@ -12,13 +12,13 @@ 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 "paddle/fluid/distributed/table/ctr_accessor.h"
+#include "paddle/fluid/distributed/ps/table/ctr_accessor.h"
 #include <cmath>
 #include <iostream>
 #include "gtest/gtest.h"
 #include "paddle/fluid/distributed/common/registerer.h"
 #include "paddle/fluid/distributed/ps.pb.h"
-#include "paddle/fluid/distributed/table/sparse_sgd_rule.h"
+#include "paddle/fluid/distributed/ps/table/sparse_sgd_rule.h"
 
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/test/dense_table_test.cc

@@ -16,7 +16,7 @@ limitations under the License. */
 #include <vector>
 #include "gtest/gtest.h"
 #include "paddle/fluid/distributed/ps.pb.h"
-#include "paddle/fluid/distributed/table/common_dense_table.h"
+#include "paddle/fluid/distributed/ps/table/common_dense_table.h"
 
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/test/feature_value_test.cc

@@ -12,7 +12,7 @@ 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 "paddle/fluid/distributed/table/depends/feature_value.h"
+#include "paddle/fluid/distributed/ps/table/depends/feature_value.h"
 #include <vector>
 #include "gtest/gtest.h"
 

paddle/fluid/distributed/test/geo_table_test.cc

@@ -21,11 +21,11 @@ limitations under the License. */
 #include "google/protobuf/text_format.h"
 #include "gtest/gtest.h"
 #include "paddle/fluid/distributed/ps.pb.h"
-#include "paddle/fluid/distributed/table/common_dense_table.h"
-#include "paddle/fluid/distributed/table/common_sparse_table.h"
-#include "paddle/fluid/distributed/table/depends/sparse_utils.h"
-#include "paddle/fluid/distributed/table/sparse_geo_table.h"
-#include "paddle/fluid/distributed/table/table.h"
+#include "paddle/fluid/distributed/ps/table/common_dense_table.h"
+#include "paddle/fluid/distributed/ps/table/common_sparse_table.h"
+#include "paddle/fluid/distributed/ps/table/depends/sparse_utils.h"
+#include "paddle/fluid/distributed/ps/table/sparse_geo_table.h"
+#include "paddle/fluid/distributed/ps/table/table.h"
 
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/test/graph_node_split_test.cc

@@ -21,16 +21,16 @@ limitations under the License. */
 
 #include "gtest/gtest.h"
 #include "paddle/fluid/distributed/ps.pb.h"
-#include "paddle/fluid/distributed/service/brpc_ps_client.h"
-#include "paddle/fluid/distributed/service/brpc_ps_server.h"
-#include "paddle/fluid/distributed/service/env.h"
-#include "paddle/fluid/distributed/service/graph_brpc_client.h"
-#include "paddle/fluid/distributed/service/graph_brpc_server.h"
-#include "paddle/fluid/distributed/service/graph_py_service.h"
-#include "paddle/fluid/distributed/service/ps_client.h"
-#include "paddle/fluid/distributed/service/sendrecv.pb.h"
-#include "paddle/fluid/distributed/service/service.h"
-#include "paddle/fluid/distributed/table/graph/graph_node.h"
+#include "paddle/fluid/distributed/ps/service/brpc_ps_client.h"
+#include "paddle/fluid/distributed/ps/service/brpc_ps_server.h"
+#include "paddle/fluid/distributed/ps/service/env.h"
+#include "paddle/fluid/distributed/ps/service/graph_brpc_client.h"
+#include "paddle/fluid/distributed/ps/service/graph_brpc_server.h"
+#include "paddle/fluid/distributed/ps/service/ps_client.h"
+#include "paddle/fluid/distributed/ps/service/ps_service/graph_py_service.h"
+#include "paddle/fluid/distributed/ps/service/ps_service/service.h"
+#include "paddle/fluid/distributed/ps/service/sendrecv.pb.h"
+#include "paddle/fluid/distributed/ps/table/graph/graph_node.h"
 #include "paddle/fluid/framework/lod_tensor.h"
 #include "paddle/fluid/framework/program_desc.h"
 #include "paddle/fluid/framework/scope.h"

paddle/fluid/distributed/test/graph_node_test.cc

@@ -21,16 +21,16 @@ limitations under the License. */
 
 #include "gtest/gtest.h"
 #include "paddle/fluid/distributed/ps.pb.h"
-#include "paddle/fluid/distributed/service/brpc_ps_client.h"
-#include "paddle/fluid/distributed/service/brpc_ps_server.h"
-#include "paddle/fluid/distributed/service/env.h"
-#include "paddle/fluid/distributed/service/graph_brpc_client.h"
-#include "paddle/fluid/distributed/service/graph_brpc_server.h"
-#include "paddle/fluid/distributed/service/graph_py_service.h"
-#include "paddle/fluid/distributed/service/ps_client.h"
-#include "paddle/fluid/distributed/service/sendrecv.pb.h"
-#include "paddle/fluid/distributed/service/service.h"
-#include "paddle/fluid/distributed/table/graph/graph_node.h"
+#include "paddle/fluid/distributed/ps/service/brpc_ps_client.h"
+#include "paddle/fluid/distributed/ps/service/brpc_ps_server.h"
+#include "paddle/fluid/distributed/ps/service/env.h"
+#include "paddle/fluid/distributed/ps/service/graph_brpc_client.h"
+#include "paddle/fluid/distributed/ps/service/graph_brpc_server.h"
+#include "paddle/fluid/distributed/ps/service/ps_client.h"
+#include "paddle/fluid/distributed/ps/service/ps_service/graph_py_service.h"
+#include "paddle/fluid/distributed/ps/service/ps_service/service.h"
+#include "paddle/fluid/distributed/ps/service/sendrecv.pb.h"
+#include "paddle/fluid/distributed/ps/table/graph/graph_node.h"
 #include "paddle/fluid/framework/lod_tensor.h"
 #include "paddle/fluid/framework/program_desc.h"
 #include "paddle/fluid/framework/scope.h"

paddle/fluid/distributed/test/large_scale_test.cc

@@ -21,9 +21,9 @@ limitations under the License. */
 #include "google/protobuf/text_format.h"
 #include "gtest/gtest.h"
 #include "paddle/fluid/distributed/ps.pb.h"
-#include "paddle/fluid/distributed/table/common_sparse_table.h"
-#include "paddle/fluid/distributed/table/depends/large_scale_kv.h"
-#include "paddle/fluid/distributed/table/table.h"
+#include "paddle/fluid/distributed/ps/table/common_sparse_table.h"
+#include "paddle/fluid/distributed/ps/table/depends/large_scale_kv.h"
+#include "paddle/fluid/distributed/ps/table/table.h"
 
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/test/memory_sparse_table_test.cc

@@ -21,8 +21,8 @@ limitations under the License. */
 #include "google/protobuf/text_format.h"
 #include "gtest/gtest.h"
 #include "paddle/fluid/distributed/ps.pb.h"
-#include "paddle/fluid/distributed/table/memory_sparse_table.h"
-#include "paddle/fluid/distributed/table/table.h"
+#include "paddle/fluid/distributed/ps/table/memory_sparse_table.h"
+#include "paddle/fluid/distributed/ps/table/table.h"
 
 namespace paddle {
 namespace distributed {

paddle/fluid/distributed/test/sparse_sgd_rule_test.cc

@@ -12,7 +12,7 @@ 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 "paddle/fluid/distributed/table/sparse_sgd_rule.h"
+#include "paddle/fluid/distributed/ps/table/sparse_sgd_rule.h"
 #include <cmath>
 #include <iostream>
 #include "gtest/gtest.h"

paddle/fluid/distributed/test/sparse_table_test.cc

@@ -21,10 +21,10 @@ limitations under the License. */
 #include "google/protobuf/text_format.h"
 #include "gtest/gtest.h"
 #include "paddle/fluid/distributed/ps.pb.h"
-#include "paddle/fluid/distributed/table/common_dense_table.h"
-#include "paddle/fluid/distributed/table/common_sparse_table.h"
-#include "paddle/fluid/distributed/table/sparse_geo_table.h"
-#include "paddle/fluid/distributed/table/table.h"
+#include "paddle/fluid/distributed/ps/table/common_dense_table.h"
+#include "paddle/fluid/distributed/ps/table/common_sparse_table.h"
+#include "paddle/fluid/distributed/ps/table/sparse_geo_table.h"
+#include "paddle/fluid/distributed/ps/table/table.h"
 
 namespace paddle {
 namespace distributed {