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

benchmark/fluid/mnist.py

@@ -159,6 +159,7 @@ def run_benchmark(model, args):
         paddle.dataset.mnist.train(), batch_size=args.batch_size)
 
     accuracy = fluid.metrics.Accuracy()
+    train_exe = fluid.ParallelExecutor(use_cuda=True, loss_name=avg_cost.name)
     iters, num_samples, start_time = 0, 0, time.time()
     for pass_id in range(args.pass_num):
         accuracy.reset()
@@ -175,17 +176,20 @@ def run_benchmark(model, args):
             y_data = np.array(map(lambda x: x[1], data)).astype("int64")
             y_data = y_data.reshape([len(y_data), 1])
 
-            outs = exe.run(
-                fluid.default_main_program(),
+            outs = train_exe.run(
                 feed={"pixel": img_data,
                       "label": y_data},
-                fetch_list=[avg_cost, batch_acc, batch_size_tensor]
+                fetch_list=[
+                    avg_cost.name, batch_acc.name, batch_size_tensor.name
+                ]
             )  # The accuracy is the accumulation of batches, but not the current batch.
-            accuracy.update(value=outs[1], weight=outs[2])
+            accuracy.update(
+                value=np.array(np.mean(outs[1])),
+                weight=np.mean(np.array(outs[2])))
             iters += 1
             num_samples += len(y_data)
-            loss = np.array(outs[0])
-            acc = np.array(outs[1])
+            loss = np.mean(np.array(outs[0]))
+            acc = np.mean(np.array(outs[1]))
             train_losses.append(loss)
             train_accs.append(acc)
             print("Pass: %d, Iter: %d, Loss: %f, Accuracy: %f" %

benchmark/fluid/resnet.py

@@ -241,6 +241,7 @@ def run_benchmark(model, args):
     exe = fluid.Executor(place)
     exe.run(fluid.default_startup_program())
     accuracy = fluid.average.WeightedAverage()
+    train_exe = fluid.ParallelExecutor(use_cuda=True, loss_name=avg_cost.name)
     if args.use_fake_data:
         data = train_reader().next()
         image = np.array(map(lambda x: x[0].reshape(dshape), data)).astype(
@@ -264,14 +265,17 @@ def run_benchmark(model, args):
                                      data)).astype('float32')
                 label = np.array(map(lambda x: x[1], data)).astype('int64')
                 label = label.reshape([-1, 1])
-            loss, acc, weight = exe.run(
-                fluid.default_main_program(),
+            loss, acc, weight = train_exe.run(
                 feed={'data': image,
                       'label': label},
-                fetch_list=[avg_cost, batch_acc, batch_size_tensor])
+                fetch_list=[
+                    avg_cost.name, batch_acc.name, batch_size_tensor.name
+                ])
             iters += 1
             num_samples += len(label)
-            accuracy.add(value=acc, weight=weight)
+            accuracy.add(value=np.array(np.mean(acc)), weight=np.mean(weight))
+            loss = np.mean(np.array(loss))
+            acc = np.mean(np.array(acc))
             train_losses.append(loss)
             train_accs.append(acc)
             print("Pass: %d, Iter: %d, Loss: %f, Accuracy: %f" %

benchmark/fluid/vgg.py

@@ -169,6 +169,7 @@ def main():
 
     iters, num_samples, start_time = 0, 0, time.time()
     accuracy = fluid.average.WeightedAverage()
+    train_exe = fluid.ParallelExecutor(use_cuda=True, loss_name=avg_cost.name)
     for pass_id in range(args.pass_num):
         accuracy.reset()
         train_accs = []
@@ -184,14 +185,17 @@ def main():
             y_data = np.array(map(lambda x: x[1], data)).astype("int64")
             y_data = y_data.reshape([-1, 1])
 
-            loss, acc, weight = exe.run(
-                fluid.default_main_program(),
+            loss, acc, weight = train_exe.run(
                 feed={"pixel": img_data,
                       "label": y_data},
-                fetch_list=[avg_cost, batch_acc, batch_size_tensor])
-            accuracy.add(value=acc, weight=weight)
+                fetch_list=[
+                    avg_cost.name, batch_acc.name, batch_size_tensor.name
+                ])
+            accuracy.add(value=np.array(np.mean(acc)), weight=np.mean(weight))
             iters += 1
             num_samples += len(y_data)
+            loss = np.mean(np.array(loss))
+            acc = np.mean(np.array(acc))
             print(
                 "Pass = %d, Iter = %d, Loss = %f, Accuracy = %f" %
                 (pass_id, iters, loss, acc)

cmake/external/eigen.cmake

@@ -21,11 +21,12 @@ else()
     ExternalProject_Add(
         extern_eigen3
         ${EXTERNAL_PROJECT_LOG_ARGS}
-        GIT_REPOSITORY  "https://github.com/RLovelett/eigen.git"
+        GIT_REPOSITORY  "https://github.com/eigenteam/eigen-git-mirror"
         # eigen on cuda9.1 missing header of math_funtions.hpp
         # https://stackoverflow.com/questions/43113508/math-functions-hpp-not-found-when-using-cuda-with-eigen
         GIT_TAG         917060c364181f33a735dc023818d5a54f60e54c
         PREFIX          ${EIGEN_SOURCE_DIR}
+        DOWNLOAD_NAME   "eigen"
         UPDATE_COMMAND  ""
         CONFIGURE_COMMAND ""
         BUILD_COMMAND     ""

cmake/external/snappy.cmake

@@ -47,8 +47,6 @@ ExternalProject_Add(
                      -DCMAKE_INSTALL_LIBDIR:PATH=${SNAPPY_INSTALL_DIR}/lib
                      -DCMAKE_POSITION_INDEPENDENT_CODE:BOOL=ON
                      -DCMAKE_BUILD_TYPE:STRING=${THIRD_PARTY_BUILD_TYPE}
-    BUILD_COMMAND   make -j8
-    INSTALL_COMMAND make install
 )
 
 add_library(snappy STATIC IMPORTED GLOBAL)

cmake/external/snappystream.cmake

@@ -46,8 +46,6 @@ ExternalProject_Add(
                         -DCMAKE_INSTALL_PREFIX:PATH=${SNAPPYSTREAM_INSTALL_DIR}
                         -DCMAKE_INSTALL_LIBDIR:PATH=${SNAPPYSTREAM_INSTALL_DIR}/lib
                         -DCMAKE_BUILD_TYPE:STRING=${THIRD_PARTY_BUILD_TYPE}
-        BUILD_COMMAND   make -j8
-        INSTALL_COMMAND make install
         DEPENDS snappy
 )
 

cmake/inference_lib.cmake

@@ -70,6 +70,12 @@ copy(glog_lib
   DSTS ${dst_dir} ${dst_dir}/lib
 )
 
+set(dst_dir "${CMAKE_INSTALL_PREFIX}/third_party/boost/")
+copy(boost_lib
+  SRCS ${BOOST_INCLUDE_DIR}/boost
+  DSTS ${dst_dir}
+)
+
 if(NOT PROTOBUF_FOUND)
     set(dst_dir "${CMAKE_INSTALL_PREFIX}/third_party/install/protobuf")
     copy(protobuf_lib
@@ -92,6 +98,14 @@ elseif (WITH_MKLML)
     )
 endif()
 
+if(WITH_MKLDNN)
+  set(dst_dir "${CMAKE_INSTALL_PREFIX}/third_party/install/mkldnn")
+  copy(mkldnn_lib
+    SRCS ${MKLDNN_INC_DIR} ${MKLDNN_SHARED_LIB}
+    DSTS ${dst_dir} ${dst_dir}/lib
+  )
+endif()
+
 if(NOT MOBILE_INFERENCE AND NOT RPI)
   set(dst_dir "${CMAKE_INSTALL_PREFIX}/third_party/install/snappy")
   copy(snappy_lib
@@ -142,4 +156,10 @@ copy(string_lib
   DSTS ${dst_dir}/${module} ${dst_dir}/${module}/tinyformat
 )
 
+set(module "pybind")
+copy(pybind_lib
+  SRCS ${CMAKE_CURRENT_BINARY_DIR}/paddle/fluid/${module}/pybind.h
+  DSTS ${dst_dir}/${module}
+)
+
 add_custom_target(inference_lib_dist DEPENDS ${inference_lib_dist_dep}) 

doc/fluid/design/concepts/functions_operators_layers.md

@@ -40,7 +40,7 @@ template <typename T>
 class FCOp : public OperatorBase {
  public:
   void Run(...) {
-    add(mul(Input<T>("X"), Input<T>("W")), Input<T>("b");
+    add(mul(Input<T>("X"), Input<T>("W")), Input<T>("b"));
   }
 };
 REGISTER_OP(FCOp, "fc");

doc/fluid/design/dist_train/async_update.md

@@ -4,34 +4,37 @@
 
 For the typical synchronous distributed training, some significant steps are as follows:
 
-1. A Trainer will compute the gradients and SEND them to the Parameter Server(PServer) nodes.
-1. After the PServer node received gradients came from all the Trainers, It will aggregate the
+1. A trainer process will compute the gradients and **send** them to the parameter server (PS) nodes.
+1. After the PS node received gradients came from all the Trainers, It will aggregate the
 gradient variables for the same parameter into one gradient variable and then apply the aggregated
 gradient to the respective parameter, finally using an optimize algorithms(SGD, Monument...)
 to update the parameters.
-1. The Trainer would wait for the PServers finished the optimize stage, and GET the parameters from PServer,
+1. The Trainer would wait for the PS finished the optimize stage, and GET the parameters from PS,
 so all the Trainers would get the same parameters.
 
-In the synchronously distributed training, there should be a `Barrier` to synchronise the
-parameters after the optimizing stage. The performance of a distributed training job would
-depend on the slowest node if there were hundreds or thousands of training nodes in a
-Job, the performance of synchronously distributed training might be very poor because of
-the slow node. So this design doc would introduce an approach to implement
-*asynchronously* distributed training in PaddlePaddle Fluid.
+In Synchronous Distributed Training, there is a **barrier** on each PS to wait until all trainers processes
+have completed running current mini-batch. After that, all trainers can continue to run the next
+mini-batch. So, we can find that the overall performance of Synchronous Distributed Training depends 
+on the slowest node.
+
+In Asynchronous Distributed Training, we don't need to wait for a global mini-bach, the optimizer on
+the PS will run immediately when the gradient is uploaded to the PS from one trainer. This mode would
+train such models that achieve scaling, better throughput. In this design doc, we will introduce how to 
+implement the Asynchronous Distributed Training base on PaddlePaddle Fluid.
 
 ## Design
 
 <img src="./src/async_update.png" width="600"/>
 
-As the figure above, we describe a global view of asynchronously update process and use
+As the figure above, we describe a global view of the asynchronous update process and use
 the parameter `w1` as an example to introduce the steps:
 1. For each gradient variables, they may distribute on different GPU card and aggregate
 them while they are all calculated.
-1. Split the gradient variable into multiple blocks according to the number of PServer
+1. Split the gradient variable into multiple blocks according to the number of PS
 instances and then send them.
-1. PServer would run an `Optimize Block` using a specified optimize algorithm to update
+1. PS would run an `Optimize Block` using a specified optimize algorithm to update
 the specified parameter.
-1. The trainer will fetch latest parameter from PServer before running forward Op which depends
+1. The trainer will fetch the latest parameter from PS before running forward Op which depends
 on the specified parameter.
 1. Broadcast the received variable into multiple GPU cards and continue to run the next
 mini-batch.
@@ -40,8 +43,8 @@ mini-batch.
 
 - For the multiple devices distributed training, we need to aggregate the gradient
 variables which placed on different devices firstly and then schedule a `SendVars` Operator to
-send the gradient variables to the multiple PServer instances.
-- Schedule `FetchVars` operator to fetch the latest parameter from PServer before running
+send the gradient variables to the multiple PS instances.
+- Schedule `FetchVars` operator to fetch the latest parameter from PS before running
 the forward ops.
 - There could be a large number of gradient variables to be sent, so we need to use another
 thread pool(IO Threadpool) whose a number of the schedulable threads is larger than the

paddle/CMakeLists.txt

@@ -24,6 +24,6 @@ if(NOT WITH_FLUID_ONLY)
 endif()
 
 add_subdirectory(testing)
-if(NOT MOBILE_INFERENCE AND NOT RPI)
+if(NOT MOBILE_INFERENCE AND NOT RPI AND NOT WITH_C_API)
   add_subdirectory(fluid)
 endif()

paddle/fluid/framework/CMakeLists.txt

@@ -5,11 +5,11 @@ proto_library(framework_proto SRCS framework.proto)
 cc_library(ddim SRCS ddim.cc DEPS eigen3 boost)
 cc_test(ddim_test SRCS ddim_test.cc DEPS ddim)
 nv_test(dim_test SRCS dim_test.cu DEPS ddim)
-
+cc_library(data_type SRCS data_type.cc DEPS framework_proto ddim device_context)
 if(WITH_GPU)
-  nv_library(tensor SRCS tensor.cc tensor_util.cu DEPS ddim place memory device_context framework_proto)
+  nv_library(tensor SRCS tensor.cc tensor_util.cu DEPS place memory data_type)
 else()
-  cc_library(tensor SRCS tensor.cc tensor_util.cc DEPS ddim place memory device_context framework_proto)
+  cc_library(tensor SRCS tensor.cc tensor_util.cc DEPS place memory data_type)
 endif()
 
 cc_test(tensor_test SRCS tensor_test.cc DEPS tensor)

paddle/fluid/framework/data_type.cc

+//   Copyright (c) 2018 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 "paddle/fluid/framework/data_type.h"
+#include <stdint.h>
+#include <string>
+#include <unordered_map>
+
+namespace paddle {
+namespace framework {
+
+struct DataTypeMap {
+  std::unordered_map<std::type_index, proto::VarType::Type> cpp_to_proto_;
+  std::unordered_map<int, std::type_index> proto_to_cpp_;
+  std::unordered_map<int, std::string> proto_to_str_;
+  std::unordered_map<std::type_index, size_t> cpp_to_size_;
+};
+
+static DataTypeMap* InitDataTypeMap();
+static DataTypeMap& gDataTypeMap() {
+  static DataTypeMap* g_data_type_map_ = InitDataTypeMap();
+  return *g_data_type_map_;
+}
+
+template <typename T>
+static inline void RegisterType(DataTypeMap* map,
+                                proto::VarType::Type proto_type,
+                                const std::string& name) {
+  map->proto_to_cpp_.emplace(static_cast<int>(proto_type), typeid(T));
+  map->cpp_to_proto_.emplace(typeid(T), proto_type);
+  map->proto_to_str_.emplace(static_cast<int>(proto_type), name);
+  map->cpp_to_size_.emplace(typeid(T), sizeof(T));
+}
+
+static DataTypeMap* InitDataTypeMap() {
+  auto retv = new DataTypeMap();
+
+#define RegType(cc_type, proto_type) \
+  RegisterType<cc_type>(retv, proto_type, #cc_type)
+
+  // NOTE: Add your customize type here.
+  RegType(platform::float16, proto::VarType::FP16);
+  RegType(float, proto::VarType::FP32);
+  RegType(double, proto::VarType::FP64);
+  RegType(int, proto::VarType::INT32);
+  RegType(int64_t, proto::VarType::INT64);
+  RegType(bool, proto::VarType::BOOL);
+  RegType(size_t, proto::VarType::SIZE_T);
+  RegType(int16_t, proto::VarType::INT16);
+  RegType(uint8_t, proto::VarType::UINT8);
+
+#undef RegType
+  return retv;
+}
+
+proto::VarType::Type ToDataType(std::type_index type) {
+  auto it = gDataTypeMap().cpp_to_proto_.find(type);
+  if (it != gDataTypeMap().cpp_to_proto_.end()) {
+    return it->second;
+  }
+  PADDLE_THROW("Not support %s as tensor type", type.name());
+}
+
+std::type_index ToTypeIndex(proto::VarType::Type type) {
+  auto it = gDataTypeMap().proto_to_cpp_.find(static_cast<int>(type));
+  if (it != gDataTypeMap().proto_to_cpp_.end()) {
+    return it->second;
+  }
+  PADDLE_THROW("Not support proto::VarType::Type(%d) as tensor type",
+               static_cast<int>(type));
+}
+
+std::string DataTypeToString(const proto::VarType::Type type) {
+  auto it = gDataTypeMap().proto_to_str_.find(static_cast<int>(type));
+  if (it != gDataTypeMap().proto_to_str_.end()) {
+    return it->second;
+  }
+  PADDLE_THROW("Not support proto::VarType::Type(%d) as tensor type",
+               static_cast<int>(type));
+}
+
+size_t SizeOfType(std::type_index type) {
+  auto it = gDataTypeMap().cpp_to_size_.find(type);
+  if (it != gDataTypeMap().cpp_to_size_.end()) {
+    return it->second;
+  }
+  PADDLE_THROW("Not support %s as tensor type", type.name());
+}
+
+}  // namespace framework
+}  // namespace paddle

paddle/fluid/framework/data_type.h

@@ -17,51 +17,14 @@ limitations under the License. */
 #include <typeindex>
 #include "paddle/fluid/framework/framework.pb.h"
 #include "paddle/fluid/platform/enforce.h"
+
 #include "paddle/fluid/platform/float16.h"
 
 namespace paddle {
 namespace framework {
 
-inline proto::VarType::Type ToDataType(std::type_index type) {
-  if (typeid(platform::float16).hash_code() == type.hash_code()) {
-    return proto::VarType::FP16;
-  } else if (typeid(const float).hash_code() == type.hash_code()) {
-    // CPPLint complains Using C-style cast.  Use static_cast<float>() instead
-    // One fix to this is to replace float with const float because
-    // typeid(T) == typeid(const T)
-    // http://en.cppreference.com/w/cpp/language/typeid
-    return proto::VarType::FP32;
-  } else if (typeid(const double).hash_code() == type.hash_code()) {
-    return proto::VarType::FP64;
-  } else if (typeid(const int).hash_code() == type.hash_code()) {
-    return proto::VarType::INT32;
-  } else if (typeid(const int64_t).hash_code() == type.hash_code()) {
-    return proto::VarType::INT64;
-  } else if (typeid(const bool).hash_code() == type.hash_code()) {
-    return proto::VarType::BOOL;
-  } else {
-    PADDLE_THROW("Not supported");
-  }
-}
-
-inline std::type_index ToTypeIndex(proto::VarType::Type type) {
-  switch (type) {
-    case proto::VarType::FP16:
-      return typeid(platform::float16);
-    case proto::VarType::FP32:
-      return typeid(float);
-    case proto::VarType::FP64:
-      return typeid(double);
-    case proto::VarType::INT32:
-      return typeid(int);
-    case proto::VarType::INT64:
-      return typeid(int64_t);
-    case proto::VarType::BOOL:
-      return typeid(bool);
-    default:
-      PADDLE_THROW("Not support type %d", type);
-  }
-}
+extern proto::VarType::Type ToDataType(std::type_index type);
+extern std::type_index ToTypeIndex(proto::VarType::Type type);
 
 template <typename Visitor>
 inline void VisitDataType(proto::VarType::Type type, Visitor visitor) {
@@ -84,37 +47,23 @@ inline void VisitDataType(proto::VarType::Type type, Visitor visitor) {
     case proto::VarType::BOOL:
       visitor.template operator()<bool>();
       break;
-    default:
-      PADDLE_THROW("Not supported");
-  }
-}
-
-inline std::string DataTypeToString(const proto::VarType::Type type) {
-  switch (type) {
-    case proto::VarType::FP16:
-      return "float16";
-    case proto::VarType::FP32:
-      return "float32";
-    case proto::VarType::FP64:
-      return "float64";
+    case proto::VarType::UINT8:
+      visitor.template operator()<uint8_t>();
+      break;
     case proto::VarType::INT16:
-      return "int16";
-    case proto::VarType::INT32:
-      return "int32";
-    case proto::VarType::INT64:
-      return "int64";
-    case proto::VarType::BOOL:
-      return "bool";
+      visitor.template operator()<int16_t>();
+      break;
     default:
-      PADDLE_THROW("Not support type %d", type);
+      PADDLE_THROW("Not supported %d", type);
   }
 }
 
+extern std::string DataTypeToString(const proto::VarType::Type type);
+extern size_t SizeOfType(std::type_index type);
 inline std::ostream& operator<<(std::ostream& out,
                                 const proto::VarType::Type& type) {
   out << DataTypeToString(type);
   return out;
 }
-
 }  // namespace framework
 }  // namespace paddle

paddle/fluid/framework/details/build_strategy.h

+// Copyright (c) 2018 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.
+
+#pragma once
+
+namespace paddle {
+namespace framework {
+namespace details {
+
+struct BuildStrategy {
+  enum class ReduceStrategy { kAllReduce = 0, kReduce = 1 };
+
+  enum class GradientScaleStrategy {
+    kCoeffNumDevice = 0,
+    kOne = 1,
+    kCustomized = 2,
+  };
+
+  ReduceStrategy reduce_{ReduceStrategy::kAllReduce};
+  GradientScaleStrategy gradient_scale_{GradientScaleStrategy::kCoeffNumDevice};
+};
+
+}  // namespace details
+}  // namespace framework
+}  // namespace paddle

paddle/fluid/framework/details/execution_strategy.h

+// Copyright (c) 2018 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.
+
+#pragma once
+
+namespace paddle {
+namespace framework {
+namespace details {
+
+struct ExecutionStrategy {
+  size_t num_threads_{0};
+  bool use_event_{true};
+  bool allow_op_delay_{false};
+};
+
+}  //  namespace details
+}  //  namespace framework
+}  //  namespace paddle

paddle/fluid/framework/details/fetch_op_handle.cc

@@ -48,17 +48,18 @@ void FetchOpHandle::RunImpl() {
   WaitInputVarGenerated(platform::CPUPlace());
 
   tensors_.resize(inputs_.size());
-  auto *var_handle = static_cast<VarHandle *>(inputs_[0]);
-  auto &var_name = var_handle->name_;
   platform::CPUPlace cpu;
   auto &scopes = *local_scopes_;
 
-  for (size_t i = 0; i < scopes.size(); ++i) {
-    auto &scope = scopes[i];
-    auto *var =
-        scope->FindVar(kLocalExecScopeName)->Get<Scope *>()->FindVar(var_name);
+  for (size_t i = 0; i < inputs_.size(); ++i) {
+    auto *var_handle = static_cast<VarHandle *>(inputs_[i]);
+    auto &scope = scopes.at(var_handle->scope_idx_);
+    auto *var = scope->FindVar(kLocalExecScopeName)
+                    ->Get<Scope *>()
+                    ->FindVar(var_handle->name_);
     PADDLE_ENFORCE_NOT_NULL(var, "Cannot find variable %s in execution scope",
-                            var_name);
+                            var_handle->name_);
+
     auto &t = var->Get<framework::LoDTensor>();
     if (platform::is_gpu_place(t.place())) {
 #ifdef PADDLE_WITH_CUDA

paddle/fluid/framework/details/multi_devices_graph_builder.cc

@@ -37,31 +37,26 @@ MultiDevSSAGraphBuilder::MultiDevSSAGraphBuilder(
     const std::string &loss_var_name,
     const std::unordered_set<std::string> &params,
     const std::vector<Scope *> &local_scopes,
-    platform::NCCLContextMap *nccl_ctxs, bool use_default_grad_scale,
-    bool balance_parameter_opt_between_cards)
+    platform::NCCLContextMap *nccl_ctxs, const BuildStrategy &strategy)
     : loss_var_name_(loss_var_name),
       places_(places),
       local_scopes_(local_scopes),
       nccl_ctxs_(nccl_ctxs),
-      balance_parameter_opt_between_cards_(
-          balance_parameter_opt_between_cards) {
+      strategy_(strategy) {
 #else
 MultiDevSSAGraphBuilder::MultiDevSSAGraphBuilder(
     const std::vector<platform::Place> &places,
     const std::string &loss_var_name,
     const std::unordered_set<std::string> &params,
-    const std::vector<Scope *> &local_scopes, bool use_default_grad_scale,
-    bool balance_parameter_opt_between_cards)
+    const std::vector<Scope *> &local_scopes, const BuildStrategy &strategy)
     : loss_var_name_(loss_var_name),
       places_(places),
       local_scopes_(local_scopes),
-      balance_parameter_opt_between_cards_(
-          balance_parameter_opt_between_cards) {
+      strategy_(strategy) {
 #endif
   for (auto &p : params) {
     grad_names_.insert(GradVarName(p));
   }
-  use_default_grad_scale_ = use_default_grad_scale;
 }
 
 void MultiDevSSAGraphBuilder::CreateOpHandleIOs(SSAGraph *result,
@@ -146,7 +141,8 @@ std::unique_ptr<SSAGraph> MultiDevSSAGraphBuilder::Build(
       CreateComputationalOps(&result, *op, 1);
     } else if (IsScaleLossOp(*op)) {
       // user can customize loss@grad if not use_default_grad_scale_
-      if (use_default_grad_scale_) {
+      if (strategy_.gradient_scale_ !=
+          BuildStrategy::GradientScaleStrategy::kCustomized) {
         CreateScaleLossGradOp(&result);
       }
       is_forwarding = false;
@@ -165,19 +161,22 @@ std::unique_ptr<SSAGraph> MultiDevSSAGraphBuilder::Build(
         // broadcast, and each gradient is only broadcast once.
         for (auto &og : op->OutputArgumentNames()) {
           if (IsParameterGradientOnce(og, &og_has_been_broadcast)) {
-            if (balance_parameter_opt_between_cards_) {
-              CreateReduceOp(&result, og, cur_device_id);
-              var_name_on_devices[cur_device_id].emplace(og);
-              bcast_var_name_set[cur_device_id].emplace(
-                  og.substr(0, og.size() - strlen(kGradVarSuffix)));
-              cur_device_id = (cur_device_id + 1) % places_.size();
-            } else {
-              if (IsSparseGradient(var_types, og)) {
-                CreateReduceOp(&result, og, 0);
-                CreateBroadcastOp(&result, og, 0);
-              } else {
-                InsertNCCLAllReduceOp(&result, og);
-              }
+            switch (strategy_.reduce_) {
+              case BuildStrategy::ReduceStrategy::kReduce:
+                CreateReduceOp(&result, og, cur_device_id);
+                var_name_on_devices[cur_device_id].emplace(og);
+                bcast_var_name_set[cur_device_id].emplace(
+                    og.substr(0, og.size() - strlen(kGradVarSuffix)));
+                cur_device_id = (cur_device_id + 1) % places_.size();
+                break;
+              case BuildStrategy::ReduceStrategy::kAllReduce:
+                if (IsSparseGradient(var_types, og)) {
+                  CreateReduceOp(&result, og, 0);
+                  CreateBroadcastOp(&result, og, 0);
+                } else {
+                  InsertNCCLAllReduceOp(&result, og);
+                }
+                break;
             }
           }
         }
@@ -303,7 +302,7 @@ bool MultiDevSSAGraphBuilder::IsParameterGradientOnce(
 int MultiDevSSAGraphBuilder::GetOpDeviceID(
     const std::vector<std::unordered_set<std::string>> &var_name_on_devices,
     const OpDesc &op) const {
-  if (!balance_parameter_opt_between_cards_) {
+  if (strategy_.reduce_ != BuildStrategy::ReduceStrategy::kReduce) {
     return -1;
   }
 

paddle/fluid/framework/details/multi_devices_graph_builder.h

@@ -17,6 +17,7 @@
 #include <utility>
 #include <vector>
 
+#include "paddle/fluid/framework/details/build_strategy.h"
 #include "paddle/fluid/framework/details/ssa_graph_builder.h"
 
 namespace paddle {
@@ -36,15 +37,13 @@ class MultiDevSSAGraphBuilder : public SSAGraphBuilder {
                           const std::unordered_set<std::string> &params,
                           const std::vector<Scope *> &local_scopes,
                           platform::NCCLContextMap *nccl_ctxs,
-                          bool use_default_grad_scale,
-                          bool balance_parameter_opt_between_cards);
+                          const BuildStrategy &strategy);
 #else
   MultiDevSSAGraphBuilder(const std::vector<platform::Place> &places,
                           const std::string &loss_var_name,
                           const std::unordered_set<std::string> &params,
                           const std::vector<Scope *> &local_scopes,
-                          bool use_default_grad_scale,
-                          bool balance_parameter_opt_between_cards);
+                          const BuildStrategy &strategy);
 #endif
 
   std::unique_ptr<SSAGraph> Build(const ProgramDesc &program) const override;
@@ -62,8 +61,6 @@ class MultiDevSSAGraphBuilder : public SSAGraphBuilder {
 #ifdef PADDLE_WITH_CUDA
   platform::NCCLContextMap *nccl_ctxs_;
 #endif
-  bool balance_parameter_opt_between_cards_;
-  bool use_default_grad_scale_;
 
   bool IsScaleLossOp(const OpDesc &op) const;
 
@@ -105,6 +102,9 @@ class MultiDevSSAGraphBuilder : public SSAGraphBuilder {
   bool IsSparseGradient(
       const std::unordered_map<std::string, proto::VarType::Type> &var_types,
       const std::string &og) const;
+
+ private:
+  BuildStrategy strategy_;
 };
 }  // namespace details
 }  // namespace framework

paddle/fluid/framework/details/op_handle_base.h

@@ -70,6 +70,14 @@ class OpHandleBase {
 
   const std::vector<VarHandleBase *> &Inputs() const { return inputs_; }
 
+  size_t NoDupInputSize() const {
+    std::unordered_set<VarHandleBase *> res;
+    for (auto *var : inputs_) {
+      res.emplace(var);
+    }
+    return res.size();
+  }
+
   const std::vector<VarHandleBase *> &Outputs() const { return outputs_; }
 
  protected:

paddle/fluid/framework/details/threaded_ssa_graph_executor.cc

@@ -18,18 +18,17 @@ namespace paddle {
 namespace framework {
 namespace details {
 ThreadedSSAGraphExecutor::ThreadedSSAGraphExecutor(
-    size_t num_threads, bool use_event,
-    const std::vector<Scope *> &local_scopes,
+    const ExecutionStrategy &strategy, const std::vector<Scope *> &local_scopes,
     const std::vector<platform::Place> &places,
-    std::unique_ptr<SSAGraph> &&graph, bool allow_op_delay)
+    std::unique_ptr<SSAGraph> &&graph)
     : SSAGraphExecutor(std::move(graph)),
-      pool_(num_threads >= 2 ? new ::ThreadPool(num_threads) : nullptr),
+      pool_(strategy.num_threads_ >= 2 ? new ::ThreadPool(strategy.num_threads_)
+                                       : nullptr),
       local_scopes_(local_scopes),
       places_(places),
       fetch_ctxs_(places),
-      use_event_(use_event),
       running_ops_(0),
-      allow_op_delay_(allow_op_delay) {}
+      strategy_(strategy) {}
 
 FeedFetchList ThreadedSSAGraphExecutor::Run(
     const std::vector<std::string> &fetch_tensors) {
@@ -86,7 +85,7 @@ FeedFetchList ThreadedSSAGraphExecutor::Run(
     //
     // NOTE: DelayedOps have a lower priority. It will be scheduled after all
     // ready_ops have been performed.
-    if (ready_ops.empty() && allow_op_delay_ && running_ops_ == 0) {
+    if (ready_ops.empty() && strategy_.allow_op_delay_ && running_ops_ == 0) {
       run_all_ops(delayed_ops);
     } else {
       run_all_ops(ready_ops);
@@ -113,7 +112,7 @@ FeedFetchList ThreadedSSAGraphExecutor::Run(
         auto &deps = pending_ops[op];
         --deps;
         if (deps == 0) {
-          if (op->IsMultiDeviceTransfer() && allow_op_delay_) {
+          if (op->IsMultiDeviceTransfer() && strategy_.allow_op_delay_) {
             delayed_ops.insert(op);
           } else {
             ready_ops.insert(op);
@@ -175,7 +174,7 @@ void ThreadedSSAGraphExecutor::InsertFetchOps(
 void ThreadedSSAGraphExecutor::InsertPendingOp(
     std::unordered_map<OpHandleBase *, size_t> *pending_ops,
     OpHandleBase *op_instance) const {
-  pending_ops->insert({op_instance, op_instance->Inputs().size()});
+  pending_ops->insert({op_instance, op_instance->NoDupInputSize()});
 }
 
 void ThreadedSSAGraphExecutor::InsertPendingVar(
@@ -191,7 +190,7 @@ void ThreadedSSAGraphExecutor::RunOp(
   auto op_run = [ready_var_q, op, this] {
     try {
       VLOG(10) << op << " " << op->Name() << " : " << op->DebugString();
-      op->Run(use_event_);
+      op->Run(strategy_.use_event_);
       VLOG(10) << op << " " << op->Name() << " Done ";
       running_ops_--;
       ready_var_q->Extend(op->Outputs());

paddle/fluid/framework/details/threaded_ssa_graph_executor.h

@@ -23,6 +23,7 @@
 #include <functional>
 #include "ThreadPool.h"  // ThreadPool in thrird party
 #include "paddle/fluid/framework/blocking_queue.h"
+#include "paddle/fluid/framework/details/execution_strategy.h"
 #include "paddle/fluid/framework/details/fetch_op_handle.h"
 #include "paddle/fluid/framework/details/ssa_graph_executor.h"
 
@@ -34,11 +35,10 @@ namespace details {
 
 class ThreadedSSAGraphExecutor : public SSAGraphExecutor {
  public:
-  ThreadedSSAGraphExecutor(size_t num_threads, bool use_event,
+  ThreadedSSAGraphExecutor(const ExecutionStrategy &strategy,
                            const std::vector<Scope *> &local_scopes,
                            const std::vector<platform::Place> &places,
-                           std::unique_ptr<SSAGraph> &&graph,
-                           bool allow_op_delay);
+                           std::unique_ptr<SSAGraph> &&graph);
 
   // Run a SSAGraph by a thread pool
   // Use topological sort algorithm
@@ -55,10 +55,8 @@ class ThreadedSSAGraphExecutor : public SSAGraphExecutor {
   std::vector<Scope *> local_scopes_;
   std::vector<platform::Place> places_;
   platform::DeviceContextPool fetch_ctxs_;
-  const bool use_event_;
   std::unique_ptr<platform::EnforceNotMet> exception_;
   std::atomic<int> running_ops_;
-  bool allow_op_delay_;
 
   void InsertPendingOp(std::unordered_map<OpHandleBase *, size_t> *pending_ops,
                        OpHandleBase *op_instance) const;
@@ -74,6 +72,9 @@ class ThreadedSSAGraphExecutor : public SSAGraphExecutor {
       std::unordered_map<OpHandleBase *, size_t> *pending_ops,
       std::unordered_set<VarHandleBase *> *pending_vars,
       BlockingQueue<VarHandleBase *> *ready_vars, FeedFetchList *fetch_data);
+
+ private:
+  ExecutionStrategy strategy_;
 };
 
 }  // namespace details

paddle/fluid/framework/framework.proto

@@ -101,6 +101,9 @@ message VarType {
     FP16 = 4;
     FP32 = 5;
     FP64 = 6;
+    // Tensor<size_t> is used in C++.
+    SIZE_T = 19;
+    UINT8 = 20;
 
     // Other types that may need additional descriptions
     LOD_TENSOR = 7;

paddle/fluid/framework/lod_tensor_test.cc

@@ -228,11 +228,12 @@ TEST(LoD, CheckAbsLoD) {
   ASSERT_FALSE(CheckAbsLoD(abs_lod0));
 }
 
-TEST(LoDTensor, RecordIO) {
+template <typename T>
+static void TestRecordIO() {
   LoDTensor tensor;
-  int* tmp = tensor.mutable_data<int>(make_ddim({4, 5}), platform::CPUPlace());
+  T* tmp = tensor.mutable_data<T>(make_ddim({4, 5}), platform::CPUPlace());
   for (int i = 0; i < 20; ++i) {
-    tmp[i] = i;
+    tmp[i] = static_cast<T>(i);
   }
 
   std::stringstream* stream = new std::stringstream();
@@ -247,7 +248,7 @@ TEST(LoDTensor, RecordIO) {
 
   auto assert_tensor_ok = [](const LoDTensor& tensor) {
     for (int i = 0; i < 20; ++i) {
-      ASSERT_EQ(tensor.data<int>()[i], i);
+      ASSERT_EQ(tensor.data<T>()[i], static_cast<T>(i));
     }
   };
 
@@ -265,5 +266,13 @@ TEST(LoDTensor, RecordIO) {
   }
 }
 
+TEST(LoDTensor, RecordIO) {
+  TestRecordIO<int>();
+  TestRecordIO<int16_t>();
+  TestRecordIO<uint8_t>();
+  TestRecordIO<float>();
+  TestRecordIO<double>();
+}
+
 }  // namespace framework
 }  // namespace paddle

paddle/fluid/framework/op_kernel_type_test.cc

@@ -27,7 +27,7 @@ TEST(OpKernelType, ToString) {
                               LibraryType::kCUDNN);
 
   ASSERT_EQ(paddle::framework::KernelTypeToString(op_kernel_type),
-            "data_type[float32]:data_layout[NCHW]:place[CPUPlace]:library_type["
+            "data_type[float]:data_layout[NCHW]:place[CPUPlace]:library_type["
             "CUDNN]");
 }
 

paddle/fluid/framework/operator.h

@@ -33,7 +33,6 @@ limitations under the License. */
 #include "paddle/fluid/framework/tensor.h"
 #include "paddle/fluid/platform/device_context.h"
 #include "paddle/fluid/platform/variant.h"
-#include "paddle/utils/Error.h"
 
 namespace paddle {
 namespace framework {

paddle/fluid/framework/parallel_executor.cc

@@ -52,13 +52,12 @@ std::vector<Scope *> &ParallelExecutor::GetLocalScopes() {
 }
 
 ParallelExecutor::ParallelExecutor(
-    size_t num_threads, bool use_event,
     const std::vector<platform::Place> &places,
     const std::unordered_set<std::string> &params,
     const std::unordered_set<std::string> &bcast_vars,
     const ProgramDesc &main_program, const std::string &loss_var_name,
-    Scope *scope, const std::vector<Scope *> &local_scopes, bool allow_op_delay,
-    bool use_default_grad_scale, bool balance_parameter_opt_between_cards,
+    Scope *scope, const std::vector<Scope *> &local_scopes,
+    const ExecutionStrategy &exec_strategy, const BuildStrategy &build_strategy,
     size_t num_trainers, size_t trainer_id)
     : member_(new ParallelExecutorPrivate(places)) {
   member_->global_scope_ = scope;
@@ -100,18 +99,16 @@ ParallelExecutor::ParallelExecutor(
 #ifdef PADDLE_WITH_CUDA
   details::MultiDevSSAGraphBuilder builder(
       member_->places_, loss_var_name, params, member_->local_scopes_,
-      member_->nccl_ctxs_.get(), use_default_grad_scale,
-      balance_parameter_opt_between_cards);
+      member_->nccl_ctxs_.get(), build_strategy);
 #else
-  details::MultiDevSSAGraphBuilder builder(
-      member_->places_, loss_var_name, params, member_->local_scopes_,
-      use_default_grad_scale, balance_parameter_opt_between_cards);
+  details::MultiDevSSAGraphBuilder builder(member_->places_, loss_var_name,
+                                           params, member_->local_scopes_,
+                                           build_strategy);
 #endif
   auto graph = builder.Build(main_program);
 
   member_->executor_.reset(new details::ThreadedSSAGraphExecutor(
-      num_threads, use_event, member_->local_scopes_, places, std::move(graph),
-      allow_op_delay));
+      exec_strategy, member_->local_scopes_, places, std::move(graph)));
 
   // Step 3. Create vars in each scope;
   for (auto *var : main_program.Block(0).AllVars()) {

paddle/fluid/framework/parallel_executor.h

@@ -14,57 +14,60 @@ limitations under the License. */
 
 #pragma once
 
+#include <paddle/fluid/framework/details/build_strategy.h>
 #include <string>
 #include <unordered_set>
 #include <vector>
+#include "paddle/fluid/framework/details/execution_strategy.h"
 #include "paddle/fluid/framework/executor.h"
 #include "paddle/fluid/framework/op_info.h"
 #include "paddle/fluid/framework/program_desc.h"
 #include "paddle/fluid/framework/scope.h"
 #include "paddle/fluid/framework/tensor.h"
 #include "paddle/fluid/platform/device_context.h"
-
 namespace paddle {
 namespace framework {
 
 class ParallelExecutorPrivate;
 
+using details::BuildStrategy;
+using details::ExecutionStrategy;
+
 class ParallelExecutor {
   DISABLE_COPY_AND_ASSIGN(ParallelExecutor);
 
  public:
-  explicit ParallelExecutor(size_t num_threads, bool use_event,
-                            const std::vector<platform::Place>& places,
-                            const std::unordered_set<std::string>& params,
-                            const std::unordered_set<std::string>& bcast_vars,
-                            const ProgramDesc& main_program,
-                            const std::string& loss_var_name, Scope* scope,
-                            const std::vector<Scope*>& local_scopes,
-                            bool allow_op_delay, bool use_default_grad_scale,
-                            bool balance_parameter_opt_between_cards,
+  explicit ParallelExecutor(const std::vector<platform::Place> &places,
+                            const std::unordered_set<std::string> &params,
+                            const std::unordered_set<std::string> &bcast_vars,
+                            const ProgramDesc &main_program,
+                            const std::string &loss_var_name, Scope *scope,
+                            const std::vector<Scope *> &local_scopes,
+                            const ExecutionStrategy &exec_strategy,
+                            const BuildStrategy &build_strategy,
                             size_t num_trainers = 1, size_t trainer_id = 0);
 
   ~ParallelExecutor();
 
-  std::vector<Scope*>& GetLocalScopes();
+  std::vector<Scope *> &GetLocalScopes();
 
   /**
    * Feed tensors to local scopes. The size of tensors should be equal to the
    * size of local scopes.
    */
   void FeedTensorsIntoLocalScopes(
-      const std::vector<std::unordered_map<std::string, LoDTensor>>& tensors);
+      const std::vector<std::unordered_map<std::string, LoDTensor>> &tensors);
 
   void FeedAndSplitTensorIntoLocalScopes(
-      const std::unordered_map<std::string, LoDTensor>& tensors);
+      const std::unordered_map<std::string, LoDTensor> &tensors);
 
-  void Run(const std::vector<std::string>& fetch_tensors,
-           const std::string& fetched_var_name);
+  void Run(const std::vector<std::string> &fetch_tensors,
+           const std::string &fetched_var_name);
 
-  void BCastParamsToGPUs(const std::unordered_set<std::string>& vars) const;
+  void BCastParamsToGPUs(const std::unordered_set<std::string> &vars) const;
 
  private:
-  ParallelExecutorPrivate* member_;
+  ParallelExecutorPrivate *member_;
 };
 
 }  // namespace framework

paddle/fluid/framework/tensor_impl.h

@@ -13,54 +13,14 @@ See the License for the specific language governing permissions and
 limitations under the License. */
 
 #pragma once
+#include "paddle/fluid/framework/data_type.h"
 #include "paddle/fluid/memory/memcpy.h"
 #include "paddle/fluid/platform/enforce.h"
 #include "paddle/fluid/platform/float16.h"
 
 namespace paddle {
 namespace framework {
-
-template <typename... T>
-struct SizeOfTypeFunctor;
-
-template <typename T>
-struct SizeOfTypeFunctor<T> {
-  size_t operator()(std::type_index type) const {
-    if (typeid(T).hash_code() == type.hash_code()) {
-      return sizeof(T);
-    } else {
-      return 0UL;
-    }
-  }
-};
-
-template <>
-struct SizeOfTypeFunctor<> {
-  size_t operator()(std::type_index type) const { return 0UL; }
-};
-
-template <typename HEAD, typename... TAIL>
-struct SizeOfTypeFunctor<HEAD, TAIL...> {
-  size_t operator()(std::type_index type) const {
-    SizeOfTypeFunctor<HEAD> head;
-    size_t head_size = head(type);
-    if (head_size != 0) {
-      return head_size;
-    }
-    SizeOfTypeFunctor<TAIL...> tail;
-    return tail(type);
-  }
-};
-
-static inline size_t SizeOfType(std::type_index type) {
-  SizeOfTypeFunctor<int, float, double, int16_t, int64_t, bool, size_t,
-                    platform::float16>
-      functor;
-  size_t size = functor(type);
-  PADDLE_ENFORCE(size != 0UL, "Cannot get size of type %s", type.name());
-  return size;
-}
-
+extern size_t SizeOfType(std::type_index type);
 inline void Tensor::check_memory_size() const {
   PADDLE_ENFORCE_NOT_NULL(
       holder_, "Tensor holds no memory. Call Tensor::mutable_data first.");

paddle/fluid/inference/analysis/device.h

@@ -11,6 +11,7 @@ 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. */
+#pragma once
 
 #pragma once
 

paddle/fluid/inference/tensorrt/convert/op_converter.h

@@ -49,7 +49,7 @@ class OpConverter {
   // convert fluid block to tensorrt network
   void ConvertBlock(const framework::proto::BlockDesc& block,
                     TensorRTEngine* engine) {
-    for (size_t i = 0; i < block.ops_size(); i++) {
+    for (int i = 0; i < block.ops_size(); i++) {
       const auto& op = block.ops(i);
       OpConverter::Run(op, engine);
     }

paddle/fluid/operators/CMakeLists.txt

@@ -186,11 +186,7 @@ endif()
 
 add_subdirectory(detail)
 if(WITH_DISTRIBUTE)
-    if(WITH_GPU)
-        op_library(gen_nccl_id_op DEPS nccl_common)
-    else()
-        set(DEPS_OPS ${DEPS_OPS} gen_nccl_id_op)
-    endif()
+    
     set(DISTRIBUTE_DEPS sendrecvop_grpc grpc++_unsecure grpc_unsecure gpr cares zlib protobuf)
     set(DISTRIBUTE_COMPILE_FLAGS "-Wno-non-virtual-dtor -Wno-error=non-virtual-dtor -Wno-error=delete-non-virtual-dtor")
     op_library(send_op DEPS ${DISTRIBUTE_DEPS})
@@ -210,6 +206,12 @@ if(WITH_DISTRIBUTE)
     set_source_files_properties(send_recv_op_test.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
     cc_test(test_send_recv SRCS send_recv_op_test.cc DEPS prefetch_op send_op listen_and_serv_op sum_op executor)
     cc_test(test_send_nccl_id SRCS test_send_nccl_id.cc DEPS send_op listen_and_serv_op executor)
+    if(WITH_GPU)
+        op_library(gen_nccl_id_op DEPS nccl_common sendrecvop_grpc)
+        set_source_files_properties(gen_nccl_id_op.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
+    else()
+        set(DEPS_OPS ${DEPS_OPS} gen_nccl_id_op)
+    endif()
 else()
     set(DEPS_OPS ${DEPS_OPS} send_op prefetch_op recv_op listen_and_serv_op send_vars_op send_barrier_op fetch_barrier_op gen_nccl_id_op)
 endif()

paddle/fluid/operators/detail/grpc_server.cc

@@ -306,7 +306,7 @@ void AsyncGRPCServer::TryToRegisterNewPrefetchOne() {
   }
   RequestPrefetch* prefetch =
       new RequestPrefetch(&service_, cq_prefetch_.get(), sync_mode_, scope_,
-                          dev_ctx_, executor_, program_, prefetch_ctx_);
+                          dev_ctx_, executor_, program_, prefetch_ctx_.get());
 
   VLOG(4) << "Create RequestPrefetch status:" << prefetch->Status();
 }

paddle/fluid/operators/detail/grpc_server.h

@@ -64,8 +64,9 @@ class AsyncGRPCServer final {
 
   void SetExecutor(framework::Executor *executor) { executor_ = executor; }
 
-  void SetPrefetchPreparedCtx(framework::ExecutorPrepareContext *prepared) {
-    prefetch_ctx_ = prepared;
+  void SetPrefetchPreparedCtx(
+      std::unique_ptr<framework::ExecutorPrepareContext> prepared) {
+    prefetch_ctx_.reset(prepared.release());
   }
 
   int GetSelectedPort() const { return selected_port_; }
@@ -116,7 +117,7 @@ class AsyncGRPCServer final {
   std::unique_ptr<std::thread> t_get_;
   std::unique_ptr<std::thread> t_prefetch_;
 
-  framework::ExecutorPrepareContext *prefetch_ctx_;
+  std::unique_ptr<framework::ExecutorPrepareContext> prefetch_ctx_;
   framework::ProgramDesc *program_;
   framework::Executor *executor_;
   int selected_port_;

paddle/fluid/operators/detail/grpc_server_test.cc

@@ -100,7 +100,7 @@ void StartServer(const std::string& endpoint) {
   InitTensorsOnServer(&scope, &place, 10);
 
   rpc_service_->SetProgram(&program);
-  rpc_service_->SetPrefetchPreparedCtx(prepared.get());
+  rpc_service_->SetPrefetchPreparedCtx(std::move(prepared));
   rpc_service_->SetDevCtx(&ctx);
   rpc_service_->SetScope(&scope);
   rpc_service_->SetExecutor(&exe);

paddle/fluid/operators/listen_and_serv_op.cc

@@ -322,8 +322,7 @@ void ListenAndServOp::RunImpl(const framework::Scope &scope,
   // prepare for prefetch
   VLOG(3) << "prefetch block id is " << prefetch_block->ID();
   auto prefetch_prepared = executor.Prepare(*program, prefetch_block->ID());
-  rpc_service_->SetPrefetchPreparedCtx(prefetch_prepared.get());
-  prefetch_prepared.release();
+  rpc_service_->SetPrefetchPreparedCtx(std::move(prefetch_prepared));
 
   // start the server listening after all member initialized.
   server_thread_.reset(new std::thread(RunServer, rpc_service_));

paddle/fluid/operators/load_combine_op.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 <fstream>
-
+#include "paddle/fluid/framework/data_type_transform.h"
 #include "paddle/fluid/framework/op_registry.h"
 #include "paddle/fluid/platform/device_context.h"
 
@@ -31,6 +31,7 @@ class LoadCombineOp : public framework::OperatorBase {
   void RunImpl(const framework::Scope &scope,
                const platform::Place &place) const override {
     auto filename = Attr<std::string>("file_path");
+    auto load_as_fp16 = Attr<bool>("load_as_fp16");
 
     std::ifstream fin(filename);
     PADDLE_ENFORCE(static_cast<bool>(fin),
@@ -59,17 +60,25 @@ class LoadCombineOp : public framework::OperatorBase {
       // Get data from fin to tensor
       DeserializeFromStream(fin, tensor, dev_ctx);
 
-      if (platform::is_gpu_place(place)) {
-        // copy CPU to GPU
-        framework::LoDTensor cpu_tensor;
-        cpu_tensor.ShareDataWith(*tensor);
-        cpu_tensor.set_lod(tensor->lod());
-
-        // reset tensor
+      auto in_dtype = framework::ToDataType(tensor->type());
+      auto out_dtype =
+          load_as_fp16 ? framework::proto::VarType::FP16 : in_dtype;
+
+      if (in_dtype != out_dtype) {
+        // convert to float16 tensor
+        auto in_kernel_type = framework::OpKernelType(in_dtype, place);
+        auto out_kernel_type = framework::OpKernelType(out_dtype, place);
+        framework::LoDTensor fp16_tensor;
+        // copy LoD info to the new tensor
+        fp16_tensor.set_lod(tensor->lod());
+        framework::TransDataType(in_kernel_type, out_kernel_type, *tensor,
+                                 &fp16_tensor);
+
+        // reset output tensor
         out_var->Clear();
         tensor = out_var->GetMutable<framework::LoDTensor>();
-        tensor->set_lod(cpu_tensor.lod());
-        TensorCopy(cpu_tensor, place, dev_ctx, tensor);
+        tensor->set_lod(fp16_tensor.lod());
+        tensor->ShareDataWith(fp16_tensor);
       }
     }
   }
@@ -82,6 +91,13 @@ class LoadCombineOpProtoMaker : public framework::OpProtoAndCheckerMaker {
         "Out",
         "(vector) The output LoDTensors that will be read from the input file.")
         .AsDuplicable();
+    AddAttr<bool>(
+        "load_as_fp16",
+        "(boolean, default false)"
+        "If true, the tensor will be first loaded and then "
+        "converted to float16 data type. Otherwise, the tensor will be "
+        "directly loaded without data type conversion.")
+        .SetDefault(false);
     AddAttr<std::string>("file_path",
                          "(string) "
                          "LoDTensors will be loaded from \"file_path\".")

paddle/fluid/operators/math/math_function.cc

@@ -38,7 +38,9 @@ template struct SetConstant<platform::CPUDeviceContext, bool>;
   template struct Transpose<platform::CPUDeviceContext, double, RANK>;     \
   template struct Transpose<platform::CPUDeviceContext, int, RANK>;        \
   template struct Transpose<platform::CPUDeviceContext, int64_t, RANK>;    \
-  template struct Transpose<platform::CPUDeviceContext, bool, RANK>;
+  template struct Transpose<platform::CPUDeviceContext, bool, RANK>;       \
+  template struct Transpose<platform::CPUDeviceContext, int16_t, RANK>;    \
+  template struct Transpose<platform::CPUDeviceContext, uint8_t, RANK>;
 
 DEFINE_CPU_TRANS(1);
 DEFINE_CPU_TRANS(2);

paddle/fluid/operators/save_load_combine_op_test.cc

@@ -139,8 +139,9 @@ TEST(SaveLoadCombineOp, CPU) {
   CheckValues<int, int>(expect4, actual4, expect_lod4, actual_lod4, numel4);
 }
 
-// FP16 version of SaveLoadCombineOp Test
-TEST(SaveLoadCombineFP16Op, CPU) {
+// FP16 version of SaveLoadCombineOp Test, only altering the saving aspect
+// to save as FP16.
+TEST(SaveCombineFP16Op, CPU) {
   paddle::framework::Scope scope;
   paddle::platform::CPUPlace place;
 
@@ -169,7 +170,7 @@ TEST(SaveLoadCombineFP16Op, CPU) {
       20, 50, lod4, "test_var4", place, &scope, &expect_lod4);
 
   // Set attributes
-  std::string filename = "check_tensor_fp16.ls";
+  std::string filename = "check_tensor_fp16_save.ls";
   paddle::framework::AttributeMap attrs;
   attrs.insert({"file_path", std::string(filename)});
   attrs.insert({"save_as_fp16", true});
@@ -216,6 +217,89 @@ TEST(SaveLoadCombineFP16Op, CPU) {
                                                 actual_lod4, numel4);
 }
 
+// FP16 version of SaveLoadCombineOp Test, only altering the loading aspect
+// to load tensors with FP16 precision.
+TEST(LoadCombineFP16Op, CPU) {
+  paddle::framework::Scope scope;
+  paddle::platform::CPUPlace place;
+
+  std::vector<int> lod1 = {0, 1, 2, 3, 10};
+  int numel1 = 100;
+  paddle::framework::LoD expect_lod1;
+  float* expect1 = CreateForSaveCombineOp<float, paddle::platform::float16>(
+      10, 10, lod1, "test_var1", place, &scope, &expect_lod1);
+
+  std::vector<int> lod2 = {0, 2, 5, 10};
+  int numel2 = 200;
+  paddle::framework::LoD expect_lod2;
+  float* expect2 = CreateForSaveCombineOp<float, paddle::platform::float16>(
+      10, 20, lod2, "test_var2", place, &scope, &expect_lod2);
+
+  std::vector<int> lod3 = {0, 20};
+  int numel3 = 4000;
+  paddle::framework::LoD expect_lod3;
+  float* expect3 = CreateForSaveCombineOp<float, paddle::platform::float16>(
+      20, 200, lod3, "test_var3", place, &scope, &expect_lod3);
+
+  std::vector<int> lod4 = {0, 1, 20};
+  int numel4 = 1000;
+  paddle::framework::LoD expect_lod4;
+  float* expect4 = CreateForSaveCombineOp<float, paddle::platform::float16>(
+      20, 50, lod4, "test_var4", place, &scope, &expect_lod4);
+
+  // Set attributes
+  std::string filename = "check_tensor_fp16_load.ls";
+  paddle::framework::AttributeMap attrs;
+  attrs.insert({"file_path", std::string(filename)});
+
+  // Run the save_combine_op
+  auto save_combine_op = paddle::framework::OpRegistry::CreateOp(
+      "save_combine",
+      {{"X", {"test_var1", "test_var2", "test_var3", "test_var4"}}}, {}, attrs);
+  save_combine_op->Run(scope, place);
+
+  // Set up output vars
+  auto load_var1 = scope.Var("out_var1");
+  auto load_var2 = scope.Var("out_var2");
+  auto load_var3 = scope.Var("out_var3");
+  auto load_var4 = scope.Var("out_var4");
+
+  attrs.insert({"load_as_fp16", true});
+  // Run the load_combine_op
+  auto load_combine_op = paddle::framework::OpRegistry::CreateOp(
+      "load_combine", {},
+      {{"Out", {"out_var1", "out_var2", "out_var3", "out_var4"}}}, attrs);
+  load_combine_op->Run(scope, place);
+
+  auto* target1 = load_var1->GetMutable<paddle::framework::LoDTensor>();
+  auto* target2 = load_var2->GetMutable<paddle::framework::LoDTensor>();
+  auto* target3 = load_var3->GetMutable<paddle::framework::LoDTensor>();
+  auto* target4 = load_var4->GetMutable<paddle::framework::LoDTensor>();
+
+  paddle::framework::LoD actual_lod1, actual_lod2, actual_lod3, actual_lod4;
+  paddle::platform::float16* actual1 =
+      GetValuesAfterLoadCombineOp<paddle::platform::float16>(target1, scope,
+                                                             &actual_lod1);
+  paddle::platform::float16* actual2 =
+      GetValuesAfterLoadCombineOp<paddle::platform::float16>(target2, scope,
+                                                             &actual_lod2);
+  paddle::platform::float16* actual3 =
+      GetValuesAfterLoadCombineOp<paddle::platform::float16>(target3, scope,
+                                                             &actual_lod3);
+  paddle::platform::float16* actual4 =
+      GetValuesAfterLoadCombineOp<paddle::platform::float16>(target4, scope,
+                                                             &actual_lod4);
+
+  CheckValues<float, paddle::platform::float16>(expect1, actual1, expect_lod1,
+                                                actual_lod1, numel1);
+  CheckValues<float, paddle::platform::float16>(expect2, actual2, expect_lod2,
+                                                actual_lod2, numel2);
+  CheckValues<float, paddle::platform::float16>(expect3, actual3, expect_lod3,
+                                                actual_lod3, numel3);
+  CheckValues<float, paddle::platform::float16>(expect4, actual4, expect_lod4,
+                                                actual_lod4, numel4);
+}
+
 // Test with original SaveLoadTest
 TEST(SaveLoadTestWithCombineOp, CPU) {
   paddle::framework::Scope scope;

paddle/fluid/operators/smooth_l1_loss_op.cc

@@ -105,7 +105,7 @@ class SmoothL1LossGradOp : public framework::OperatorWithKernel {
   using framework::OperatorWithKernel::OperatorWithKernel;
 
   void InferShape(framework::InferShapeContext* ctx) const override {
-    auto in_dims = ctx->GetInputDim("X");
+    auto in_dims = ctx->GetInputDim("Diff");
     auto out_dims = ctx->GetInputDim(framework::GradVarName("Out"));
 
     PADDLE_ENFORCE_GE(out_dims.size(), 2,
@@ -127,12 +127,33 @@ class SmoothL1LossGradOp : public framework::OperatorWithKernel {
   }
 };
 
+class SmoothL1LossGradMaker : public framework::SingleGradOpDescMaker {
+ public:
+  using framework::SingleGradOpDescMaker::SingleGradOpDescMaker;
+
+ protected:
+  std::unique_ptr<framework::OpDesc> Apply() const override {
+    auto* op = new framework::OpDesc();
+    op->SetType("smooth_l1_loss_grad");
+    op->SetInput("InsideWeight", Input("InsideWeight"));
+    op->SetInput("OutsideWeight", Input("OutsideWeight"));
+    op->SetInput("Diff", Output("Diff"));
+    op->SetInput(framework::GradVarName("Out"), OutputGrad("Out"));
+
+    op->SetAttrMap(Attrs());
+
+    op->SetOutput(framework::GradVarName("X"), InputGrad("X"));
+    op->SetOutput(framework::GradVarName("Y"), InputGrad("Y"));
+    return std::unique_ptr<framework::OpDesc>(op);
+  }
+};
+
 }  // namespace operators
 }  // namespace paddle
 
 namespace ops = paddle::operators;
 REGISTER_OPERATOR(smooth_l1_loss, ops::SmoothL1LossOp, ops::SmoothL1LossOpMaker,
-                  paddle::framework::DefaultGradOpDescMaker<true>);
+                  ops::SmoothL1LossGradMaker);
 REGISTER_OPERATOR(smooth_l1_loss_grad, ops::SmoothL1LossGradOp);
 REGISTER_OP_CPU_KERNEL(
     smooth_l1_loss,

paddle/fluid/platform/CMakeLists.txt

-proto_library(profiler_proto SRCS profiler.proto)
+proto_library(profiler_proto SRCS profiler.proto DEPS framework_proto)
 py_proto_compile(profiler_py_proto SRCS profiler.proto)
 
 add_custom_target(profiler_py_proto_init ALL COMMAND ${CMAKE_COMMAND} -E touch __init__.py)

paddle/fluid/platform/nccl_helper.h

@@ -53,7 +53,7 @@ class NCCLGroupGuard {
   }
 
   inline ~NCCLGroupGuard() {
-    PADDLE_ENFORCE(dynload::ncclGroupEnd());
+    CHECK_EQ(dynload::ncclGroupEnd(), ncclSuccess);
     NCCLMutex().unlock();
   }
 };

paddle/fluid/pybind/pybind.cc

@@ -494,23 +494,61 @@ All parameter, weight, gradient are variables in Paddle.
   m.def("disable_profiler", platform::DisableProfiler);
   m.def("reset_profiler", platform::ResetProfiler);
 
-  py::class_<ParallelExecutor>(m, "ParallelExecutor")
-      .def("__init__",
-           [](ParallelExecutor &self, size_t num_threads, bool use_event,
-              const std::vector<platform::Place> &places,
-              const std::unordered_set<std::string> &params,
-              const std::unordered_set<std::string> &bcast_vars,
-              const ProgramDesc &main_program, const std::string &loss_var_name,
-              Scope *scope, std::vector<Scope *> &local_scopes,
-              bool allow_op_delay, bool use_default_grad_scale,
-              bool balance_parameter_opt_between_cards, size_t num_trainers,
-              size_t trainer_id) {
-             new (&self) ParallelExecutor(
-                 num_threads, use_event, places, params, bcast_vars,
-                 main_program, loss_var_name, scope, local_scopes,
-                 allow_op_delay, use_default_grad_scale,
-                 balance_parameter_opt_between_cards, num_trainers, trainer_id);
-           })
+  // -- python binds for parallel executor.
+  py::class_<ParallelExecutor> pe(m, "ParallelExecutor");
+  py::class_<ExecutionStrategy>(pe, "ExecutionStrategy")
+      .def(py::init())
+      .def_property(
+          "num_threads",
+          [](const ExecutionStrategy &self) { return self.num_threads_; },
+          [](ExecutionStrategy &self, size_t num_threads) {
+            self.num_threads_ = num_threads;
+          })
+      .def_property(
+          "use_event",
+          [](const ExecutionStrategy &self) { return self.use_event_; },
+          [](ExecutionStrategy &self, bool use_event) {
+            self.use_event_ = use_event;
+          })
+      .def_property(
+          "allow_op_delay",
+          [](const ExecutionStrategy &self) { return self.allow_op_delay_; },
+          [](ExecutionStrategy &self, bool allow_op_delay) {
+            self.allow_op_delay_ = allow_op_delay;
+          });
+  py::class_<BuildStrategy> build_strategy(pe, "BuildStrategy");
+
+  py::enum_<BuildStrategy::ReduceStrategy>(build_strategy, "ReduceStrategy")
+      .value("Reduce", BuildStrategy::ReduceStrategy::kReduce)
+      .value("AllReduce", BuildStrategy::ReduceStrategy::kAllReduce);
+  py::enum_<BuildStrategy::GradientScaleStrategy>(build_strategy,
+                                                  "GradientScaleStrategy")
+      .value("CoeffNumDevice",
+             BuildStrategy::GradientScaleStrategy::kCoeffNumDevice)
+      .value("One", BuildStrategy::GradientScaleStrategy::kOne)
+      .value("Customized", BuildStrategy::GradientScaleStrategy::kCustomized);
+
+  build_strategy.def(py::init())
+      .def_property(
+          "reduce_strategy",
+          [](const BuildStrategy &self) { return self.reduce_; },
+          [](BuildStrategy &self, BuildStrategy::ReduceStrategy strategy) {
+            self.reduce_ = strategy;
+          })
+      .def_property(
+          "gradient_scale_strategy",
+          [](const BuildStrategy &self) { return self.gradient_scale_; },
+          [](BuildStrategy &self,
+             BuildStrategy::GradientScaleStrategy strategy) {
+            self.gradient_scale_ = strategy;
+          });
+
+  pe.def(py::init<const std::vector<platform::Place> &,
+                  const std::unordered_set<std::string> &,
+                  const std::unordered_set<std::string> &, const ProgramDesc &,
+                  const std::string &, Scope *, std::vector<Scope *> &,
+                  const ExecutionStrategy &, const BuildStrategy &, size_t,
+                  size_t>())
       .def("bcast_params", &ParallelExecutor::BCastParamsToGPUs)
       // NOTE: even we return a vec<Scope*>* to Python use reference policy.
       // We still cannot get local_scope from this vector, since the element

paddle/scripts/paddle_build.sh

@@ -20,19 +20,15 @@
 #=================================================
 
 function print_usage() {
-    RED='\033[0;31m'
-    BLUE='\033[0;34m'
-    BOLD='\033[1m'
-    NONE='\033[0m'
-    
     echo -e "\n${RED}Usage${NONE}:
-    ${BOLD}$0${NONE} [OPTION]"
+    ${BOLD}${SCRIPT_NAME}${NONE} [OPTION]"
     
     echo -e "\n${RED}Options${NONE}:
     ${BLUE}build${NONE}: run build for x86 platform
     ${BLUE}build_android${NONE}: run build for android platform
     ${BLUE}build_ios${NONE}: run build for ios platform
     ${BLUE}test${NONE}: run all unit tests
+    ${BLUE}single_test${NONE}: run a single unit test
     ${BLUE}bind_test${NONE}: parallel tests bind to different GPU
     ${BLUE}doc${NONE}: generate paddle documents
     ${BLUE}html${NONE}: convert C++ source code into HTML
@@ -45,7 +41,15 @@ function print_usage() {
 }
 
 function init() {
+    RED='\033[0;31m'
+    BLUE='\033[0;34m'
+    BOLD='\033[1m'
+    NONE='\033[0m'
+
     PADDLE_ROOT="$( cd "$( dirname "${BASH_SOURCE[0]}")/../../" && pwd )"
+    if [ -z "${SCRIPT_NAME}" ]; then
+        SCRIPT_NAME=$0
+    fi
 }
 
 function cmake_gen() {
@@ -309,6 +313,25 @@ EOF
     fi
 }
 
+function single_test() {
+    TEST_NAME=$1
+    if [ -z "${TEST_NAME}" ]; then
+        echo -e "${RED}Usage:${NONE}"
+        echo -e "${BOLD}${SCRIPT_NAME}${NONE} ${BLUE}single_test${NONE} [test_name]"
+        exit 1
+    fi
+    mkdir -p ${PADDLE_ROOT}/build
+    cd ${PADDLE_ROOT}/build
+    if [ ${WITH_TESTING:-ON} == "ON" ] ; then
+    cat <<EOF
+    ========================================
+    Running ${TEST_NAME} ...
+    ========================================
+EOF
+        ctest --output-on-failure -R ${TEST_NAME}
+    fi
+}
+
 function bind_test() {
     # the number of process to run tests
     NUM_PROC=6
@@ -480,6 +503,7 @@ function main() {
       build)
         cmake_gen ${PYTHON_ABI:-""}
         build
+        gen_dockerfile
         ;;
       build_android)
         build_android
@@ -490,6 +514,9 @@ function main() {
       test)
         run_test
         ;;
+      single_test)
+        single_test $2
+        ;;
       bind_test)
         bind_test
         ;;
@@ -504,6 +531,7 @@ function main() {
         ;;
       capi)
         cmake_gen ${PYTHON_ABI:-""}
+        build
         gen_capi_package
         ;;
       fluid_inference_lib)

paddle/scripts/paddle_docker_build.sh

@@ -63,6 +63,7 @@ EOL
     ${DOCKER_CMD} run -it \
         --name $CONTAINER_ID \
         ${DOCKER_ENV} \
+        -e SCRIPT_NAME=$0 \
         -v $PADDLE_ROOT:/paddle \
         -v ${HOME}/.ccache:/root/.ccache \
         -w /paddle \

python/paddle/fluid/__init__.py

@@ -44,42 +44,44 @@ import transpiler
 from param_attr import ParamAttr, WeightNormParamAttr
 from data_feeder import DataFeeder
 from core import LoDTensor, CPUPlace, CUDAPlace, CUDAPinnedPlace
-from transpiler import DistributeTranspiler, SimpleDistributeTranspiler, InferenceTranspiler, memory_optimize, release_memory
+from transpiler import DistributeTranspiler, SimpleDistributeTranspiler, \
+    InferenceTranspiler, memory_optimize, release_memory
 from concurrency import (Go, make_channel, channel_send, channel_recv,
                          channel_close, Select)
 import clip
 import profiler
 import unique_name
 import recordio_writer
-from parallel_executor import ParallelExecutor
+import parallel_executor
+from parallel_executor import *
 
 Tensor = LoDTensor
 
-__all__ = framework.__all__ + executor.__all__ + concurrency.__all__ +\
-          trainer.__all__ + inferencer.__all__ + transpiler.__all__ + [
-    'io',
-    'initializer',
-    'layers',
-    'transpiler'
-    'nets',
-    'optimizer',
-    'learning_rate_decay',
-    'backward',
-    'regularizer',
-    'LoDTensor',
-    'CPUPlace',
-    'CUDAPlace',
-    'CUDAPinnedPlace',
-    'Tensor',
-    'ParamAttr',
-    'WeightNormParamAttr',
-    'DataFeeder',
-    'clip',
-    'profiler',
-    'unique_name',
-    'recordio_writer',
-    'ParallelExecutor',
-]
+__all__ = framework.__all__ + executor.__all__ + concurrency.__all__ + \
+          trainer.__all__ + inferencer.__all__ + transpiler.__all__ + \
+          parallel_executor.__all__ + [
+              'io',
+              'initializer',
+              'layers',
+              'transpiler'
+              'nets',
+              'optimizer',
+              'learning_rate_decay',
+              'backward',
+              'regularizer',
+              'LoDTensor',
+              'CPUPlace',
+              'CUDAPlace',
+              'CUDAPinnedPlace',
+              'Tensor',
+              'ParamAttr',
+              'WeightNormParamAttr',
+              'DataFeeder',
+              'clip',
+              'profiler',
+              'unique_name',
+              'recordio_writer',
+          ]
 
 
 def __bootstrap__():

python/paddle/fluid/data_feeder.py

@@ -54,9 +54,9 @@ class DataToLoDTensorConverter(object):
             self.data.append(data)
         else:
             cur_lod_len = len(data)
-            lod[-1].append(lod[-1][-1] + cur_lod_len)
+            lod[0].append(lod[0][-1] + cur_lod_len)
             for each_data in data:
-                self._feed_impl_(each_data, lod[:-1], lod_level - 1)
+                self._feed_impl_(each_data, lod[1:], lod_level - 1)
 
     def done(self):
         arr = numpy.array(self.data, dtype=self.dtype).reshape(self.shape)

python/paddle/fluid/inferencer.py

@@ -13,29 +13,35 @@
 # limitations under the License.
 
 import core
-import framework
+
 import executor
+import framework
 import io
+import unique_name
 from trainer import check_and_get_place
 
 __all__ = ['Inferencer', ]
 
 
 class Inferencer(object):
-    def __init__(self, param_path, place=None):
+    def __init__(self, infer_func, param_path, place=None):
         """
-        :param param_path: the path where the inference model is saved by fluid.io.save_inference_model
+        :param infer_func: a function that will return predict Variable
+        :param param_path: the path where the inference model is saved by fluid.io.save_params
         :param place: place to do the inference
         """
         self.param_path = param_path
         self.scope = core.Scope()
 
+        self.inference_program = framework.Program()
+        with framework.program_guard(self.inference_program):
+            with unique_name.guard():
+                self.predict_var = infer_func()
+
         self.exe = executor.Executor(check_and_get_place(place))
         with executor.scope_guard(self.scope):
             # load params from param_path into scope
-            [self.inference_program, _,
-             self.fetch_targets] = io.load_inference_model(
-                 executor=self.exe, dirname=param_path)
+            io.load_params(self.exe, param_path, self.inference_program)
 
     def infer(self, inputs, return_numpy=True):
         """
@@ -51,7 +57,7 @@ class Inferencer(object):
         with executor.scope_guard(self.scope):
             results = self.exe.run(self.inference_program,
                                    feed=inputs,
-                                   fetch_list=self.fetch_targets,
+                                   fetch_list=[self.predict_var],
                                    return_numpy=return_numpy)
 
         return results

python/paddle/fluid/layers/nn.py

@@ -1329,6 +1329,8 @@ def sequence_pool(input, pool_type):
          sqrt   : out.data = [2.82, 6.93, 4.24], where 2.82=(1+3)/sqrt(2),
                     6.93=(2+4+6)/sqrt(3), 4.24=(5+1)/sqrt(2)
          max    : out.data = [3, 6, 5], where 3=max(1,3), 6=max(2,4,6), 5=max(5,1)
+         last   : out.data = [3, 6, 1], where 3=last(1,3), 6=last(2,4,6), 1=last(5,1)
+         first  : out.data = [1, 2, 5], where 1=first(1,3), 2=first(2,4,6), 5=first(5,1)
 
     Args:
         input(variable): The input variable which is a LoDTensor.
@@ -1348,6 +1350,8 @@ def sequence_pool(input, pool_type):
              sum_x = fluid.layers.sequence_pool(input=x, pool_type='sum')
              sqrt_x = fluid.layers.sequence_pool(input=x, pool_type='sqrt')
              max_x = fluid.layers.sequence_pool(input=x, pool_type='max')
+             last_x = fluid.layers.sequence_pool(input=x, pool_type='last')
+             first_x = fluid.layers.sequence_pool(input=x, pool_type='first')
     """
     helper = LayerHelper('sequence_pool', **locals())
     dtype = helper.input_dtype()
@@ -3263,35 +3267,35 @@ def smooth_l1(x, y, inside_weight=None, outside_weight=None, sigma=None):
     """
     **Smooth L1 Loss Operator. **
 
-    This operator computes the smooth l1 loss for X and Y.
+    This operator computes the smooth L1 loss for X and Y.
     The operator takes the first dimension of X and Y as batch size.
-    For each instance, it computes the smooth l1 loss element by element first
+    For each instance, it computes the smooth L1 loss element by element first
     and then sums all the losses. So the shape of Out is [batch_size, 1].
 
     Args:
         x (Variable): A tensor with rank at least 2. The input value of smooth
-            l1 loss op with shape [batch_size, dim1, ..., dimN].
+            L1 loss op with shape [batch_size, dim1, ..., dimN].
         y (Variable): A tensor with rank at least 2. The target value of smooth
-            l1 loss op with same shape as x.
+            L1 loss op with same shape as x.
         inside_weight (Variable|None):  A tensor with rank at least 2. This
             input is optional and should have same shape with x. If provided,
             the result of (x - y) will be multiplied by this tensor element by
             element.
         outside_weight (Variable|None): A tensor with rank at least 2. This
             input is optional and should have same shape with x. If provided,
-            the out smooth l1 loss will be multiplied by this tensor element
+            the out smooth L1 loss will be multiplied by this tensor element
             by element.
-        sigma (float|None): Hyper parameter of smooth l1 loss op. A float scalar
+        sigma (float|None): Hyper parameter of smooth L1 loss op. A float scalar
             with default value 1.0.
     Returns:
-        Variable: A tensor with rank be 2. The output smooth l1 loss with
+        Variable: A tensor with rank be 2. The output smooth L1 loss with
             shape [batch_size, 1].
 
     Examples:
         .. code-block:: python
 
             data = fluid.layers.data(name='data', shape=[128], dtype='float32')
-            label = fluid.layers.data(name='label', shape=[100], dtype='int64')
+            label = fluid.layers.data(name='label', shape=[100], dtype='float32')
             fc = fluid.layers.fc(input=data, size=100)
             out = fluid.layers.smooth_l1(x=fc, y=label)
     """
@@ -3769,13 +3773,13 @@ def label_smooth(label,
 
 def roi_pool(input, rois, pooled_height=1, pooled_width=1, spatial_scale=1.0):
     """
-    Region of interest pooling (also known as RoI pooling) is to perform 
+    Region of interest pooling (also known as RoI pooling) is to perform
         is to perform max pooling on inputs of nonuniform sizes to obtain
         fixed-size feature maps (e.g. 7*7).
-    The operator has three steps: 
-        1. Dividing each region proposal into equal-sized sections with 
-           the pooled_width and pooled_height 
-        2. Finding the largest value in each section 
+    The operator has three steps:
+        1. Dividing each region proposal into equal-sized sections with
+           the pooled_width and pooled_height
+        2. Finding the largest value in each section
         3. Copying these max values to the output buffer
 
     Args:
@@ -3783,8 +3787,8 @@ def roi_pool(input, rois, pooled_height=1, pooled_width=1, spatial_scale=1.0):
         rois (Variable): ROIs (Regions of Interest) to pool over. It should
                          be a 2-D one level LoTensor of shape [num_rois, 4].
                          The layout is [x1, y1, x2, y2], where (x1, y1)
-                         is the top left coordinates, and (x2, y2) is the 
-                         bottom right coordinates. The num_rois is the 
+                         is the top left coordinates, and (x2, y2) is the
+                         bottom right coordinates. The num_rois is the
                          total number of ROIs in this batch data.
         pooled_height (integer): The pooled output height. Default: 1
         pooled_width (integer): The pooled output width. Default: 1
@@ -3793,11 +3797,11 @@ def roi_pool(input, rois, pooled_height=1, pooled_width=1, spatial_scale=1.0):
                                to the scale used when pooling. Default: 1.0
 
     Returns:
-        pool_out (Variable): The output is a 4-D tensor of the shape 
+        pool_out (Variable): The output is a 4-D tensor of the shape
                              (num_rois, channels, pooled_h, pooled_w).
 
     Examples:
-            pool_out = fluid.layers.roi_pool(input=x, rois=rois, 7, 7, 1.0) 
+            pool_out = fluid.layers.roi_pool(input=x, rois=rois, 7, 7, 1.0)
     """
     helper = LayerHelper('roi_pool', **locals())
     dtype = helper.input_dtype()

python/paddle/fluid/parallel_executor.py

@@ -19,7 +19,10 @@ import executor
 import warnings
 import sys
 
-__all__ = ['ParallelExecutor']
+__all__ = ['ParallelExecutor', 'ExecutionStrategy', 'BuildStrategy']
+
+ExecutionStrategy = core.ParallelExecutor.ExecutionStrategy
+BuildStrategy = core.ParallelExecutor.BuildStrategy
 
 
 class ParallelExecutor(object):
@@ -27,13 +30,12 @@ class ParallelExecutor(object):
                  use_cuda,
                  loss_name=None,
                  main_program=None,
-                 num_threads=None,
-                 allow_op_delay=False,
                  share_vars_from=None,
-                 use_default_grad_scale=True,
-                 balance_parameter_opt_between_cards=False,
+                 exec_strategy=None,
+                 build_strategy=None,
                  num_trainers=1,
-                 trainer_id=0):
+                 trainer_id=0,
+                 **kwargs):
         """
         ParallelExecutor can run program in parallel.
 
@@ -42,21 +44,8 @@ class ParallelExecutor(object):
             loss_name(str, default None): The loss name must set in training.
             main_program(Program, default None): The program that need to run,
                 if not provided, then default_main_program will be used.
-            num_threads(int, default None): How many threads are used for
-                training.
-            allow_op_delay(bool, default False): Whether to delay and buffer
-                some operators together for scheduling or not, which may
-                improve performance in some cases, default False.
             share_vars_from(ParallelExecutor, default None): If provied,
                 it will share variables from the specified ParallelExecutor.
-            use_default_grad_scale(bool, default True): If set True, a default
-                scale value equal to `1./device_count` would be multiplied to
-                gradients of each device and scaled gradients would be
-                aggregated. Otherwise, a customized scale value should be fed
-                to the network.
-            balance_parameter_opt_between_cards(bool, default True): Whether
-                updating different gradients on different cards. Currently, it
-                is not recommended.
             num_trainers(int, default 1): If greater than 1, NCCL will be
                 initialized with multpile rank of nodes, each node should have
                 same number of GPUs. Distributed training will be enabled then.
@@ -83,6 +72,25 @@ class ParallelExecutor(object):
               train_loss, = train_exe.run([loss.name], feed=feed_dict)
               test_loss, = test_exe.run([loss.name], feed=feed_dict)
         """
+        if len(kwargs) != 0:
+            err_msg = ""
+            for key in kwargs:
+                if key in dir(ExecutionStrategy):
+                    err_msg += \
+                        "Setting {0} by constructor is deprecated. Use " \
+                        "strategy=ExecutionStrategy(); strategy.{0}=xxx; " \
+                        "pe=ParallelExecutor(exec_strategy=strategy) " \
+                        "instead.\n ".format(key)
+                elif key in dir(BuildStrategy):
+                    err_msg += \
+                        "Setting {0} by constructor is deprecated. Use " \
+                        "strategy=BuildStrategy(); See help(" \
+                        "paddle.fluid.ParallelExecutor.BuildStrategy) \n".format(
+                            key)
+                else:
+                    err_msg += "Setting {0} by constructor is deprecated. Use strategy.\n".format(
+                        key)
+            raise ValueError(err_msg)
 
         self._places = []
         self._act_places = []
@@ -100,15 +108,25 @@ class ParallelExecutor(object):
                 self._places.append(p)
         assert self._places, "no place for execution"
 
-        if num_threads is None:
+        if exec_strategy is None:
+            exec_strategy = ExecutionStrategy()
+            if use_cuda:
+                exec_strategy.use_event = True
+            else:
+                exec_strategy.use_event = False
+
+        if exec_strategy.num_threads == 0:
             if use_cuda:
                 # Experiments on se-resnext shows that too many threads hurt
                 # performance. Worth tunning for other models in the future.
-                num_threads = len(self._places) * 2
+                exec_strategy.num_threads = len(self._places) * 2
             else:
-                num_threads = min(
+                exec_strategy.num_threads = min(
                     len(self._places) * 2, multiprocessing.cpu_count())
 
+        if build_strategy is None:
+            build_strategy = BuildStrategy()
+
         main = main_program
         main = main if main else framework.default_main_program()
         scope = executor.global_scope()
@@ -127,23 +145,14 @@ class ParallelExecutor(object):
         ]
 
         self.executor = core.ParallelExecutor(
-            num_threads,
-            True if use_cuda else False,  # use_event
             self._places,
             set([
                 p.name for p in main.global_block().iter_parameters()
                 if not p.stop_gradient
             ]),
-            set(self.persistable_vars),
-            main.desc,
-            loss_name if loss_name else '',
-            scope,
-            local_scopes,
-            allow_op_delay,
-            use_default_grad_scale,
-            balance_parameter_opt_between_cards,
-            num_trainers,
-            trainer_id)
+            set(self.persistable_vars), main.desc, loss_name
+            if loss_name else '', scope, local_scopes, exec_strategy,
+            build_strategy, num_trainers, trainer_id)
         self.scope = scope
 
     def run(self, fetch_list, feed=None, feed_dict=None):

python/paddle/fluid/tests/book/high-level-api/fit_a_line/test_fit_a_line.py

@@ -48,12 +48,11 @@ def linear():
     return avg_loss
 
 
-def train(use_cuda, save_dirname):
+def train(use_cuda, train_program, save_dirname):
     place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
 
     trainer = fluid.Trainer(
-        train_func=linear,
-        infer_func=inference_program,
+        train_func=train_program,
         place=place,
         optimizer=fluid.optimizer.SGD(learning_rate=0.001))
 
@@ -72,11 +71,7 @@ def train(use_cuda, save_dirname):
             '''
             if float(test_metrics[0]) < 20.0:
                 if save_dirname is not None:
-                    # NOT clear yet
-                    # fluid.io.save_inference_model(save_dirname, ['x'], [y_predict])
-                    # trainer.save_params(save_dirname)
-                    # https://github.com/PaddlePaddle/Paddle/pull/10445
-                    trainer.save_inference_model(save_dirname)
+                    trainer.save_params(save_dirname)
             return
 
     trainer.train(
@@ -87,12 +82,13 @@ def train(use_cuda, save_dirname):
 
 
 # infer
-def infer(use_cuda, save_dirname=None):
+def infer(use_cuda, inference_program, save_dirname=None):
     if save_dirname is None:
         return
 
     place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
-    inferencer = fluid.Inferencer(param_path=save_dirname, place=place)
+    inferencer = fluid.Inferencer(
+        infer_func=inference_program, param_path=save_dirname, place=place)
 
     batch_size = 10
     tensor_x = numpy.random.uniform(0, 10, [batch_size, 13]).astype("float32")
@@ -108,8 +104,8 @@ def main(use_cuda):
     # Directory for saving the trained model
     save_dirname = "fit_a_line.inference.model"
 
-    train(use_cuda, save_dirname)
-    infer(use_cuda, save_dirname)
+    train(use_cuda, linear, save_dirname)
+    infer(use_cuda, inference_program, save_dirname)
 
 
 class TestFitALine(unittest.TestCase):

python/paddle/fluid/tests/book/high-level-api/recognize_digits/test_recognize_digits_conv.py

@@ -53,48 +53,40 @@ def train_program():
     predict = inference_program()
     cost = fluid.layers.cross_entropy(input=predict, label=label)
     avg_cost = fluid.layers.mean(cost)
-    # acc = fluid.layers.accuracy(input=predict, label=label)
-    # return avg_cost, acc
-    return avg_cost
+    acc = fluid.layers.accuracy(input=predict, label=label)
+    return [avg_cost, acc]
 
 
-def train(use_cuda, save_dirname):
+def train(use_cuda, train_program, save_dirname):
     place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
     optimizer = fluid.optimizer.Adam(learning_rate=0.001)
 
     trainer = fluid.Trainer(
         train_func=train_program,
-        infer_func=inference_program,
         place=place,
-        optimizer=optimizer)
+        optimizer=optimizer,
+        parallel=True)
 
     def event_handler(event):
         if isinstance(event, fluid.EndEpochEvent):
-            # if (event.epoch + 1) % 10 == 0:
-            # trainer.save_params(save_dirname)
-            trainer.save_inference_model(save_dirname)
-
-            # TODO: Uncomment this part once we are sure that .train is working
-            # test_reader = paddle.batch(
-            #     paddle.dataset.mnist.test(), batch_size=BATCH_SIZE)
-            # test_metrics = trainer.test(reader=test_reader)
-            # avg_cost_set = test_metrics[0]
-            # acc_set = test_metrics[1]
-            #
-            # # get test acc and loss
-            # acc = numpy.array(acc_set).mean()
-            # avg_cost = numpy.array(avg_cost_set).mean()
-            #
-            # print("avg_cost: %s" % avg_cost)
-            # print("acc     : %s" % acc)
-            #
-            # if float(acc) > 0.2:  # Smaller value to increase CI speed
-            #     trainer.save_params(save_dirname)
-            # else:
-            #     print('BatchID {0}, Test Loss {1:0.2}, Acc {2:0.2}'.format(
-            #         event.epoch + 1, float(avg_cost), float(acc)))
-            #     if math.isnan(float(avg_cost)):
-            #         sys.exit("got NaN loss, training failed.")
+            test_reader = paddle.batch(
+                paddle.dataset.mnist.test(), batch_size=BATCH_SIZE)
+            avg_cost, acc = trainer.test(
+                reader=test_reader, feed_order=['img', 'label'])
+
+            print("avg_cost: %s" % avg_cost)
+            print("acc     : %s" % acc)
+
+            if acc > 0.2:  # Smaller value to increase CI speed
+                trainer.save_params(save_dirname)
+            else:
+                print('BatchID {0}, Test Loss {1:0.2}, Acc {2:0.2}'.format(
+                    event.epoch + 1, avg_cost, acc))
+                if math.isnan(avg_cost):
+                    sys.exit("got NaN loss, training failed.")
+        elif isinstance(event, fluid.EndStepEvent):
+            print("Step {0}, Epoch {1} Metrics {2}".format(
+                event.step, event.epoch, map(numpy.array, event.metrics)))
 
     train_reader = paddle.batch(
         paddle.reader.shuffle(
@@ -108,10 +100,11 @@ def train(use_cuda, save_dirname):
         feed_order=['img', 'label'])
 
 
-def infer(use_cuda, save_dirname=None):
+def infer(use_cuda, inference_program, save_dirname=None):
     place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
 
-    inferencer = fluid.Inferencer(param_path=save_dirname, place=place)
+    inferencer = fluid.Inferencer(
+        infer_func=inference_program, param_path=save_dirname, place=place)
 
     batch_size = 1
     tensor_img = numpy.random.uniform(-1.0, 1.0,
@@ -126,10 +119,16 @@ def main(use_cuda):
     save_dirname = "recognize_digits_conv.inference.model"
 
     # call train() with is_local argument to run distributed train
-    train(use_cuda=use_cuda, save_dirname=save_dirname)
-    infer(use_cuda=use_cuda, save_dirname=save_dirname)
+    train(
+        use_cuda=use_cuda,
+        train_program=train_program,
+        save_dirname=save_dirname)
+    infer(
+        use_cuda=use_cuda,
+        inference_program=inference_program,
+        save_dirname=save_dirname)
 
 
 if __name__ == '__main__':
     # for use_cuda in (False, True):
-    main(use_cuda=False)
+    main(use_cuda=True)

python/paddle/fluid/tests/book/high-level-api/recognize_digits/test_recognize_digits_mlp.py

@@ -40,47 +40,34 @@ def train_program():
     predict = inference_program()
     cost = fluid.layers.cross_entropy(input=predict, label=label)
     avg_cost = fluid.layers.mean(cost)
-    # acc = fluid.layers.accuracy(input=predict, label=label)
-    # return avg_cost, acc
-    return avg_cost
+    acc = fluid.layers.accuracy(input=predict, label=label)
+    return [avg_cost, acc]
 
 
-def train(use_cuda, save_dirname):
+def train(use_cuda, train_program, save_dirname):
     place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
     optimizer = fluid.optimizer.Adam(learning_rate=0.001)
 
     trainer = fluid.Trainer(
-        train_func=train_program,
-        infer_func=inference_program,
-        place=place,
-        optimizer=optimizer)
+        train_func=train_program, place=place, optimizer=optimizer)
 
     def event_handler(event):
         if isinstance(event, fluid.EndEpochEvent):
-            # if (event.epoch + 1) % 10 == 0:
-            trainer.save_inference_model(save_dirname)
-
-            # TODO: Uncomment this part once we are sure that .train is working
-            # test_reader = paddle.batch(
-            #     paddle.dataset.mnist.test(), batch_size=BATCH_SIZE)
-            # test_metrics = trainer.test(reader=test_reader)
-            # avg_cost_set = test_metrics[0]
-            # acc_set = test_metrics[1]
-            #
-            # # get test acc and loss
-            # acc = numpy.array(acc_set).mean()
-            # avg_cost = numpy.array(avg_cost_set).mean()
-            #
-            # print("avg_cost: %s" % avg_cost)
-            # print("acc     : %s" % acc)
-            #
-            # if float(acc) > 0.2:  # Smaller value to increase CI speed
-            #     trainer.save_params(save_dirname)
-            # else:
-            #     print('BatchID {0}, Test Loss {1:0.2}, Acc {2:0.2}'.format(
-            #         event.epoch + 1, float(avg_cost), float(acc)))
-            #     if math.isnan(float(avg_cost)):
-            #         sys.exit("got NaN loss, training failed.")
+            test_reader = paddle.batch(
+                paddle.dataset.mnist.test(), batch_size=BATCH_SIZE)
+            avg_cost, acc = trainer.test(
+                reader=test_reader, feed_order=['img', 'label'])
+
+            print("avg_cost: %s" % avg_cost)
+            print("acc     : %s" % acc)
+
+            if acc > 0.2:  # Smaller value to increase CI speed
+                trainer.save_params(save_dirname)
+            else:
+                print('BatchID {0}, Test Loss {1:0.2}, Acc {2:0.2}'.format(
+                    event.epoch + 1, avg_cost, acc))
+                if math.isnan(avg_cost):
+                    sys.exit("got NaN loss, training failed.")
 
     train_reader = paddle.batch(
         paddle.reader.shuffle(
@@ -94,10 +81,11 @@ def train(use_cuda, save_dirname):
         feed_order=['img', 'label'])
 
 
-def infer(use_cuda, save_dirname=None):
+def infer(use_cuda, inference_program, save_dirname=None):
     place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
 
-    inferencer = fluid.Inferencer(param_path=save_dirname, place=place)
+    inferencer = fluid.Inferencer(
+        infer_func=inference_program, param_path=save_dirname, place=place)
 
     batch_size = 1
     tensor_img = numpy.random.uniform(-1.0, 1.0,
@@ -112,8 +100,14 @@ def main(use_cuda):
     save_dirname = "recognize_digits_mlp.inference.model"
 
     # call train() with is_local argument to run distributed train
-    train(use_cuda=use_cuda, save_dirname=save_dirname)
-    infer(use_cuda=use_cuda, save_dirname=save_dirname)
+    train(
+        use_cuda=use_cuda,
+        train_program=train_program,
+        save_dirname=save_dirname)
+    infer(
+        use_cuda=use_cuda,
+        inference_program=inference_program,
+        save_dirname=save_dirname)
 
 
 if __name__ == '__main__':

python/paddle/fluid/tests/book/high-level-api/word2vec/no_test_word2vec_new_api.py

@@ -90,7 +90,7 @@ def train_program(is_sparse):
     return avg_cost
 
 
-def train(use_cuda, is_sparse, save_path):
+def train(use_cuda, train_program, save_path):
     train_reader = paddle.batch(
         paddle.dataset.imikolov.train(word_dict, N), BATCH_SIZE)
     test_reader = paddle.batch(
@@ -105,23 +105,21 @@ def train(use_cuda, is_sparse, save_path):
             print("loss= ", avg_cost)
 
             if avg_cost < 5.0:
-                trainer.save_inference_model(save_path)
+                trainer.save_params(save_path)
                 return
             if math.isnan(avg_cost):
                 sys.exit("got NaN loss, training failed.")
 
     trainer = fluid.Trainer(
-        partial(train_program, is_sparse),
-        partial(inference_program, is_sparse),
-        fluid.optimizer.SGD(learning_rate=0.001),
-        place=place)
+        train_program, fluid.optimizer.SGD(learning_rate=0.001), place=place)
     trainer.train(
         reader=train_reader, num_epochs=1, event_handler=event_handler)
 
 
-def infer(use_cuda, is_sparse, save_path):
+def infer(use_cuda, inference_program, save_path):
     place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
-    inferencer = fluid.Inferencer(param_path=save_path, place=place)
+    inferencer = fluid.Inferencer(
+        infer_func=inference_program, param_path=save_path, place=place)
 
     lod = [0, 1]
     first_word = create_random_lodtensor(lod, place, low=0, high=dict_size - 1)
@@ -144,9 +142,9 @@ def main(use_cuda, is_sparse):
     if use_cuda and not fluid.core.is_compiled_with_cuda():
         return
 
-    save_path = "word2vec.inference.model"
-    train(use_cuda, is_sparse, save_path)
-    infer(use_cuda, is_sparse, save_path)
+    save_path = "word2vec.params"
+    train(use_cuda, partial(train_program, is_sparse), save_path)
+    infer(use_cuda, partial(inference_program, is_sparse), save_path)
 
 
 if __name__ == '__main__':

python/paddle/fluid/tests/book/test_label_semantic_roles.py

@@ -182,12 +182,6 @@ def train(use_cuda, save_dirname=None, is_local=True):
     crf_decode = fluid.layers.crf_decoding(
         input=feature_out, param_attr=fluid.ParamAttr(name='crfw'))
 
-    chunk_evaluator = fluid.evaluator.ChunkEvaluator(
-        input=crf_decode,
-        label=target,
-        chunk_scheme="IOB",
-        num_chunk_types=int(math.ceil((label_dict_len - 1) / 2.0)))
-
     train_data = paddle.batch(
         paddle.reader.shuffle(
             paddle.dataset.conll05.test(), buf_size=8192),
@@ -203,7 +197,6 @@ def train(use_cuda, save_dirname=None, is_local=True):
 
     def train_loop(main_program):
         exe.run(fluid.default_startup_program())
-
         embedding_param = fluid.global_scope().find_var(
             embedding_name).get_tensor()
         embedding_param.set(
@@ -213,27 +206,19 @@ def train(use_cuda, save_dirname=None, is_local=True):
         start_time = time.time()
         batch_id = 0
         for pass_id in xrange(PASS_NUM):
-            chunk_evaluator.reset(exe)
             for data in train_data():
-                cost, precision, recall, f1_score = exe.run(
-                    main_program,
-                    feed=feeder.feed(data),
-                    fetch_list=[avg_cost] + chunk_evaluator.metrics)
-                pass_precision, pass_recall, pass_f1_score = chunk_evaluator.eval(
-                    exe)
+                cost = exe.run(main_program,
+                               feed=feeder.feed(data),
+                               fetch_list=[avg_cost])
+                cost = cost[0]
 
                 if batch_id % 10 == 0:
-                    print("avg_cost:" + str(cost) + " precision:" + str(
-                        precision) + " recall:" + str(recall) + " f1_score:" +
-                          str(f1_score) + " pass_precision:" + str(
-                              pass_precision) + " pass_recall:" + str(
-                                  pass_recall) + " pass_f1_score:" + str(
-                                      pass_f1_score))
+                    print("avg_cost:" + str(cost))
                     if batch_id != 0:
                         print("second per batch: " + str((time.time(
                         ) - start_time) / batch_id))
                     # Set the threshold low to speed up the CI test
-                    if float(pass_precision) > 0.01:
+                    if float(cost) < 60.0:
                         if save_dirname is not None:
                             # TODO(liuyiqun): Change the target to crf_decode
                             fluid.io.save_inference_model(save_dirname, [

python/paddle/fluid/tests/test_data_feeder.py

@@ -13,15 +13,62 @@
 # limitations under the License.
 
 import paddle.fluid as fluid
+import unittest
 
 
-def test_converter():
-    img = fluid.layers.data(name='image', shape=[1, 28, 28])
-    label = fluid.layers.data(name='label', shape=[1], dtype='int64')
-    feeder = fluid.DataFeeder([img, label], fluid.CPUPlace())
-    result = feeder.feed([[[0] * 784, [9]], [[1] * 784, [1]]])
-    print(result)
+class TestDataFeeder(unittest.TestCase):
+    def test_lod_level_0_converter(self):
+        img = fluid.layers.data(name='image', shape=[1, 28, 28])
+        label = fluid.layers.data(name='label', shape=[1], dtype='int64')
+        feeder = fluid.DataFeeder([img, label], fluid.CPUPlace())
+        result = feeder.feed([([0] * 784, [9]), ([1] * 784, [1])])
+        print(result)
+
+        self.assertEqual(result['image'].shape(), [2, 1, 28, 28])
+        self.assertEqual(result['label'].shape(), [2, 1])
+        self.assertEqual(result['image'].lod(), [])
+        self.assertEqual(result['label'].lod(), [])
+
+    def test_lod_level_1_converter(self):
+        # lod_level = 1
+        # each sentence has a different number of words
+        sentences = fluid.layers.data(
+            name='sentences', shape=[1], dtype='int64', lod_level=1)
+        label = fluid.layers.data(name='label', shape=[1], dtype='int64')
+        feeder = fluid.DataFeeder([sentences, label], fluid.CPUPlace())
+
+        # lod = [[0, 3, 5, 9]]
+        # data = [[1, 2, 3], [4, 5], [6, 7, 8, 9]]
+        # label = [1] * len(data)
+        result = feeder.feed(
+            [([1, 2, 3], [1]), ([4, 5], [1]), ([6, 7, 8, 9], [1])])
+        print(result)
+
+        self.assertEqual(result['sentences'].shape(), [9, 1])
+        self.assertEqual(result['label'].shape(), [3, 1])
+        self.assertEqual(result['sentences'].lod(), [[0, 3, 5, 9]])
+        self.assertEqual(result['label'].lod(), [])
+
+    def test_lod_level_2_converter(self):
+        # lod_level = 2
+        # paragraphs -> sentences -> words
+        paragraphs = fluid.layers.data(
+            name='paragraphs', shape=[1], dtype='int64', lod_level=2)
+        label = fluid.layers.data(name='label', shape=[1], dtype='int64')
+        feeder = fluid.DataFeeder([paragraphs, label], fluid.CPUPlace())
+
+        # lod = [[0, 2, 3], [0, 3, 5, 9]]
+        # data = [[[1, 2, 3], [4, 5]], [[6, 7, 8, 9]]]
+        # label = [1] * len(data)
+        result = feeder.feed(
+            [([[1, 2, 3], [4, 5]], [1]), ([[6, 7, 8, 9]], [1])])
+        print(result)
+
+        self.assertEqual(result['paragraphs'].shape(), [9, 1])
+        self.assertEqual(result['label'].shape(), [2, 1])
+        self.assertEqual(result['paragraphs'].lod(), [[0, 2, 3], [0, 3, 5, 9]])
+        self.assertEqual(result['label'].lod(), [])
 
 
 if __name__ == '__main__':
-    test_converter()
+    unittest.main()

python/paddle/fluid/tests/unittests/CMakeLists.txt

@@ -28,11 +28,11 @@ function(py_test_modules TARGET_NAME)
   if(WITH_TESTING)
     set(options "")
     set(oneValueArgs "")
-    set(multiValueArgs MODULES DEPS ARGS ENVS)
+    set(multiValueArgs MODULES DEPS ENVS)
     cmake_parse_arguments(py_test_modules "${options}" "${oneValueArgs}" "${multiValueArgs}" ${ARGN})
     add_test(NAME ${TARGET_NAME}
              COMMAND env PYTHONPATH=${PADDLE_BINARY_DIR}/python ${py_test_modules_ENVS}
-             ${PYTHON_EXECUTABLE} -u -m unittest --verbose ${py_test_modules_MODULES} ${py_test_modules_ARGS}
+             ${PYTHON_EXECUTABLE} ${PADDLE_SOURCE_DIR}/tools/test_runner.py ${py_test_modules_MODULES}
              WORKING_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR})
   endif()
 endfunction()
@@ -66,6 +66,7 @@ list(REMOVE_ITEM TEST_OPS test_fetch_var)
 list(REMOVE_ITEM TEST_OPS test_parallel_op)
 list(REMOVE_ITEM TEST_OPS test_dynrnn_static_input)
 list(REMOVE_ITEM TEST_OPS test_dist_train)
+list(REMOVE_ITEM TEST_OPS test_network_with_dtype)
 
 # tests that can be bundled together in one python process for speed.
 if(WITH_FAST_BUNDLE_TEST)
@@ -83,6 +84,7 @@ py_test_modules(test_parallel_executor MODULES test_parallel_executor)
 py_test_modules(test_warpctc_op MODULES test_warpctc_op ENVS FLAGS_warpctc_dir=${WARPCTC_LIB_DIR})
 py_test_modules(test_train_dyn_rnn MODULES test_dyn_rnn)
 py_test_modules(test_mul_op MODULES test_mul_op)
+py_test_modules(test_network_with_dtype MODULES test_network_with_dtype)
 
 # tests that need to be run in separate process.
 py_test_modules(test_multihead_attention MODULES test_multihead_attention)

python/paddle/fluid/tests/unittests/test_network_with_dtype.py

@@ -24,33 +24,30 @@ BATCH_SIZE = 20
 
 
 class TestNetWithDtype(unittest.TestCase):
-    def set_network(self):
+    def setUp(self):
         self.dtype = "float64"
         self.init_dtype()
-        main = fluid.Program()
-        with fluid.program_guard(main):
-            self.x = fluid.layers.data(name='x', shape=[13], dtype=self.dtype)
-            self.y = fluid.layers.data(name='y', shape=[1], dtype=self.dtype)
-            y_predict = fluid.layers.fc(input=self.x, size=1, act=None)
 
-            cost = fluid.layers.square_error_cost(input=y_predict, label=self.y)
+    def run_net_on_place(self, place):
+        main = fluid.Program()
+        startup = fluid.Program()
+        with fluid.program_guard(main, startup):
+            x = fluid.layers.data(name='x', shape=[13], dtype=self.dtype)
+            y = fluid.layers.data(name='y', shape=[1], dtype=self.dtype)
+            y_predict = fluid.layers.fc(input=x, size=1, act=None)
+            cost = fluid.layers.square_error_cost(input=y_predict, label=y)
             avg_cost = fluid.layers.mean(cost)
-        self.program = main
-        self.fetch_list = [avg_cost]
+            sgd_optimizer = fluid.optimizer.SGD(learning_rate=0.001)
+            sgd_optimizer.minimize(avg_cost)
 
-        sgd_optimizer = fluid.optimizer.SGD(learning_rate=0.001)
-        sgd_optimizer.minimize(avg_cost)
-
-    def run_net_on_place(self, place):
+        fetch_list = [avg_cost]
         train_reader = paddle.batch(
             paddle.dataset.uci_housing.train(), batch_size=BATCH_SIZE)
-        feeder = fluid.DataFeeder(place=place, feed_list=[self.x, self.y])
+        feeder = fluid.DataFeeder(place=place, feed_list=[x, y])
         exe = fluid.Executor(place)
-        exe.run(fluid.default_startup_program())
+        exe.run(startup)
         for data in train_reader():
-            exe.run(self.program,
-                    feed=feeder.feed(data),
-                    fetch_list=self.fetch_list)
+            exe.run(main, feed=feeder.feed(data), fetch_list=fetch_list)
             # the main program is runable, the datatype is fully supported
             break
 
@@ -58,14 +55,12 @@ class TestNetWithDtype(unittest.TestCase):
         pass
 
     def test_cpu(self):
-        self.set_network()
         place = fluid.CPUPlace()
         self.run_net_on_place(place)
 
     def test_gpu(self):
         if not core.is_compiled_with_cuda():
             return
-        self.set_network()
         place = fluid.CUDAPlace(0)
         self.run_net_on_place(place)
 

python/paddle/fluid/tests/unittests/test_parallel_executor.py

@@ -232,14 +232,18 @@ class TestParallelExecutorBase(unittest.TestCase):
             place = fluid.CUDAPlace(0)
             startup_exe = fluid.Executor(place)
             startup_exe.run(startup)
+            exec_strategy = fluid.ExecutionStrategy()
+            exec_strategy.allow_op_delay = allow_op_delay
+
+            build_strategy = fluid.BuildStrategy()
+            build_strategy.reduce_strategy = fluid.BuildStrategy.ReduceStrategy.Reduce if balance_parameter_opt_between_cards else fluid.BuildStrategy.ReduceStrategy.AllReduce
 
             if use_parallel_executor:
                 exe = fluid.ParallelExecutor(
                     True,
                     loss_name=loss.name,
-                    allow_op_delay=allow_op_delay,
-                    balance_parameter_opt_between_cards=balance_parameter_opt_between_cards
-                )
+                    exec_strategy=exec_strategy,
+                    build_strategy=build_strategy)
             else:
                 exe = fluid.Executor(place=place)
 
@@ -548,7 +552,7 @@ class TestTransformer(TestParallelExecutorBase):
 
 
 class ParallelExecutorTestingDuringTraining(unittest.TestCase):
-    def check_network_convergence(self, balance_parameter_opt_between_cards):
+    def check_network_convergence(self, build_strategy=None):
         main = fluid.Program()
         startup = fluid.Program()
         with fluid.program_guard(main, startup):
@@ -571,15 +575,13 @@ class ParallelExecutorTestingDuringTraining(unittest.TestCase):
                 use_cuda=True,
                 loss_name=loss.name,
                 main_program=main,
-                balance_parameter_opt_between_cards=balance_parameter_opt_between_cards
-            )
+                build_strategy=build_strategy)
 
             test_exe = fluid.ParallelExecutor(
                 use_cuda=True,
                 main_program=test_program,
                 share_vars_from=train_exe,
-                balance_parameter_opt_between_cards=balance_parameter_opt_between_cards
-            )
+                build_strategy=build_strategy)
 
             for i in xrange(5):
                 test_loss, = test_exe.run([loss.name], feed=feed_dict)
@@ -594,10 +596,14 @@ class ParallelExecutorTestingDuringTraining(unittest.TestCase):
                     str(test_loss))
 
     def test_parallel_testing(self):
-        self.check_network_convergence(False)
+        build_strategy = fluid.BuildStrategy()
+        build_strategy.reduce_strategy = fluid.BuildStrategy.ReduceStrategy.AllReduce
+        self.check_network_convergence(build_strategy)
 
     def test_parallel_testing_with_new_strategy(self):
-        self.check_network_convergence(True)
+        build_strategy = fluid.BuildStrategy()
+        build_strategy.reduce_strategy = fluid.BuildStrategy.ReduceStrategy.Reduce
+        self.check_network_convergence(build_strategy)
 
 
 import paddle.dataset.conll05 as conll05
@@ -617,7 +623,7 @@ embedding_name = 'emb'
 
 
 def db_lstm(word, predicate, ctx_n2, ctx_n1, ctx_0, ctx_p1, ctx_p2, mark,
-            is_sparse, balance_parameter_opt_between_cards, **ignored):
+            is_sparse, **ignored):
     # 8 features
     predicate_embedding = fluid.layers.embedding(
         input=predicate,
@@ -686,9 +692,7 @@ def db_lstm(word, predicate, ctx_n2, ctx_n1, ctx_0, ctx_p1, ctx_p2, mark,
 
 
 class TestCRFModel(unittest.TestCase):
-    def check_network_convergence(self,
-                                  is_sparse,
-                                  balance_parameter_opt_between_cards=False):
+    def check_network_convergence(self, is_sparse, build_strategy=None):
         main = fluid.Program()
         startup = fluid.Program()
         with fluid.program_guard(main, startup):
@@ -739,8 +743,7 @@ class TestCRFModel(unittest.TestCase):
             pe = fluid.ParallelExecutor(
                 use_cuda=True,
                 loss_name=avg_cost.name,
-                balance_parameter_opt_between_cards=balance_parameter_opt_between_cards
-            )
+                build_strategy=build_strategy)
 
             feeder = fluid.DataFeeder(
                 feed_list=[
@@ -756,19 +759,29 @@ class TestCRFModel(unittest.TestCase):
                           pe.run(feed=feeder.feed(cur_batch),
                                  fetch_list=[avg_cost.name]))[0]
 
-    def test_update_sparse_parameter(self):
-        self.check_network_convergence(is_sparse=True)
+    def test_update_sparse_parameter_all_reduce(self):
+        build_strategy = fluid.BuildStrategy()
+        build_strategy.reduce_strategy = fluid.BuildStrategy.ReduceStrategy.AllReduce
+        self.check_network_convergence(
+            is_sparse=True, build_strategy=build_strategy)
 
-    def test_update_dense_parameter(self):
-        self.check_network_convergence(is_sparse=False)
+    def test_update_dense_parameter_all_reduce(self):
+        build_strategy = fluid.BuildStrategy()
+        build_strategy.reduce_strategy = fluid.BuildStrategy.ReduceStrategy.AllReduce
+        self.check_network_convergence(
+            is_sparse=False, build_strategy=build_strategy)
 
-    def test_update_sparse_parameter_with_new_strategy(self):
+    def test_update_sparse_parameter_reduce(self):
+        build_strategy = fluid.BuildStrategy()
+        build_strategy.reduce_strategy = fluid.BuildStrategy.ReduceStrategy.Reduce
         self.check_network_convergence(
-            is_sparse=False, balance_parameter_opt_between_cards=True)
+            is_sparse=True, build_strategy=build_strategy)
 
-    def test_update_dense_parameter_with_new_strategy(self):
+    def test_update_dense_parameter_reduce(self):
+        build_strategy = fluid.BuildStrategy()
+        build_strategy.reduce_strategy = fluid.BuildStrategy.ReduceStrategy.Reduce
         self.check_network_convergence(
-            is_sparse=False, balance_parameter_opt_between_cards=True)
+            is_sparse=False, build_strategy=build_strategy)
 
 
 # test fetch all the variables of global_block
@@ -836,7 +849,7 @@ class TestFetchOp(unittest.TestCase):
                     assert not math.isnan(np.sum(ret[i])) and \
                            not math.isinf(np.sum(ret[i]))
 
-    def test_update_sparse_parameter(self):
+    def test_fetch_op(self):
         tst_reader = paddle.batch(flowers.test(use_xmap=False), batch_size=16)
         tst_reader_iter = tst_reader()
 

python/paddle/fluid/tests/unittests/test_split_var.py

@@ -21,15 +21,7 @@ import random
 
 
 class TestSplitVar(unittest.TestCase):
-    def test_check_output(self):
-        # split below shapes to 10 servers
-        shapes = [[3, 5], [1024], [28, 784], [8, 1020], [800, 10]]
-        expected_sizes = [
-            [15], [1024],
-            [2352, 2352, 2352, 2352, 2352, 2352, 2352, 2352, 2352, 784],
-            [2040, 2040, 2040, 2040],
-            [1150, 1150, 1150, 1150, 1150, 1150, 1100]
-        ]
+    def check_split_output(self, shapes, expected_sizes, min_size):
         var_list = []
         program = fluid.Program()
         for shape in shapes:
@@ -39,7 +31,7 @@ class TestSplitVar(unittest.TestCase):
                 # dtype=core.VarDesc.VarType.LOD_TENSOR,
                 shape=shape)
             var_list.append(var)
-        blocks = split_dense_variable(var_list, 10)
+        blocks = split_dense_variable(var_list, 10, min_size)
         all_sizes = []
         for s in expected_sizes:
             for s2 in s:
@@ -48,6 +40,25 @@ class TestSplitVar(unittest.TestCase):
             varname, block_id, size = block_str.split(":")
             self.assertEqual(int(size), all_sizes[i])
 
+    def test_1k(self):
+        shapes = [[3, 5], [1024], [28, 784], [8, 1020], [800, 10]]
+        expected_sizes = [
+            [15], [1024],
+            [2352, 2352, 2352, 2352, 2352, 2352, 2352, 2352, 2352, 784],
+            [2040, 2040, 2040, 2040],
+            [1150, 1150, 1150, 1150, 1150, 1150, 1100]
+        ]
+
+        self.check_split_output(shapes, expected_sizes, 1024)
+
+    def test_check_output_8k(self):
+        shapes = [[3, 5], [1024], [28, 784], [8, 1020], [800, 10],
+                  [6, 33, 33, 33]]
+        expected_sizes = [[15], [1024], [10976, 10976], [8160], [8000],
+                          [35937, 35937, 35937, 35937, 35937, 35937]]
+
+        self.check_split_output(shapes, expected_sizes, 8192)
+
 
 if __name__ == '__main__':
     unittest.main()

python/paddle/fluid/trainer.py

@@ -20,6 +20,7 @@ import data_feeder
 import contextlib
 import io
 import unique_name
+import parallel_executor
 
 # optimizer is same as the parameter of Trainer.__init__. Rename it to opt_module
 import optimizer as opt_module
@@ -48,12 +49,14 @@ class BeginStepEvent(object):
     def __init__(self, epoch_id, step_id):
         self.epoch = epoch_id
         self.step = step_id
+        self.fetch_metrics = True
 
 
 class EndStepEvent(object):
-    def __init__(self, epoch_id, step_id):
+    def __init__(self, epoch_id, step_id, metrics):
         self.epoch = epoch_id
         self.step = step_id
+        self.metrics = metrics
 
 
 def check_and_get_place(place):
@@ -87,24 +90,23 @@ class Trainer(object):
 
     Args:
         train_func(callable): A function which will return loss. The loss must be a scalar.
-        infer_func(callable): A function which will return predict, used to save inference model
         optimizer(optimizer.Optimizer): The optimizer should be an instance of Optimizer
         place: The device place of this trainer.
     """
 
     def __init__(self,
                  train_func,
-                 infer_func,
                  optimizer,
                  param_path=None,
-                 place=None):
+                 place=None,
+                 parallel=False):
+        self.parallel = parallel
         # 1. we need to generate a framework.Program by calling
         # program_func. Reference: fluid.program_guard in
         # test_word2vec.py
         if not isinstance(optimizer, opt_module.Optimizer):
             raise TypeError("The optimizer should be an instance of Optimizer")
 
-        self.infer_func = infer_func
         self.scope = core.Scope()
 
         self.startup_program = framework.Program()
@@ -112,14 +114,14 @@ class Trainer(object):
 
         with framework.program_guard(self.train_program, self.startup_program):
             program_func_outs = train_func()
-            self.test_outputs = program_func_outs if isinstance(
+            self.train_func_outputs = program_func_outs if isinstance(
                 program_func_outs, list) else [program_func_outs]
             self.test_program = self.train_program.clone()
             if not isinstance(optimizer, opt_module.Optimizer):
                 raise TypeError(
                     "The optimizer should be an instance of Optimizer")
             # The fisrt element of program_func_outs is loss.
-            loss = self.test_outputs[0]
+            loss = self.train_func_outputs[0]
             optimize_ops, params_grads = optimizer.minimize(loss)
 
         self.place = check_and_get_place(place)
@@ -137,7 +139,40 @@ class Trainer(object):
             # load params from param_path into scope
             io.load_persistables(exe, dirname=param_path)
 
+    def _transpile_nccl2_dist(self):
+        # PADDLE_TRAINER_IPS
+        if "PADDLE_TRAINER_IPS" not in os.environ:
+            self.nccl_id_var = None
+        else:
+            self.trainer_id = int(os.getenv("PADDLE_TRAINER_ID"))
+            port = os.getenv("PADDLE_PSERVER_PORT")
+            worker_ips = os.getenv("PADDLE_TRAINER_IPS")
+            worker_endpoints = []
+            for ip in worker_ips.split(","):
+                worker_endpoints.append(':'.join([ip, port]))
+            self.num_trainers = len(worker_endpoints)
+            current_endpoint = os.getenv("POD_IP") + ":" + port
+            worker_endpoints.remove(current_endpoint)
+            # TODO(wuyi): use self.nccl_id_var, self.num_trainers and self.trainer_id
+            # in ParallelExecutor to start
+            # distributed training using NCCL2
+            self.nccl_id_var = self.startup_program.global_block().create_var(
+                name="NCCLID", persistable=True, type=core.VarDesc.VarType.RAW)
+            self.startup_program.global_block().append_op(
+                type="gen_nccl_id",
+                inputs={},
+                outputs={"NCCLID": self.nccl_id_var},
+                attrs={
+                    "endpoint": current_endpoint,
+                    "endpoint_list": worker_endpoints,
+                    "trainer_id": self.trainer_id
+                })
+
     def _dist_transpile_if_necessary(self, optimize_ops, params_grads):
+        self._transpile_nccl2_dist()
+        if self.nccl_id_var != None:
+            return
+
         if "PADDLE_TRAINING_ROLE" not in os.environ:
             return
 
@@ -175,12 +210,7 @@ class Trainer(object):
                     'TRAINING_ROLE environment variable must be either TRAINER or PSERVER'
                 )
 
-    def train(self,
-              num_epochs,
-              event_handler,
-              reader=None,
-              parallel=False,
-              feed_order=None):
+    def train(self, num_epochs, event_handler, reader=None, feed_order=None):
         """
         Train the model.
 
@@ -188,27 +218,26 @@ class Trainer(object):
             num_epochs: The number of epoch. An epoch will process all data in reader
             event_handler: The event handler. A function with type (ev:Event)->void
             reader:
-            parallel: True if use multi-CPUs or multi-GPUs
             feed_order: Feeding order of reader. None will following the defining
                 order in program
 
         Returns:
 
         """
-        if parallel:
-            raise NotImplementedError(
-                "Parallel Executor version of trainer is not implemented")
-
         training_role = os.getenv("PADDLE_TRAINING_ROLE", "")
         if training_role == "PSERVER":
             with self._prog_and_scope_guard():
                 exe = executor.Executor(self.place)
                 exe.run()
                 return
+        if self.parallel:
+            self._train_by_parallel_executor(num_epochs, event_handler, reader,
+                                             feed_order)
+        else:
+            self._train_by_executor(num_epochs, event_handler, reader,
+                                    feed_order)
 
-        self._train_by_executor(num_epochs, event_handler, reader, feed_order)
-
-    def test(self, reader, feed_order=None):
+    def test(self, reader, feed_order):
         """
         Test the model on given test data
 
@@ -218,7 +247,8 @@ class Trainer(object):
                 order in program
         """
 
-        return self._test_by_executor(reader, feed_order, self.test_outputs)
+        return self._test_by_executor(reader, feed_order,
+                                      self.train_func_outputs)
 
     def save_params(self, param_path):
         # reference: save_persistables in io.py
@@ -226,15 +256,6 @@ class Trainer(object):
             exe = executor.Executor(self.place)
             io.save_persistables(exe, dirname=param_path)
 
-    def save_inference_model(self, model_path):
-        inference_program = framework.Program()
-        with framework.program_guard(inference_program):
-            with unique_name.guard():
-                predict_var = self.infer_func()
-        predict_var = self.train_program.block(0).var(predict_var.name)
-        exe = executor.Executor(self.place)
-        io.save_inference_model(model_path, [], [predict_var], exe)
-
     @contextlib.contextmanager
     def _prog_and_scope_guard(self):
         with framework.program_guard(
@@ -261,13 +282,25 @@ class Trainer(object):
             feeder = data_feeder.DataFeeder(
                 feed_list=feed_var_list, place=self.place)
             exe = executor.Executor(self.place)
-            for epoch_id in range(num_epochs):
-                event_handler(BeginEpochEvent(epoch_id))
-                for step_id, data in enumerate(reader()):
-                    event_handler(BeginStepEvent(epoch_id, step_id))
-                    exe.run(feed=feeder.feed(data), fetch_list=[])
-                    event_handler(EndStepEvent(epoch_id, step_id))
-                event_handler(EndEpochEvent(epoch_id))
+            reader = feeder.decorate_reader(reader, multi_devices=False)
+            self._train_by_any_executor(event_handler, exe, num_epochs, reader)
+
+    def _train_by_any_executor(self, event_handler, exe, num_epochs, reader):
+        for epoch_id in range(num_epochs):
+            event_handler(BeginEpochEvent(epoch_id))
+            for step_id, data in enumerate(reader()):
+                begin_event = BeginStepEvent(epoch_id, step_id)
+                event_handler(begin_event)
+                if begin_event.fetch_metrics:
+                    metrics = exe.run(feed=data,
+                                      fetch_list=[
+                                          var.name
+                                          for var in self.train_func_outputs
+                                      ])
+                else:
+                    metrics = exe.run(feed=data, fetch_list=[])
+                event_handler(EndStepEvent(epoch_id, step_id, metrics))
+            event_handler(EndEpochEvent(epoch_id))
 
     def _test_by_executor(self, reader, feed_order, fetch_list):
         with executor.scope_guard(self.scope):
@@ -286,17 +319,34 @@ class Trainer(object):
 
             return [x / count for x in accumulated]
 
+    def _train_by_parallel_executor(self, num_epochs, event_handler, reader,
+                                    feed_order):
+        with self._prog_and_scope_guard():
+            pe = self._get_or_create_parallel_executor()
+            feed_var_list = build_feed_var_list(self.train_program, feed_order)
+            feeder = data_feeder.DataFeeder(
+                feed_list=feed_var_list, place=self.place)
+            reader = feeder.decorate_reader(reader, multi_devices=True)
+            for epoch_id in range(num_epochs):
+                self._train_by_any_executor(event_handler, pe, num_epochs,
+                                            reader)
+
+    def _get_parallel_executor(self):
+        return getattr(self, 'parallel_executor', None)
+
+    def _get_or_create_parallel_executor(self):
+        if self._get_parallel_executor() is None:
+            self.parallel_executor = parallel_executor.ParallelExecutor(
+                use_cuda=isinstance(self.place, core.CUDAPlace),
+                loss_name=self.train_func_outputs[0].name)
+        return self._get_parallel_executor()
+
 
 def build_feed_var_list(program, feed_order):
     if not isinstance(program, framework.Program):
         raise TypeError("The 'program' should be an object of Program")
 
-    if feed_order is None:
-        feed_var_list = [
-            var for var in program.global_block().vars.itervalues()
-            if var.is_data
-        ]
-    elif isinstance(feed_order, list):
+    if isinstance(feed_order, list):
         feed_var_list = [
             program.global_block().var(var_name) for var_name in feed_order
         ]

python/paddle/fluid/transpiler/distribute_transpiler.py

@@ -93,30 +93,33 @@ def same_or_split_var(p_name, var_name):
     return p_name == var_name or p_name.startswith(var_name + ".block")
 
 
-def split_dense_variable(var_list,
-                         pserver_count,
-                         min_block_size=1024,
-                         max_block_size=1048576):
+def split_dense_variable(var_list, service_count, min_block_size=8192):
     """
-        We may need to split dense tensor to one or more blocks and put
-        them equally onto parameter server. One block is a sub-tensor
-        aligned by dim[0] of the tensor.
-
-        We need to have a minimal block size so that the calculations in
-        the parameter server side can gain better performance. By default
-        minimum block size is 1024. The max block size is used to prevent
-        very large blocks that may cause send error.
-        :return: A list of VarBlocks. Each VarBlock specifies a shard of
-           the var.
+    We may need to split dense tensor to one or more blocks and put
+    them equally onto parameter server. One block is a sub-tensor
+    aligned by dim[0] of the tensor.
+
+    We need to have a minimal block size so that the calculations in
+    the parameter server side can gain better performance. By default
+    minimum block size 8K elements (maybe 16bit or 32bit or 64bit). 
+
+    Args:
+        var_list (list): List of variables.
+        service_count (int): Numel of pserver services. A pserver may have two
+            or more listening ports.
+        min_block_size (int): Minimum splitted block size.
+    Returns:
+        blocks (list[(varname, block_id, current_block_size)]): A list 
+            of VarBlocks. Each VarBlock specifies a shard of the var.
     """
     blocks = []
     for var in var_list:
-        split_count = pserver_count
+        split_count = service_count
         var_numel = reduce(lambda x, y: x * y, var.shape)
         max_pserver_count = int(math.floor(var_numel / float(min_block_size)))
         if max_pserver_count == 0:
             max_pserver_count = 1
-        if max_pserver_count < pserver_count:
+        if max_pserver_count < service_count:
             split_count = max_pserver_count
         block_size = int(math.ceil(var_numel / float(split_count)))
 
@@ -803,9 +806,18 @@ class DistributeTranspiler:
         Create vars for each split.
         NOTE: only grads need to be named for different trainers, use
               add_trainer_suffix to rename the grad vars.
-        :return: A dict mapping from original var name to each var split.
+        Args:
+            program (ProgramDesc): ProgramDesc which gradients blong.
+            block_list (list[(varname, block_id, block_size)]): List of gradient blocks.
+            add_trainer_suffix (Bool): Add trainer suffix to new variable's name if set True.
+        Returns: 
+            var_mapping (dict(varname->[new_varname_variable])):A dict mapping 
+                from original var name to each var split.
         """
+
+        # varname->[(block_id, current_block_size)]
         block_map = dict()
+
         var_mapping = dict()
         for block_str in block_list:
             varname, offset, size = block_str.split(":")
@@ -1190,6 +1202,12 @@ class DistributeTranspiler:
         return lr_ops
 
     def _get_optimize_pass(self):
+        """
+        Get optimizer operators, paramters and gradients from origin_program
+        Returns:
+            opt_ops (list): optimize operators.
+            params_grads (dict): paramter->gradient.
+        """
         block = self.origin_program.global_block()
         opt_ops = []
         params_grads = []

tools/test_runner.py

+# Copyright (c) 2018 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.
+
+import unittest
+import os
+import sys
+import paddle.fluid as fluid
+import importlib
+import cStringIO
+
+
+def main():
+    sys.path.append(os.getcwd())
+    some_test_failed = False
+    for module_name in sys.argv[1:]:
+        buffer = cStringIO.StringIO()
+        main = fluid.Program()
+        startup = fluid.Program()
+        scope = fluid.core.Scope()
+        with fluid.program_guard(main, startup):
+            with fluid.scope_guard(scope):
+                with fluid.unique_name.guard():
+                    test_loader = unittest.TestLoader()
+                    module = importlib.import_module(module_name)
+                    tests = test_loader.loadTestsFromModule(module)
+                    res = unittest.TextTestRunner(stream=buffer).run(tests)
+                    if not res.wasSuccessful():
+                        some_test_failed = True
+                        print >> sys.stderr, module_name, 'failed\n', buffer.getvalue(
+                        )
+
+    if some_test_failed:
+        exit(1)
+
+
+if __name__ == '__main__':
+    main()