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

.clang-format

@@ -19,7 +19,7 @@ BasedOnStyle:  Google
 IndentWidth:     2
 TabWidth:        2
 ContinuationIndentWidth: 4
-AccessModifierOffset: -2  # The private/protected/public has no indent in class
+AccessModifierOffset: -1  # The private/protected/public has no indent in class
 Standard:  Cpp11 
 AllowAllParametersOfDeclarationOnNextLine: true
 BinPackParameters: false

.pre-commit-config.yaml

@@ -34,6 +34,14 @@ repos:
         entry: bash ./tools/codestyle/cpplint_pre_commit.hook
         language: system
         files: \.(c|cc|cxx|cpp|cu|h|hpp|hxx)$
+-   repo: local
+    hooks:
+    -   id: pylint-doc-string
+        name: pylint
+        description: Check python docstring style using docstring_checker.
+        entry: bash ./tools/codestyle/pylint_pre_commit.hook
+        language: system
+        files: \.(py)$
 -   repo: https://github.com/PaddlePaddle/pre-commit-golang
     sha: 8337620115c25ff8333f1b1a493bd031049bd7c0
     hooks:

.travis.yml

@@ -18,6 +18,8 @@ env:
 addons:
   ssh_known_hosts: 13.229.163.131
 before_install:
+  # For pylint dockstring checker
+  - sudo pip install pylint pytest astroid isort
   - |
     function timeout() { perl -e 'alarm shift; exec @ARGV' "$@"; }
 script:

Dockerfile

@@ -79,6 +79,9 @@ RUN pip install pre-commit 'ipython==5.3.0' && \
     pip install 'ipykernel==4.6.0' 'jupyter==1.0.0' && \
     pip install opencv-python
 
+#For docstring checker
+RUN pip install pylint pytest astroid isort
+
 COPY ./python/requirements.txt /root/
 RUN pip install -r /root/requirements.txt
 

benchmark/fluid/README.md

@@ -24,22 +24,22 @@ Currently supported `--model` argument include:
 
 * Run the following command to start a benchmark job locally:
     ```bash
-      python fluid_benchmark.py --model mnist --parallel 1 --device GPU --with_test
+      python fluid_benchmark.py --model mnist  --device GPU
     ```
     You can choose to use GPU/CPU training. With GPU training, you can specify
-    `--parallel 1` to run multi GPU training.
+    `--gpus <gpu_num>` to run multi GPU training.
 * Run distributed training with parameter servers:
     * start parameter servers:
         ```bash
-        PADDLE_TRAINING_ROLE=PSERVER PADDLE_PSERVER_PORT=7164 PADDLE_PSERVER_IPS=127.0.0.1 PADDLE_TRAINERS=1 PADDLE_CURRENT_IP=127.0.0.1 PADDLE_TRAINER_ID=0 python fluid_benchmark.py --model mnist --parallel 0 --device GPU --update_method pserver
+        PADDLE_TRAINING_ROLE=PSERVER PADDLE_PSERVER_PORT=7164 PADDLE_PSERVER_IPS=127.0.0.1 PADDLE_TRAINERS=1 PADDLE_CURRENT_IP=127.0.0.1 PADDLE_TRAINER_ID=0 python fluid_benchmark.py --model mnist  --device GPU --update_method pserver
         ```
     * start trainers:
         ```bash
-        PADDLE_TRAINING_ROLE=PSERVER PADDLE_PSERVER_PORT=7164 PADDLE_PSERVER_IPS=127.0.0.1 PADDLE_TRAINERS=1 PADDLE_CURRENT_IP=127.0.0.1 PADDLE_TRAINER_ID=0 python fluid_benchmark.py --model mnist --parallel 0 --device GPU --update_method pserver
+        PADDLE_TRAINING_ROLE=TRAINER PADDLE_PSERVER_PORT=7164 PADDLE_PSERVER_IPS=127.0.0.1 PADDLE_TRAINERS=1 PADDLE_CURRENT_IP=127.0.0.1 PADDLE_TRAINER_ID=0 python fluid_benchmark.py --model mnist  --device GPU --update_method pserver
         ```
 * Run distributed training using NCCL2
     ```bash
-    PADDLE_PSERVER_PORT=7164 PADDLE_TRAINER_IPS=192.168.0.2,192.168.0.3  PADDLE_CURRENT_IP=127.0.0.1 PADDLE_TRAINER_ID=0 python fluid_benchmark.py --model mnist --parallel 0 --device GPU --update_method nccl2
+    PADDLE_PSERVER_PORT=7164 PADDLE_TRAINER_IPS=192.168.0.2,192.168.0.3  PADDLE_CURRENT_IP=127.0.0.1 PADDLE_TRAINER_ID=0 python fluid_benchmark.py --model mnist --device GPU --update_method nccl2
     ```
 
 ## Run Distributed Benchmark on Kubernetes Cluster
@@ -48,7 +48,7 @@ We provide a script `kube_gen_job.py` to generate Kubernetes yaml files to submi
 distributed benchmark jobs to your cluster. To generate a job yaml, just run:
 
 ```bash
-python kube_gen_job.py --jobname myjob --pscpu 4 --cpu 8 --gpu 8 --psmemory 20 --memory 40 --pservers 4 --trainers 4 --entry "python fluid_benchmark.py --model mnist --parallel 1 --device GPU --update_method pserver --with_test" --disttype pserver
+python kube_gen_job.py --jobname myjob --pscpu 4 --cpu 8 --gpu 8 --psmemory 20 --memory 40 --pservers 4 --trainers 4 --entry "python fluid_benchmark.py --model mnist --parallel 1 --device GPU --update_method pserver " --disttype pserver
 ```
 
 Then the yaml files are generated under directory `myjob`, you can run:
@@ -58,3 +58,14 @@ kubectl create -f myjob/
 ```
 
 The job shall start.
+
+
+## Notes for Run Fluid Distributed with NCCL2 and RDMA
+
+Before running NCCL2 distributed jobs, please check that whether your node has multiple network
+interfaces, try to add the environment variable `export NCCL_SOCKET_IFNAME=eth0` to use your actual
+network device.
+
+To run high-performance distributed training, you must prepare your hardware environment to be
+able to run RDMA enabled network communication, please check out [this](https://github.com/PaddlePaddle/Paddle/blob/develop/doc/fluid/howto/cluster/nccl2_rdma_training.md)
+note for details.

cmake/inference_lib.cmake

@@ -56,24 +56,28 @@ set(dst_dir "${FLUID_INSTALL_DIR}/third_party/eigen3")
 copy(eigen3_lib
   SRCS ${EIGEN_INCLUDE_DIR}/Eigen/Core ${EIGEN_INCLUDE_DIR}/Eigen/src ${EIGEN_INCLUDE_DIR}/unsupported/Eigen
   DSTS ${dst_dir}/Eigen ${dst_dir}/Eigen ${dst_dir}/unsupported
+  DEPS eigen3
 )
 
 set(dst_dir "${FLUID_INSTALL_DIR}/third_party/install/gflags")
 copy(gflags_lib
   SRCS ${GFLAGS_INCLUDE_DIR} ${GFLAGS_LIBRARIES}
   DSTS ${dst_dir} ${dst_dir}/lib
+  DEPS gflags
 )
 
 set(dst_dir "${FLUID_INSTALL_DIR}/third_party/install/glog")
 copy(glog_lib
   SRCS ${GLOG_INCLUDE_DIR} ${GLOG_LIBRARIES}
   DSTS ${dst_dir} ${dst_dir}/lib
+  DEPS glog
 )
 
 set(dst_dir "${FLUID_INSTALL_DIR}/third_party/boost/")
 copy(boost_lib
   SRCS ${BOOST_INCLUDE_DIR}/boost
   DSTS ${dst_dir}
+  DEPS boost
 )
 
 if(NOT PROTOBUF_FOUND)
@@ -81,6 +85,7 @@ if(NOT PROTOBUF_FOUND)
     copy(protobuf_lib
       SRCS ${PROTOBUF_INCLUDE_DIR} ${PROTOBUF_LIBRARY}
       DSTS ${dst_dir} ${dst_dir}/lib
+      DEPS extern_protobuf
     )
 endif()
 
@@ -89,12 +94,14 @@ if(NOT CBLAS_FOUND)
     copy(openblas_lib
       SRCS ${CBLAS_INSTALL_DIR}/lib ${CBLAS_INSTALL_DIR}/include
       DSTS ${dst_dir} ${dst_dir}
+      DEPS extern_openblas
     )
 elseif (WITH_MKLML)
     set(dst_dir "${FLUID_INSTALL_DIR}/third_party/install/mklml")
     copy(mklml_lib
       SRCS ${MKLML_LIB} ${MKLML_IOMP_LIB} ${MKLML_INC_DIR}
       DSTS ${dst_dir}/lib ${dst_dir}/lib ${dst_dir}
+      DEPS mklml
     )
 endif()
 
@@ -103,6 +110,7 @@ if(WITH_MKLDNN)
   copy(mkldnn_lib
     SRCS ${MKLDNN_INC_DIR} ${MKLDNN_SHARED_LIB}
     DSTS ${dst_dir} ${dst_dir}/lib
+    DEPS mkldnn
   )
 endif()
 
@@ -110,17 +118,20 @@ if(NOT MOBILE_INFERENCE AND NOT RPI)
   set(dst_dir "${FLUID_INSTALL_DIR}/third_party/install/snappy")
   copy(snappy_lib
     SRCS ${SNAPPY_INCLUDE_DIR} ${SNAPPY_LIBRARIES}
-    DSTS ${dst_dir} ${dst_dir}/lib)
+    DSTS ${dst_dir} ${dst_dir}/lib
+    DEPS snappy)
 
   set(dst_dir "${FLUID_INSTALL_DIR}/third_party/install/snappystream")
   copy(snappystream_lib
     SRCS ${SNAPPYSTREAM_INCLUDE_DIR} ${SNAPPYSTREAM_LIBRARIES}
-    DSTS ${dst_dir} ${dst_dir}/lib)
+    DSTS ${dst_dir} ${dst_dir}/lib
+    DEPS snappystream)
 
   set(dst_dir "${FLUID_INSTALL_DIR}/third_party/install/zlib")
   copy(zlib_lib
     SRCS ${ZLIB_INCLUDE_DIR} ${ZLIB_LIBRARIES}
-    DSTS ${dst_dir} ${dst_dir}/lib)
+    DSTS ${dst_dir} ${dst_dir}/lib
+    DEPS zlib)
 endif()
 
 # paddle fluid module

paddle/api/GradientMachine.cpp

@@ -94,7 +94,7 @@ void UpdateCallback::apply(Parameter* p) {
 }
 
 class UpdateCallbackWrapper {
-public:
+ public:
   explicit UpdateCallbackWrapper(const UpdateCallback& callback)
       : callback(const_cast<UpdateCallback&>(callback)) {}
 
@@ -105,7 +105,7 @@ public:
     delete p;
   }
 
-private:
+ private:
   UpdateCallback& callback;
 };
 

paddle/api/PaddleAPI.h

@@ -59,9 +59,10 @@ class RangeError {};
 
 /// Not support Error, such as access GPU memory directly, etc.
 class UnsupportError : public std::runtime_error {
-public:
-  UnsupportError() : std::runtime_error(" "){};
-  UnsupportError(const std::string& message) : std::runtime_error(message){};
+ public:
+  UnsupportError() : std::runtime_error(" ") {}
+  explicit UnsupportError(const std::string& message)
+      : std::runtime_error(message) {}
 };
 
 /// This type will map to python's list of float.
@@ -105,7 +106,7 @@ class Matrix {
   DISABLE_COPY(Matrix);
   static Matrix* createByPaddleMatrixPtr(void* sharedPtr);
 
-public:
+ public:
   virtual ~Matrix();
 
   /**
@@ -231,7 +232,7 @@ public:
 
   bool isGpu() const;
 
-private:
+ private:
   void* getSharedPtr() const;
 
   MatrixPrivate* m;
@@ -248,7 +249,7 @@ class Vector {
 
   void* getSharedPtr();
 
-public:
+ public:
   ~Vector();
 
   /// Create Vector filled with zero.
@@ -310,10 +311,10 @@ public:
   /// __len__ in python
   size_t getSize() const;
 
-private:
+ private:
   VectorPrivate* m;
 
-private:
+ private:
   friend class Parameter;
   friend class ParameterOptimizer;
   friend struct ParameterTraverseCallbackPrivate;
@@ -325,7 +326,7 @@ class IVector {
   DISABLE_COPY(IVector);
   static IVector* createByPaddleVectorPtr(void* ptr);
 
-public:
+ public:
   /// Create IVector filled with zero
   static IVector* createZero(size_t sz, bool useGpu = isUsingGpu());
 
@@ -389,7 +390,7 @@ public:
   /// This method will map to python __len__();
   size_t getSize() const;
 
-private:
+ private:
   void* getSharedPtr() const;
 
   friend class Arguments;
@@ -400,11 +401,11 @@ struct ArgumentsPrivate;
 
 /// The Arguments is actual a std::vector<paddle::Argument> in paddle.
 class Arguments {
-private:
+ private:
   Arguments();  // Internal Create.
   DISABLE_COPY(Arguments);
 
-public:
+ public:
   /**
    * Create a arguments with size.
    * Note that it can be zero.
@@ -475,12 +476,12 @@ public:
 
   float sum() const;
 
-private:
+ private:
   static Arguments* createByPaddleArgumentVector(void* ptr);
   static Arguments* createByPaddleArgument(const void* ptr);
   void* getInternalArgumentsPtr() const;
 
-private:
+ private:
   ArgumentsPrivate* m;
   friend class Trainer;
   friend class GradientMachine;
@@ -507,7 +508,7 @@ class ParameterConfig {
   static ParameterConfig* createParameterConfigFromParameterPtr(void* ptr);
   void* getRawPtr();
 
-public:
+ public:
   ~ParameterConfig();
 
   /**
@@ -515,10 +516,10 @@ public:
    */
   std::string toProtoString() const;
 
-private:
+ private:
   ParameterConfigPrivate* m;
 
-private:
+ private:
   friend class Parameter;
   friend class ParameterOptimizer;
   friend struct ParameterTraverseCallbackPrivate;
@@ -529,7 +530,7 @@ class OptimizationConfig {
   DISABLE_COPY(OptimizationConfig);
   OptimizationConfig();
 
-public:
+ public:
   static OptimizationConfig* createFromProtoString(const std::string& str);
   ~OptimizationConfig();
 
@@ -538,7 +539,7 @@ public:
    */
   std::string toProtoString();
 
-private:
+ private:
   OptimizationConfigPrivate* m;
 
   friend class TrainerConfig;
@@ -549,11 +550,11 @@ private:
 
 struct ParameterPrivate;
 class Parameter {
-private:
+ private:
   Parameter();
   DISABLE_COPY(Parameter);
 
-public:
+ public:
   virtual ~Parameter();
 
   /**
@@ -580,11 +581,11 @@ public:
 
   size_t getSize() const;
 
-private:
+ private:
   static Parameter* createFromRawPtr(void* ptr);
   static Parameter* createFromSharedPtr(void* ptr);
 
-private:
+ private:
   ParameterPrivate* m;
   friend class UpdateCallbackWrapper;
   friend class GradientMachine;
@@ -598,14 +599,14 @@ struct ModelConfigPrivate;
  * It is used by GradientMachine.
  */
 class ModelConfig {
-private:
+ private:
   ModelConfig();
   DISABLE_COPY(ModelConfig);
 
-public:
+ public:
   virtual ~ModelConfig();
 
-private:
+ private:
   ModelConfigPrivate* m;
   friend class TrainerConfig;
   friend struct TrainerConfigPrivate;
@@ -619,11 +620,11 @@ struct TrainerConfigPrivate;
  * It is used by GradientMachine.
  */
 class TrainerConfig {
-private:
+ private:
   TrainerConfig();
   DISABLE_COPY(TrainerConfig);
 
-public:
+ public:
   virtual ~TrainerConfig();
 
   static TrainerConfig* createFromTrainerConfigFile(
@@ -634,7 +635,7 @@ public:
 
   OptimizationConfig* getOptimizationConfig() const;
 
-private:
+ private:
   TrainerConfigPrivate* m;
   friend class Trainer;
 };
@@ -654,7 +655,7 @@ private:
  * @endcode
  */
 class UpdateCallback {
-public:
+ public:
   virtual ~UpdateCallback();
   virtual void apply(Parameter* p);
 };
@@ -664,14 +665,14 @@ class ParameterTraverseCallback {
   DISABLE_COPY(ParameterTraverseCallback);
   ParameterTraverseCallback();
 
-public:
+ public:
   ~ParameterTraverseCallback();
 
   void apply(const std::vector<Vector*>& vecs,
              const ParameterConfig& config,
              size_t sparseId);
 
-private:
+ private:
   ParameterTraverseCallbackPrivate* m;
   friend class ParameterOptimizer;
 };
@@ -686,7 +687,7 @@ class ParameterOptimizer {
   DISABLE_COPY(ParameterOptimizer);
   ParameterOptimizer();
 
-public:
+ public:
   static ParameterOptimizer* create(OptimizationConfig* config);
 
   ~ParameterOptimizer();
@@ -710,7 +711,7 @@ public:
   ParameterTraverseCallback* needSpecialTraversal(
       const ParameterConfig& config) const;
 
-private:
+ private:
   ParameterOptimizerPrivate* m;
 };
 
@@ -718,11 +719,11 @@ class SequenceGenerator;
 class Evaluator;
 struct GradientMachinePrivate;
 class GradientMachine {
-private:
+ private:
   GradientMachine();
   DISABLE_COPY(GradientMachine);
 
-public:
+ public:
   virtual ~GradientMachine();
 
   /**
@@ -817,7 +818,7 @@ public:
 
   void eval(Evaluator* evaluator);
 
-private:
+ private:
   GradientMachinePrivate* m;
 
   static GradientMachine* createFromPaddleModelPtr(
@@ -833,10 +834,10 @@ private:
 
 struct ParameterUpdaterPrivate;
 class ParameterUpdater {
-private:
+ private:
   ParameterUpdater();
 
-public:
+ public:
   static ParameterUpdater* createLocalUpdater(OptimizationConfig* config);
   static ParameterUpdater* createRemoteUpdater(OptimizationConfig* config,
                                                int passCount,
@@ -911,17 +912,17 @@ public:
    */
   void catchUpWith();
 
-private:
+ private:
   ParameterUpdaterPrivate* m;
 };
 
 struct EvaluatorPrivate;
 class Evaluator {
-private:
+ private:
   Evaluator();
   DISABLE_COPY(Evaluator);
 
-public:
+ public:
   ~Evaluator();
 
   /**
@@ -945,7 +946,7 @@ public:
 
   double getValue(const std::string name) const;
 
-private:
+ private:
   EvaluatorPrivate* m;
 
   friend class GradientMachine;
@@ -953,13 +954,13 @@ private:
 
 struct TrainerPrivate;
 class Trainer {
-private:
+ private:
   TrainerPrivate* m;
   Trainer();
   Trainer(TrainerConfig* optConfig, GradientMachine* gm);
   DISABLE_COPY(Trainer);
 
-public:
+ public:
   virtual ~Trainer();
 
   /// Create A Trainer By TrainerConfig. using paddle command line.
@@ -1002,7 +1003,7 @@ public:
 
 /// the N-Best results generated from one input sequence.
 class ISequenceResults {
-public:
+ public:
   virtual ~ISequenceResults();
 
   /// Number of result.
@@ -1026,7 +1027,7 @@ class SequenceGenerator {
   DISABLE_COPY(SequenceGenerator);
   SequenceGenerator();
 
-public:
+ public:
   virtual ~SequenceGenerator();
 
   /**
@@ -1044,10 +1045,10 @@ public:
   void setMaxLength(size_t maxlength);
   void setBeamSize(size_t beamSize);
 
-private:
+ private:
   static SequenceGenerator* createByGradientMachineSharedPtr(void* ptr);
   friend class GradientMachine;
 
-private:
+ private:
   SequenceGeneratorPrivate* m;
 };

paddle/api/SequenceGenerator.cpp

@@ -138,7 +138,7 @@ struct SequenceGeneratorPrivate {
         maxLength(0UL),
         feedback(__create_feedback__()) {}
 
-private:
+ private:
   static paddle::Argument __create_feedback__() {
     paddle::Argument feedback;
     feedback.ids = paddle::IVector::create(/* size= */ 1, FLAGS_use_gpu);
@@ -157,7 +157,7 @@ SequenceGenerator::~SequenceGenerator() { delete m; }
 
 class PathSequenceResults : public ISequenceResults {
   // ISequenceResults interface
-public:
+ public:
   PathSequenceResults(const std::shared_ptr<std::vector<Path>>& path,
                       const std::shared_ptr<std::vector<std::string>>& dict)
       : path_(path), dict_(dict) {}
@@ -196,7 +196,7 @@ public:
     }
   }
 
-private:
+ private:
   std::shared_ptr<std::vector<Path>> path_;
   std::shared_ptr<std::vector<std::string>> dict_;
 };

paddle/capi/gradient_machine.cpp

@@ -26,7 +26,7 @@ enum GradientMatchineCreateMode {
 namespace paddle {
 
 class MyNeuralNetwork : public NeuralNetwork {
-public:
+ public:
   MyNeuralNetwork(const std::string& name, NeuralNetwork* network)
       : NeuralNetwork(name, network) {}
 };

paddle/contrib/inference/paddle_inference_api.h

@@ -50,7 +50,7 @@ struct PaddleTensor {
 * TODO(Superjomn) Prepare another API for NLP-related usages.
 */
 class PaddlePredictor {
-public:
+ public:
   struct Config;
   PaddlePredictor() = default;
   PaddlePredictor(const PaddlePredictor&) = delete;
@@ -66,6 +66,7 @@ public:
   // be thread-safe.
   virtual std::unique_ptr<PaddlePredictor> Clone() = 0;
 
+  virtual bool InitShared() { return false; }
   // Destroy the Predictor.
   virtual ~PaddlePredictor() {}
 

paddle/contrib/inference/paddle_inference_api_impl.cc

@@ -28,7 +28,7 @@ namespace {
 
 // Timer for timer
 class Timer {
-public:
+ public:
   double start;
   double startu;
   void tic() {
@@ -135,16 +135,17 @@ bool PaddlePredictorImpl::Run(const std::vector<PaddleTensor> &inputs,
 
 std::unique_ptr<PaddlePredictor> PaddlePredictorImpl::Clone() {
   VLOG(3) << "Predictor::clone";
-  std::unique_ptr<PaddlePredictorImpl> cls(new PaddlePredictorImpl(config_));
-  if (!cls->InitShared(this)) {
+  std::unique_ptr<PaddlePredictor> cls(new PaddlePredictorImpl(config_));
+  if (!cls->InitShared()) {
     LOG(ERROR) << "fail to call InitShared";
     return nullptr;
   }
-  return cls;
+  // fix manylinux compile error.
+  return std::move(cls);
 }
 
 // TODO(panyx0718): Consider merge with Init()?
-bool PaddlePredictorImpl::InitShared(PaddlePredictorImpl *cls) {
+bool PaddlePredictorImpl::InitShared() {
   VLOG(3) << "Predictor::init_shared";
   // 1. Define place, executor, scope
   if (this->config_.device >= 0) {

paddle/contrib/inference/paddle_inference_api_impl.h

@@ -41,7 +41,7 @@ struct VisConfig : public PaddlePredictor::Config {
  * Do not use this, just a demo indicating how to customize a Predictor.
  */
 class PaddlePredictorImpl : public PaddlePredictor {
-public:
+ public:
   explicit PaddlePredictorImpl(const VisConfig &config) : config_(config) {}
 
   bool Init();
@@ -53,8 +53,8 @@ public:
 
   ~PaddlePredictorImpl() override{};
 
-private:
-  bool InitShared(PaddlePredictorImpl *cls);
+ private:
+  bool InitShared() override;
   bool SetFeed(const std::vector<PaddleTensor> &input_datas,
                std::vector<paddle::framework::LoDTensor> *feeds);
   bool GetFetch(const std::vector<paddle::framework::LoDTensor> &fetchs,

paddle/contrib/inference/test_paddle_inference_api.cc

@@ -31,7 +31,7 @@ struct DemoConfig : public PaddlePredictor::Config {
  * Do not use this, just a demo indicating how to customize a Predictor.
  */
 class DemoPredictor : public PaddlePredictor {
-public:
+ public:
   explicit DemoPredictor(const DemoConfig &config) {
     LOG(INFO) << "I get other_config " << config.other_config;
   }

paddle/contrib/inference/test_paddle_inference_api_impl.cc

@@ -44,7 +44,7 @@ TEST(paddle_inference_api_impl, word2vec) {
   VisConfig config;
   config.model_dir = FLAGS_dirname + "word2vec.inference.model";
   LOG(INFO) << "dirname  " << config.model_dir;
-  config.fraction_of_gpu_memory = 0.85;
+  config.fraction_of_gpu_memory = 0.15;
   config.device = 0;
   config.share_variables = true;
 
@@ -68,11 +68,11 @@ TEST(paddle_inference_api_impl, word2vec) {
 
   std::vector<PaddleTensor> outputs;
   ASSERT_TRUE(predictor->Run(cpu_feeds, &outputs));
-  ASSERT_EQ(outputs.size(), 1);
+  ASSERT_EQ(outputs.size(), 1UL);
   for (size_t i = 0; i < outputs.size(); ++i) {
     size_t len = outputs[i].data.length;
     float* data = static_cast<float*>(outputs[i].data.data);
-    for (int j = 0; j < len / sizeof(float); ++j) {
+    for (size_t j = 0; j < len / sizeof(float); ++j) {
       ASSERT_LT(data[j], 1.0);
       ASSERT_GT(data[j], -1.0);
     }

paddle/cuda/include/hl_activation_functions.h

@@ -31,7 +31,7 @@ namespace hppl {
  */
 template <class T>
 class Active {
-public:
+ public:
   typedef T (*forward)(T);
   typedef T (*backward)(T, T);
 };

paddle/cuda/include/hl_tensor_ops.h

@@ -23,128 +23,128 @@ namespace unary {
 
 template <class T>
 class add_scale {
-private:
+ private:
   const T p;
 
-public:
+ public:
   INLINE add_scale(const T s) : p(s) {}
   INLINE T operator()(const T a) const { return a + p; }
 };
 
 template <class T>
 class sub_scale {
-private:
+ private:
   const T p;
 
-public:
+ public:
   INLINE sub_scale(const T s) : p(s) {}
   INLINE T operator()(const T a) const { return a - p; }
 };
 
 template <class T>
 class mul_scale {
-private:
+ private:
   const T p;
 
-public:
+ public:
   INLINE mul_scale(const T s) : p(s) {}
   INLINE T operator()(const T a) const { return a * p; }
 };
 
 template <class T>
 class div_scale {
-private:
+ private:
   const T p;
 
-public:
+ public:
   INLINE div_scale(const T s) : p(s) {}
   INLINE T operator()(const T a) const { return a / p; }
 };
 
 template <class T>
 class neg {
-public:
+ public:
   INLINE T operator()(const T a) const { return -a; }
 };
 
 template <class T>
 class exp_op {
-public:
+ public:
   INLINE T operator()(const T a) const { return std::exp(a); }
 };
 
 template <class T>
 class log_op {
-public:
+ public:
   INLINE T operator()(const T a) const { return std::log(a); }
 };
 
 template <class T>
 class sqrt_op {
-public:
+ public:
   INLINE T operator()(const T a) const { return std::sqrt(a); }
 };
 
 template <class T>
 class square {
-public:
+ public:
   INLINE T operator()(const T a) const { return a * a; }
 };
 
 template <class T>
 class reciprocal {
-public:
+ public:
   INLINE T operator()(const T a) const { return T(1) / a; }
 };
 
 template <class T>
 class abs {
-public:
+ public:
   INLINE T operator()(const T a) const { return a > 0 ? a : -a; }
 };
 
 template <class T>
 class sign {
-public:
+ public:
   INLINE T operator()(const T a) const { return (a > 0) - (a < 0); }
 };
 
 template <class T>
 class min {
-private:
+ private:
   const T p;
 
-public:
+ public:
   INLINE min(const T s) : p(s) {}
   INLINE T operator()(const T a) const { return a > p ? p : a; }
 };
 
 template <class T>
 class max {
-private:
+ private:
   const T p;
 
-public:
+ public:
   INLINE max(const T s) : p(s) {}
   INLINE T operator()(const T a) const { return a < p ? p : a; }
 };
 
 template <class T>
 class pow_op {
-private:
+ private:
   const T p;
 
-public:
+ public:
   INLINE pow_op(const T s) : p(s) {}
   INLINE T operator()(const T a) const { return std::pow(a, p); }
 };
 
 template <class T>
 class constant {
-private:
+ private:
   const T p;
 
-public:
+ public:
   INLINE constant(const T s) : p(s) {}
   INLINE T operator()(int i) const { return p; }
   INLINE T operator()(int i, int j) const { return p; }
@@ -152,80 +152,80 @@ public:
 
 template <class T>
 class cmp_eq {
-private:
+ private:
   const T p;
 
-public:
+ public:
   INLINE cmp_eq(const T s) : p(s) {}
   INLINE bool operator()(const T a) const { return a == p; }
 };
 
 template <class T>
 class cmp_ne {
-private:
+ private:
   const T p;
 
-public:
+ public:
   INLINE cmp_ne(const T s) : p(s) {}
   INLINE bool operator()(const T a) const { return a != p; }
 };
 
 template <class T>
 class cmp_le {
-private:
+ private:
   const T p;
 
-public:
+ public:
   INLINE cmp_le(const T s) : p(s) {}
   INLINE bool operator()(const T a) const { return a <= p; }
 };
 
 template <class T>
 class cmp_lt {
-private:
+ private:
   const T p;
 
-public:
+ public:
   INLINE cmp_lt(const T s) : p(s) {}
   INLINE bool operator()(const T a) const { return a < p; }
 };
 
 template <class T>
 class cmp_ge {
-private:
+ private:
   const T p;
 
-public:
+ public:
   INLINE cmp_ge(const T s) : p(s) {}
   INLINE bool operator()(const T a) const { return a >= p; }
 };
 
 template <class T>
 class cmp_gt {
-private:
+ private:
   const T p;
 
-public:
+ public:
   INLINE cmp_gt(const T s) : p(s) {}
   INLINE bool operator()(const T a) const { return a > p; }
 };
 
 template <class T>
 class and_op {
-private:
+ private:
   const T p;
 
-public:
+ public:
   INLINE and_op(const T s) : p(s) {}
   INLINE bool operator()(const T a) const { return a && p; }
 };
 
 template <class T>
 class or_op {
-private:
+ private:
   const T p;
 
-public:
+ public:
   INLINE or_op(const T s) : p(s) {}
   INLINE bool operator()(const T a) const { return a || p; }
 };
@@ -235,96 +235,96 @@ public:
 namespace binary {
 template <class T>
 class add {
-public:
+ public:
   INLINE T operator()(const T a, const T b) const { return a + b; }
 };
 
 template <class T>
 class add_scale {
-private:
+ private:
   const T p1;
   const T p2;
 
-public:
+ public:
   INLINE add_scale(const T s1, const T s2) : p1(s1), p2(s2) {}
   INLINE T operator()(const T a, const T b) const { return p1 * a + p2 * b; }
 };
 
 template <class T>
 class sub {
-public:
+ public:
   INLINE T operator()(const T a, const T b) const { return a - b; }
 };
 
 template <class T>
 class mul {
-public:
+ public:
   INLINE T operator()(const T a, const T b) const { return a * b; }
 };
 
 template <class T>
 class div {
-public:
+ public:
   INLINE T operator()(const T a, const T b) const { return a / b; }
 };
 
 template <class T>
 class cmp_eq {
-public:
+ public:
   INLINE bool operator()(const T a, const T b) const { return a == b; }
 };
 
 template <class T>
 class cmp_ne {
-public:
+ public:
   INLINE bool operator()(const T a, const T b) const { return a != b; }
 };
 
 template <class T>
 class cmp_le {
-public:
+ public:
   INLINE bool operator()(const T a, const T b) const { return a <= b; }
 };
 
 template <class T>
 class cmp_lt {
-public:
+ public:
   INLINE bool operator()(const T a, const T b) const { return a < b; }
 };
 
 template <class T>
 class cmp_ge {
-public:
+ public:
   INLINE bool operator()(const T a, const T b) const { return a >= b; }
 };
 
 template <class T>
 class cmp_gt {
-public:
+ public:
   INLINE bool operator()(const T a, const T b) const { return a > b; }
 };
 
 template <class T>
 class and_op {
-public:
+ public:
   INLINE bool operator()(const T a, const T b) const { return a && b; }
 };
 
 template <class T>
 class or_op {
-public:
+ public:
   INLINE bool operator()(const T a, const T b) const { return a || b; }
 };
 
 template <class T>
 class min {
-public:
+ public:
   INLINE T operator()(const T a, const T b) const { return a > b ? b : a; }
 };
 
 template <class T>
 class max {
-public:
+ public:
   INLINE T operator()(const T a, const T b) const { return a < b ? b : a; }
 };
 
@@ -332,7 +332,7 @@ public:
 #ifndef PADDLE_TYPE_DOUBLE
 template <>
 class add<__m128> {
-public:
+ public:
   INLINE __m128 operator()(const __m128 a, const __m128 b) const {
     return _mm_add_ps(a, b);
   }
@@ -340,11 +340,11 @@ public:
 
 template <>
 class add_scale<__m128> {
-private:
+ private:
   const __m128 p1;
   const __m128 p2;
 
-public:
+ public:
   INLINE add_scale(const __m128 s1, const __m128 s2) : p1(s1), p2(s2) {}
   INLINE __m128 operator()(const __m128 a, const __m128 b) const {
     return _mm_add_ps(_mm_mul_ps(p1, a), _mm_mul_ps(p2, b));
@@ -353,7 +353,7 @@ public:
 
 template <>
 class sub<__m128> {
-public:
+ public:
   INLINE __m128 operator()(const __m128 a, const __m128 b) const {
     return _mm_sub_ps(a, b);
   }
@@ -361,7 +361,7 @@ public:
 
 template <>
 class mul<__m128> {
-public:
+ public:
   INLINE __m128 operator()(const __m128 a, const __m128 b) const {
     return _mm_mul_ps(a, b);
   }
@@ -369,7 +369,7 @@ public:
 
 template <>
 class div<__m128> {
-public:
+ public:
   INLINE __m128 operator()(const __m128 a, const __m128 b) const {
     return _mm_div_ps(a, b);
   }
@@ -377,7 +377,7 @@ public:
 
 template <>
 class min<__m128> {
-public:
+ public:
   INLINE __m128 operator()(const __m128 a, const __m128 b) const {
     return _mm_min_ps(a, b);
   }
@@ -385,7 +385,7 @@ public:
 
 template <>
 class max<__m128> {
-public:
+ public:
   INLINE __m128 operator()(const __m128 a, const __m128 b) const {
     return _mm_max_ps(a, b);
   }
@@ -393,7 +393,7 @@ public:
 #else
 template <>
 class add<__m128d> {
-public:
+ public:
   INLINE __m128d operator()(const __m128d a, const __m128d b) const {
     return _mm_add_pd(a, b);
   }
@@ -401,11 +401,11 @@ public:
 
 template <>
 class add_scale<__m128d> {
-private:
+ private:
   const __m128d p1;
   const __m128d p2;
 
-public:
+ public:
   INLINE add_scale(const __m128d s1, const __m128d s2) : p1(s1), p2(s2) {}
   INLINE __m128d operator()(const __m128d a, const __m128d b) const {
     return _mm_add_pd(_mm_mul_pd(p1, a), _mm_mul_pd(p2, b));
@@ -414,7 +414,7 @@ public:
 
 template <>
 class sub<__m128d> {
-public:
+ public:
   INLINE __m128d operator()(const __m128d a, const __m128d b) const {
     return _mm_sub_pd(a, b);
   }
@@ -422,7 +422,7 @@ public:
 
 template <>
 class mul<__m128d> {
-public:
+ public:
   INLINE __m128d operator()(const __m128d a, const __m128d b) const {
     return _mm_mul_pd(a, b);
   }
@@ -430,7 +430,7 @@ public:
 
 template <>
 class div<__m128d> {
-public:
+ public:
   INLINE __m128d operator()(const __m128d a, const __m128d b) const {
     return _mm_div_pd(a, b);
   }
@@ -438,7 +438,7 @@ public:
 
 template <>
 class min<__m128d> {
-public:
+ public:
   INLINE __m128d operator()(const __m128d a, const __m128d b) const {
     return _mm_min_pd(a, b);
   }
@@ -446,7 +446,7 @@ public:
 
 template <>
 class max<__m128d> {
-public:
+ public:
   INLINE __m128d operator()(const __m128d a, const __m128d b) const {
     return _mm_max_pd(a, b);
   }
@@ -458,7 +458,7 @@ public:
 #ifndef PADDLE_TYPE_DOUBLE
 template <>
 class add<float32x4_t> {
-public:
+ public:
   INLINE float32x4_t operator()(const float32x4_t a,
                                 const float32x4_t b) const {
     return vaddq_f32(a, b);
@@ -467,11 +467,11 @@ public:
 
 template <>
 class add_scale<float32x4_t> {
-private:
+ private:
   const float32x4_t p1;
   const float32x4_t p2;
 
-public:
+ public:
   INLINE add_scale(const float32x4_t s1, const float32x4_t s2)
       : p1(s1), p2(s2) {}
   INLINE float32x4_t operator()(const float32x4_t a,
@@ -482,7 +482,7 @@ public:
 
 template <>
 class sub<float32x4_t> {
-public:
+ public:
   INLINE float32x4_t operator()(const float32x4_t a,
                                 const float32x4_t b) const {
     return vsubq_f32(a, b);
@@ -491,7 +491,7 @@ public:
 
 template <>
 class mul<float32x4_t> {
-public:
+ public:
   INLINE float32x4_t operator()(const float32x4_t a,
                                 const float32x4_t b) const {
     return vmulq_f32(a, b);
@@ -500,7 +500,7 @@ public:
 
 template <>
 class div<float32x4_t> {
-public:
+ public:
   INLINE float32x4_t operator()(const float32x4_t a,
                                 const float32x4_t b) const {
     float32x4_t tmp = vrecpeq_f32(b);
@@ -510,7 +510,7 @@ public:
 
 template <>
 class min<float32x4_t> {
-public:
+ public:
   INLINE float32x4_t operator()(const float32x4_t a,
                                 const float32x4_t b) const {
     return vminq_f32(a, b);
@@ -519,7 +519,7 @@ public:
 
 template <>
 class max<float32x4_t> {
-public:
+ public:
   INLINE float32x4_t operator()(const float32x4_t a,
                                 const float32x4_t b) const {
     return vmaxq_f32(a, b);

paddle/cuda/src/hl_cuda_lstm.cu

@@ -30,7 +30,7 @@ bool hl_lstm_sequence_parallel(int frameSize) {
 }
 
 class frameValue {
-public:
+ public:
   real *value_;
   __device__ frameValue(real *value) : value_(value) {}
   template <int reversed, int frameSize>

paddle/fluid/framework/data_device_transform.cc

@@ -16,31 +16,25 @@ limitations under the License. */
 namespace paddle {
 namespace framework {
 
-static const platform::DeviceContext* GetDeviceContext(
-    const platform::Place& src_place, const platform::Place& dst_place) {
-  platform::DeviceContextPool& pool = platform::DeviceContextPool::Instance();
-
-  if (platform::is_gpu_place(src_place) && platform::is_cpu_place(dst_place)) {
-    return pool.Get(src_place);
-  } else if (platform::is_cpu_place(src_place) &&
-             platform::is_gpu_place(dst_place)) {
-    return pool.Get(dst_place);
-  } else {
-    PADDLE_THROW(
-        "Currently, model parallelism is only supported between CPU and CUDA");
-  }
-}
-
-void TransDataDevice(const Tensor& in, const platform::Place& dst_place,
-                     Tensor* out) {
+void TransDataDevice(const Tensor &in, const platform::Place &dst_place,
+                     Tensor *out) {
   VLOG(3) << "DeviceTransform in, src_place " << in.place()
           << " dst_place: " << dst_place;
-  auto* dev_ctx = GetDeviceContext(in.place(), dst_place);
 
-  TensorCopy(in, dst_place, *dev_ctx, out);
-  if (platform::is_gpu_place(in.place()) && platform::is_cpu_place(dst_place)) {
-    dev_ctx->Wait();
-  }
+  PADDLE_ENFORCE_NE(
+      in.place().which(), dst_place.which(),
+      "Currently, model parallelism is only supported between CPU and CUDA");
+
+  // FIXME(zcd): TransDataDevice is used to transform data from GPU to CPU and
+  // the enforced checkings have been done in GetDeviceContext, so the
+  // `dev_ctx->Wait()` is necessary. But `dev_ctx->Wait()` will make the program
+  // slow, especially when the number of elements is little, for example,
+  // the elements of learning rate are one and it's CPU side.
+  // One solution is to use a CUDA kernel to complete the copy operation when
+  // the transforming is from CPU to GPU and the number of elements is little.
+  // But the embarrassment is that this solution this solution makes training
+  // slower.
+  TensorCopySync(in, dst_place, out);
 }
 
 }  // namespace framework

paddle/fluid/framework/details/CMakeLists.txt

@@ -3,7 +3,7 @@ cc_library(op_handle_base SRCS op_handle_base.cc DEPS var_handle device_context
 cc_library(scale_loss_grad_op_handle SRCS scale_loss_grad_op_handle.cc DEPS op_handle_base scope lod_tensor ddim memory)
 cc_library(fetch_op_handle SRCS fetch_op_handle.cc DEPS op_handle_base scope lod_tensor ddim memory)
 cc_library(computation_op_handle SRCS computation_op_handle.cc DEPS framework_proto scope place operator op_registry)
-cc_library(send_op_handle SRCS send_op_handle.cc DEPS framework_proto scope place operator op_registry)
+cc_library(rpc_op_handle SRCS rpc_op_handle.cc DEPS framework_proto scope place operator op_registry)
 
 cc_library(ssa_graph SRCS ssa_graph.cc DEPS var_handle op_handle_base)
 cc_library(ssa_graph_builder SRCS ssa_graph_builder.cc DEPS ssa_graph)
@@ -26,7 +26,7 @@ endif()
 cc_library(gather_op_handle SRCS gather_op_handle.cc DEPS op_handle_base scope ddim memory variable_visitor)
 
 cc_library(multi_devices_graph_builder SRCS multi_devices_graph_builder.cc DEPS ssa_graph_builder computation_op_handle
-        scale_loss_grad_op_handle send_op_handle ${multi_devices_graph_builder_deps} reduce_op_handle broadcast_op_handle)
+        scale_loss_grad_op_handle rpc_op_handle ${multi_devices_graph_builder_deps} reduce_op_handle broadcast_op_handle)
 
 cc_library(ssa_graph_executor SRCS ssa_graph_executor.cc DEPS ssa_graph framework_proto)
 cc_library(threaded_ssa_graph_executor SRCS threaded_ssa_graph_executor.cc DEPS fetch_op_handle ssa_graph_executor scope

paddle/fluid/framework/details/multi_devices_graph_builder.cc

@@ -12,12 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 #include "paddle/fluid/framework/details/multi_devices_graph_builder.h"
+#include <fstream>
 #include <utility>
 #include "paddle/fluid/framework/details/broadcast_op_handle.h"
 #include "paddle/fluid/framework/details/computation_op_handle.h"
 #include "paddle/fluid/framework/details/reduce_op_handle.h"
+#include "paddle/fluid/framework/details/rpc_op_handle.h"
 #include "paddle/fluid/framework/details/scale_loss_grad_op_handle.h"
-#include "paddle/fluid/framework/details/send_op_handle.h"
 #include "paddle/fluid/framework/op_info.h"
 #include "paddle/fluid/framework/scope.h"
 
@@ -28,6 +29,10 @@
 #include <string>
 #include <vector>
 
+DEFINE_string(ssa_graph_path, "/tmp/ssa_graph.dot",
+              "the ssa graph path only print with GLOG_v=10,"
+              "default /tmp/graph.dot");
+
 namespace paddle {
 namespace framework {
 namespace details {
@@ -79,9 +84,44 @@ void MultiDevSSAGraphBuilder::CreateOpHandleIOs(SSAGraph *result,
   }
 }
 
-bool MultiDevSSAGraphBuilder::IsDistTrainOp(const OpDesc &op,
-                                            OpDesc *send_op) const {
-  if (send_op == nullptr) {
+std::vector<std::string> MultiDevSSAGraphBuilder::FindDistTrainSendVars(
+    const ProgramDesc &program) const {
+  std::vector<std::string> send_vars;
+  // since parameters are all in block 0,
+  // it's enough to only scan send ops in block 0
+  for (auto *op : program.Block(0).AllOps()) {
+    // TODO(Yancey1989): use a graceful method to find send op,
+    // instead of the the hard code string
+    if (op->Type() == "send_vars") {
+      auto op_vars = op->InputArgumentNames();
+      send_vars.reserve(send_vars.size() +
+                        std::distance(op_vars.begin(), op_vars.end()));
+      send_vars.insert(send_vars.end(), op_vars.begin(), op_vars.end());
+    }
+  }
+  return send_vars;
+}
+
+std::vector<std::string> MultiDevSSAGraphBuilder::FindDistTrainRecvVars(
+    const ProgramDesc &program) const {
+  std::vector<std::string> recv_vars;
+  for (auto *op : program.Block(0).AllOps()) {
+    // TODO(Yancey1989): use a graceful method to find recv op,
+    // instead of the hard code string
+    if (op->Type() == "recv") {
+      auto op_vars = op->OutputArgumentNames();
+      recv_vars.reserve(recv_vars.size() +
+                        std::distance(op_vars.begin(), op_vars.end()));
+      recv_vars.insert(recv_vars.end(), op_vars.begin(), op_vars.end());
+    }
+  }
+  return recv_vars;
+}
+
+bool MultiDevSSAGraphBuilder::IsDistTrainOp(
+    const OpDesc &op, const std::vector<std::string> &send_vars,
+    const std::vector<std::string> &recv_vars) const {
+  if (send_vars.size() == 0 || recv_vars.size() == 0) {
     return false;
   }
 
@@ -89,22 +129,21 @@ bool MultiDevSSAGraphBuilder::IsDistTrainOp(const OpDesc &op,
    * Check any of opvars contains `.block` and in sendvars
    */
   auto checker = [](const std::vector<std::string> &opvars,
-                    const std::vector<std::string> &sendvars) -> bool {
+                    const std::vector<std::string> &rpc_vars) -> bool {
     for (auto &var : opvars) {
+      // a variable name with the suffix `.block` means it's a splited
+      // variable by (DistributeTranspiler)
+      // [python/paddle/fluid/transpiler/distribute_transpiler.py]
       if (var.find(".block") != std::string::npos &&
-          std::find(sendvars.begin(), sendvars.end(), var) != sendvars.end()) {
+          std::find(rpc_vars.begin(), rpc_vars.end(), var) != rpc_vars.end()) {
         return true;
       }
     }
     return false;
   };
 
-  if (op.Type() == "split" || op.Type() == "split_byref") {
-    return checker(op.OutputArgumentNames(), send_op->InputArgumentNames());
-  } else if (op.Type() == "concat") {
-    return checker(op.InputArgumentNames(), send_op->OutputArgumentNames());
-  }
-  return false;
+  return checker(op.OutputArgumentNames(), send_vars) ||
+         checker(op.InputArgumentNames(), recv_vars);
 }
 
 std::unique_ptr<SSAGraph> MultiDevSSAGraphBuilder::Build(
@@ -123,8 +162,10 @@ std::unique_ptr<SSAGraph> MultiDevSSAGraphBuilder::Build(
       std::unordered_map<std::string, std::vector<std::unique_ptr<VarHandle>>>>(
       places_.size());
 
-  // Find "send" op first for split is in front of send.
-  OpDesc *send_op = GetSendOpDesc(program);
+  // find send/recv vars so that we can place the distributed training
+  // realted op in the place 0
+  auto send_vars = FindDistTrainSendVars(program);
+  auto recv_vars = FindDistTrainRecvVars(program);
 
   size_t cur_device_id = 0;
   std::vector<std::unordered_set<std::string>> var_name_on_devices;
@@ -134,12 +175,14 @@ std::unique_ptr<SSAGraph> MultiDevSSAGraphBuilder::Build(
 
   bool is_forwarding = true;
   for (auto *op : program.Block(0).AllOps()) {
-    if (op->Type() == "send") {
-      // append send op if program is distributed trainer main program.
+    if (boost::get<int>(
+            op->GetAttr(OpProtoAndCheckerMaker::OpRoleAttrName())) ==
+        static_cast<int>(OpRole::kRPC)) {
+      // append rpc op if program is distributed trainer main program.
       // always use the first device
-      CreateSendOp(&result, *op);
-    } else if (IsDistTrainOp(*op, send_op)) {
-      CreateComputationalOps(&result, *op, 1);
+      CreateRPCOp(&result, *op);
+    } else if (IsDistTrainOp(*op, send_vars, recv_vars)) {
+      CreateDistTrainOp(&result, *op);
     } else if (IsScaleLossOp(*op)) {
       // user can customize loss@grad if not use_default_grad_scale_
       if (strategy_.gradient_scale_ !=
@@ -218,9 +261,8 @@ std::unique_ptr<SSAGraph> MultiDevSSAGraphBuilder::Build(
   AddOutputToLeafOps(&result);
 
   if (VLOG_IS_ON(10)) {
-    std::ostringstream sout;
-    PrintGraphviz(*graph, sout);
-    VLOG(10) << sout.str();
+    std::ofstream fout(FLAGS_ssa_graph_path);
+    PrintGraphviz(*graph, fout);
   }
 
   return std::unique_ptr<SSAGraph>(graph);
@@ -270,15 +312,6 @@ void MultiDevSSAGraphBuilder::CreateComputationalOp(SSAGraph *result,
   CreateOpHandleIOs(result, op, dev_id);
 }
 
-OpDesc *MultiDevSSAGraphBuilder::GetSendOpDesc(
-    const ProgramDesc &program) const {
-  for (auto *op : program.Block(0).AllOps()) {
-    if (op->Type() == "send") {
-      return op;
-    }
-  }
-  return nullptr;
-}
 void MultiDevSSAGraphBuilder::InsertNCCLAllReduceOp(
     SSAGraph *result, const std::string &og) const {
 #ifdef PADDLE_WITH_CUDA
@@ -401,14 +434,48 @@ VarHandle *MultiDevSSAGraphBuilder::CreateReduceOp(SSAGraph *result,
   return var;
 }
 
-void MultiDevSSAGraphBuilder::CreateSendOp(SSAGraph *result,
-                                           const OpDesc &op) const {
+void MultiDevSSAGraphBuilder::ConnectOp(SSAGraph *result, OpHandleBase *op,
+                                        const std::string &prev_op_name) const {
+  for (auto &prev_op : result->ops_) {
+    if (prev_op->Name() == prev_op_name) {
+      auto *dep_var = new DummyVarHandle();
+      prev_op->AddOutput(dep_var);
+      result->dep_vars_.emplace(dep_var);
+      op->AddInput(dep_var);
+    }
+  }
+}
+
+void MultiDevSSAGraphBuilder::CreateDistTrainOp(SSAGraph *result,
+                                                const OpDesc &op) const {
+  CreateComputationalOp(result, op, 0);
+  if (op.Type() == "concat") {
+    ConnectOp(result, result->ops_.back().get(), "fetch_barrier");
+  }
+}
+
+void MultiDevSSAGraphBuilder::CreateRPCOp(SSAGraph *result,
+                                          const OpDesc &op) const {
   auto &p = places_[0];
   auto *s = local_scopes_[0];
-  // FIXME(wuyi): send op always copy from GPU 0
-  result->ops_.emplace_back(new SendOpHandle(op, s, p));
-  // Create inputs for output on original place and no ssa output
-  // is created for send op.
+  result->ops_.emplace_back(new RPCOpHandle(op, s, p, op.Type()));
+
+  if (op.Type() == "send_barrier") {
+    ConnectOp(result, result->ops_.back().get(), "send_vars");
+  } else if (op.Type() == "recv") {
+    ConnectOp(result, result->ops_.back().get(), "send_barrier");
+  } else if (op.Type() == "fetch_barrier") {
+    ConnectOp(result, result->ops_.back().get(), "recv");
+  } else if (op.Type() == "send_vars") {
+    // do nothing
+  } else {
+    PADDLE_THROW(
+        "rpc op should be in ["
+        "send_vars, send_barrier. recv, fetch_barrier]");
+  }
+
+  // TODO(Yancey1989): schedule rpc op on different place may
+  // increate throughput
   CreateOpHandleIOs(result, op, 0);
 }
 

paddle/fluid/framework/details/multi_devices_graph_builder.h

@@ -64,12 +64,24 @@ class MultiDevSSAGraphBuilder : public SSAGraphBuilder {
 
   bool IsScaleLossOp(const OpDesc &op) const;
 
-  void CreateSendOp(SSAGraph *result, const OpDesc &op) const;
+  void CreateRPCOp(SSAGraph *result, const OpDesc &op) const;
+  void CreateDistTrainOp(SSAGraph *result, const OpDesc &op) const;
 
   /**
    * Is this operator as the end-point operator before/after send operator.
    */
-  bool IsDistTrainOp(const OpDesc &op, OpDesc *send_op) const;
+  bool IsDistTrainOp(const OpDesc &op,
+                     const std::vector<std::string> &send_vars,
+                     const std::vector<std::string> &recv_vars) const;
+
+  std::vector<std::string> FindDistTrainSendVars(
+      const ProgramDesc &program) const;
+
+  std::vector<std::string> FindDistTrainRecvVars(
+      const ProgramDesc &program) const;
+
+  void ConnectOp(SSAGraph *result, OpHandleBase *op,
+                 const std::string &prev_op_name) const;
 
   void CreateComputationalOps(SSAGraph *result, const OpDesc &op,
                               size_t num_places) const;
@@ -93,12 +105,6 @@ class MultiDevSSAGraphBuilder : public SSAGraphBuilder {
   void CreateBroadcastOp(SSAGraph *result, const std::string &p_name,
                          size_t src_dev_id) const;
 
-  /**
-   * Get send op in the global block of program.
-   * nullptr if not found.
-   */
-  OpDesc *GetSendOpDesc(const ProgramDesc &program) const;
-
   bool IsSparseGradient(
       const std::unordered_map<std::string, proto::VarType::Type> &var_types,
       const std::string &og) const;

paddle/fluid/framework/details/send_op_handle.cc → paddle/fluid/framework/details/rpc_op_handle.cc

@@ -12,24 +12,26 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "paddle/fluid/framework/details/send_op_handle.h"
+#include "paddle/fluid/framework/details/rpc_op_handle.h"
 
 namespace paddle {
 namespace framework {
 namespace details {
 
-SendOpHandle::SendOpHandle(const framework::OpDesc &op_desc,
-                           const Scope *local_scope,
-                           const platform::Place &place)
+RPCOpHandle::RPCOpHandle(const framework::OpDesc &op_desc,
+                         const Scope *local_scope, const platform::Place &place,
+                         const std::string &name)
     : op_(framework::OpRegistry::CreateOp(op_desc)),
       local_scope_(local_scope),
-      place_(place) {}
+      place_(place),
+      name_(name) {}
 
-void SendOpHandle::RunImpl() {
+void RPCOpHandle::RunImpl() {
   // TODO(wuyi): need further analysis whether wait VarDummyHandle.
   // Wait input done
   for (auto *in : inputs_) {
     auto &p = static_cast<VarHandle *>(in)->place_;
+    // FIXME(Yancey1989): need a better solution instead of use DebugString()
     if (in->DebugString() == "dummy") {  // HACK
       continue;
     }
@@ -43,7 +45,7 @@ void SendOpHandle::RunImpl() {
   op_->Run(*tmp_scope, place_);
 }
 
-std::string SendOpHandle::Name() const { return "send"; }
+std::string RPCOpHandle::Name() const { return name_; }
 }  // namespace details
 }  // namespace framework
 }  // namespace paddle

paddle/fluid/framework/details/send_op_handle.h → paddle/fluid/framework/details/rpc_op_handle.h

@@ -27,9 +27,9 @@ namespace paddle {
 namespace framework {
 namespace details {
 
-struct SendOpHandle : public OpHandleBase {
-  SendOpHandle(const framework::OpDesc& op_desc, const Scope* local_scope,
-               const platform::Place& place);
+struct RPCOpHandle : public OpHandleBase {
+  RPCOpHandle(const framework::OpDesc& op_desc, const Scope* local_scope,
+              const platform::Place& place, const std::string& name);
 
   std::string Name() const override;
 
@@ -44,6 +44,7 @@ struct SendOpHandle : public OpHandleBase {
   std::unique_ptr<OperatorBase> op_;
   const Scope* local_scope_;
   const platform::Place& place_;
+  const std::string name_;
 };
 
 }  // namespace details

paddle/fluid/framework/executor.cc

@@ -24,9 +24,6 @@ limitations under the License. */
 #include "paddle/fluid/platform/profiler.h"
 
 DECLARE_bool(benchmark);
-DEFINE_bool(check_nan_inf, false,
-            "Checking whether operator produce NAN/INF or not. It will be "
-            "extremely slow so please use this flag wisely.");
 
 namespace paddle {
 namespace framework {
@@ -78,21 +75,6 @@ void InitializeVariable(Variable* var, proto::VarType::Type var_type) {
   }
 }
 
-static void CheckTensorNANOrInf(const std::string& name,
-                                const framework::Tensor& tensor) {
-  if (tensor.memory_size() == 0) {
-    return;
-  }
-  if (tensor.type().hash_code() != typeid(float).hash_code() &&   // NOLINT
-      tensor.type().hash_code() != typeid(double).hash_code()) {  // NOLINT
-    return;
-  }
-  PADDLE_ENFORCE(!framework::TensorContainsInf(tensor),
-                 "Tensor %s contains Inf", name);
-  PADDLE_ENFORCE(!framework::TensorContainsNAN(tensor),
-                 "Tensor %s contains NAN", name);
-}
-
 void Executor::CreateVariables(const ProgramDesc& pdesc, Scope* scope,
                                int block_id) {
   auto& global_block = pdesc.Block(block_id);
@@ -340,15 +322,6 @@ void Executor::RunPreparedContext(ExecutorPrepareContext* ctx, Scope* scope,
       VLOG(2) << "Memory used after operator " + op->Type() + " running: "
               << memory::memory_usage(place_);
     }
-    if (FLAGS_check_nan_inf) {
-      for (auto& vname : op->OutputVars(true)) {
-        auto* var = local_scope->FindVar(vname);
-        if (var == nullptr) continue;
-        if (var->IsType<framework::LoDTensor>()) {
-          CheckTensorNANOrInf(vname, var->Get<framework::LoDTensor>());
-        }
-      }
-    }
   }
   platform::DeviceContextPool::Instance().Get(place_)->Wait();
   if (create_vars && create_local_scope) {

paddle/fluid/framework/op_proto_maker.cc

@@ -66,7 +66,7 @@ void OpProtoAndCheckerMaker::operator()(proto::OpProto* proto,
       .InEnum(
           {static_cast<int>(OpRole::kForward),
            static_cast<int>(OpRole::kBackward),
-           static_cast<int>(OpRole::kOptimize),
+           static_cast<int>(OpRole::kOptimize), static_cast<int>(OpRole::kRPC),
            static_cast<int>(OpRole::kLoss) | static_cast<int>(OpRole::kForward),
            static_cast<int>(OpRole::kLoss) |
                static_cast<int>(OpRole::kBackward),

paddle/fluid/framework/op_proto_maker.h

@@ -24,6 +24,7 @@ enum class OpRole {
   kForward = 0x0000,
   kBackward = 0x0001,
   kOptimize = 0x0002,
+  kRPC = 0x0003,
 
   kLoss = 0x0100,
   // The default value of op's role. This should be only used for unittests and

paddle/fluid/framework/operator.cc

@@ -24,6 +24,9 @@ limitations under the License. */
 #include "paddle/fluid/platform/profiler.h"
 
 DECLARE_bool(benchmark);
+DEFINE_bool(check_nan_inf, false,
+            "Checking whether operator produce NAN/INF or not. It will be "
+            "extremely slow so please use this flag wisely.");
 
 namespace paddle {
 namespace framework {
@@ -513,6 +516,21 @@ class RuntimeInferShapeContext : public InferShapeContext {
   const Scope& scope_;
 };
 
+static void CheckTensorNANOrInf(const std::string& name,
+                                const framework::Tensor& tensor) {
+  if (tensor.memory_size() == 0) {
+    return;
+  }
+  if (tensor.type().hash_code() != typeid(float).hash_code() &&   // NOLINT
+      tensor.type().hash_code() != typeid(double).hash_code()) {  // NOLINT
+    return;
+  }
+  PADDLE_ENFORCE(!framework::TensorContainsInf(tensor),
+                 "Tensor %s contains Inf", name);
+  PADDLE_ENFORCE(!framework::TensorContainsNAN(tensor),
+                 "Tensor %s contains NAN", name);
+}
+
 void OperatorWithKernel::RunImpl(const Scope& scope,
                                  const platform::Place& place) const {
   RuntimeInferShapeContext infer_shape_ctx(*this, scope);
@@ -597,6 +615,16 @@ void OperatorWithKernel::RunImpl(const Scope& scope,
   if (FLAGS_benchmark) {
     new_dev_ctx->Wait();
   }
+
+  if (FLAGS_check_nan_inf) {
+    for (auto& vname : OutputVars(true)) {
+      auto* var = new_scope.FindVar(vname);
+      if (var == nullptr) continue;
+      if (var->IsType<framework::LoDTensor>()) {
+        CheckTensorNANOrInf(vname, var->Get<framework::LoDTensor>());
+      }
+    }
+  }
 }
 
 proto::VarType::Type OperatorWithKernel::IndicateDataType(

paddle/fluid/framework/selected_rows.cc

@@ -121,24 +121,29 @@ bool SelectedRows::HasKey(int64_t key) const {
 }
 
 std::vector<std::pair<int64_t, int64_t>> SelectedRows::Get(
-    std::vector<int64_t> keys, framework::Tensor* value) const {
+    const std::vector<int64_t>& keys, framework::Tensor* value) const {
   PADDLE_ENFORCE(value->IsInitialized(),
                  "The value tensor should be initialized.");
   std::vector<std::pair<int64_t, int64_t>> non_keys_pair;
-  int64_t value_width = value_->numel() / value_->dims()[0];
-  PADDLE_ENFORCE_EQ(value_width, value->numel() / value->dims()[0],
-                    "output tensor should have the same shape with table "
-                    "execpt the dims[0].");
-
-  for (size_t i = 0; i < keys.size(); ++i) {
-    int64_t index = Index(keys[i]);
-    if (index == -1) {
-      non_keys_pair.push_back(std::make_pair(keys[i], static_cast<int64_t>(i)));
-    } else {
-      framework::VisitDataType(
-          framework::ToDataType(value_->type()),
-          TensorCopyVisitor(value, i * value_width, *value_.get(),
-                            index * value_width, value_width));
+  if (keys.empty()) {
+    VLOG(3) << "keys is empty, please check data!";
+  } else {
+    int64_t value_width = value_->numel() / value_->dims()[0];
+    PADDLE_ENFORCE_EQ(value_width, value->numel() / value->dims()[0],
+                      "output tensor should have the same shape with table "
+                      "except the dims[0].");
+
+    for (size_t i = 0; i < keys.size(); ++i) {
+      int64_t index = Index(keys[i]);
+      if (index == -1) {
+        non_keys_pair.push_back(
+            std::make_pair(keys[i], static_cast<int64_t>(i)));
+      } else {
+        framework::VisitDataType(
+            framework::ToDataType(value_->type()),
+            TensorCopyVisitor(value, i * value_width, *value_.get(),
+                              index * value_width, value_width));
+      }
     }
   }
   return non_keys_pair;

paddle/fluid/framework/selected_rows.h

@@ -82,7 +82,7 @@ class SelectedRows {
    * @return a list of pair which contains the non-exists key and the index in
    * the value
    */
-  std::vector<std::pair<int64_t, int64_t>> Get(std::vector<int64_t> keys,
+  std::vector<std::pair<int64_t, int64_t>> Get(const std::vector<int64_t>& keys,
                                                framework::Tensor* value) const;
 
   /*

paddle/fluid/inference/analysis/data_flow_graph_tester.cc

@@ -35,7 +35,7 @@ TEST(DataFlowGraph, BFS) {
 
   GraphTraits<DataFlowGraph> trait(&dfg);
   auto nodes = trait.nodes();
-  int count = 0;
+  size_t count = 0;
   for (auto it = nodes.begin(); it != nodes.end(); ++it) {
     LOG(INFO) << "visiting " << it->name();
     ++count;
@@ -49,7 +49,7 @@ TEST(DataFlowGraph, DFS) {
   dfg.Build();
   GraphTraits<DataFlowGraph> trait(&dfg);
   auto nodes = trait.nodes_in_DFS();
-  int count = 0;
+  size_t count = 0;
   for (auto it = nodes.begin(); it != nodes.end(); ++it) {
     LOG(INFO) << "visiting " << it->name();
     ++count;

paddle/fluid/inference/analysis/helper.h

@@ -24,6 +24,15 @@ namespace paddle {
 namespace inference {
 namespace analysis {
 
+template <typename Vec>
+int AccuDims(Vec &&vec, int size) {
+  int res = 1;
+  for (int i = 0; i < size; i++) {
+    res *= std::forward<Vec>(vec)[i];
+  }
+  return res;
+}
+
 #define SET_TYPE(type__) dic_[typeid(type__).hash_code()] = #type__;
 /*
  * Map typeid to representation.
@@ -101,7 +110,5 @@ class OrderedRegistry {
 }  // namespace paddle
 
 #define PADDLE_DISALLOW_COPY_AND_ASSIGN(type__) \
-                                                \
   type__(const type__ &) = delete;              \
-                                                \
   void operator=(const type__ &) = delete;

paddle/fluid/inference/tensorrt/convert/CMakeLists.txt

-nv_test(test_op_converter SRCS test_op_converter.cc mul_op.cc conv2d_op.cc DEPS ${FLUID_CORE_MODULES})
+# Add TRT tests
+nv_test(test_op_converter SRCS test_op_converter.cc mul_op.cc conv2d_op.cc DEPS ${FLUID_CORE_MODULES} tensorrt_engine)
 # This test is not stable
 # See https://paddleci.ngrok.io/viewLog.html?tab=buildLog&buildTypeId=Paddle_PrCi2&buildId=36834&_focus=8828 
 #nv_test(test_trt_activation_op SRCS test_activation_op.cc activation_op.cc io_converter.cc
 #    DEPS ${FLUID_CORE_MODULES} activation_op tensorrt_engine
 #    SERIAL)
 nv_test(test_io_converter SRCS test_io_converter.cc io_converter.cc DEPS dynload_cuda dynamic_loader lod_tensor)
+nv_test(test_trt_mul_op SRCS test_mul_op.cc mul_op.cc
+        DEPS ${FLUID_CORE_MODULES} tensorrt_engine mul_op SERIAL)

paddle/fluid/inference/tensorrt/convert/mul_op.cc

@@ -18,11 +18,25 @@ namespace paddle {
 namespace inference {
 namespace tensorrt {
 
+/*
+ * MulOp, IMatrixMultiplyLayer in TRT. This Layer doesn't has weights.
+ */
 class MulOpConverter : public OpConverter {
  public:
   MulOpConverter() {}
   void operator()(const framework::proto::OpDesc& op) override {
-    LOG(INFO) << "convert a fluid mul op to tensorrt fc layer without bias";
+    VLOG(4) << "convert a fluid mul op to tensorrt fc layer without bias";
+
+    framework::OpDesc op_desc(op, nullptr, nullptr);
+    // Declare inputs
+    auto* input1 = engine_->GetITensor(op_desc.Input("X")[0]);
+    auto* input2 = engine_->GetITensor(op_desc.Input("Y")[0]);
+    // Both the input1 and input2 do not need transpose.
+    auto* layer = TRT_ENGINE_ADD_LAYER(
+        engine_, MatrixMultiply, *const_cast<nvinfer1::ITensor*>(input1), false,
+        *const_cast<nvinfer1::ITensor*>(input2), false);
+
+    engine_->DeclareOutput(layer, 0, op_desc.Output("Out")[0]);
   }
 };
 

paddle/fluid/inference/tensorrt/convert/test_activation_op.cc

@@ -102,3 +102,5 @@ TEST(OpConverter, ConvertRelu) {
 }  // namespace tensorrt
 }  // namespace inference
 }  // namespace paddle
+
+USE_OP(activation);

paddle/fluid/inference/tensorrt/convert/test_mul_op.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 <gtest/gtest.h>
+#include "paddle/fluid/framework/op_registry.h"
+#include "paddle/fluid/inference/tensorrt/convert/ut_helper.h"
+
+namespace paddle {
+namespace inference {
+namespace tensorrt {
+
+TEST(MulOpConverter, main) {
+  TRTConvertValidation validator(10, 1000);
+  validator.DeclInputVar("mul-X", nvinfer1::Dims2(10, 6));
+  validator.DeclInputVar("mul-Y", nvinfer1::Dims2(6, 10));
+  validator.DeclOutputVar("mul-Out", nvinfer1::Dims2(10, 10));
+
+  // Prepare Op description
+  framework::OpDesc desc;
+  desc.SetType("mul");
+  desc.SetInput("X", {"mul-X"});
+  desc.SetInput("Y", {"mul-Y"});
+  desc.SetOutput("Out", {"mul-Out"});
+
+  LOG(INFO) << "set OP";
+  validator.SetOp(*desc.Proto());
+  LOG(INFO) << "execute";
+
+  validator.Execute(10);
+}
+
+}  // namespace tensorrt
+}  // namespace inference
+}  // namespace paddle
+
+USE_OP(mul);

paddle/fluid/inference/tensorrt/convert/test_op_converter.cc

@@ -23,8 +23,6 @@ namespace tensorrt {
 TEST(OpConverter, ConvertBlock) {
   framework::ProgramDesc prog;
   auto* block = prog.MutableBlock(0);
-  auto* mul_op = block->AppendOp();
-  mul_op->SetType("mul");
   auto* conv2d_op = block->AppendOp();
   conv2d_op->SetType("conv2d");
 

paddle/fluid/inference/tensorrt/convert/ut_helper.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. */
+
+/*
+ * This file implements a UT framework to make the validation of transforming
+ * Fluid Op to TRT Layer.
+ */
+
+#pragma once
+
+#include "paddle/fluid/framework/lod_tensor.h"
+#include "paddle/fluid/framework/op_registry.h"
+#include "paddle/fluid/inference/analysis/helper.h"
+#include "paddle/fluid/inference/tensorrt/convert/op_converter.h"
+#include "paddle/fluid/inference/tensorrt/engine.h"
+
+namespace paddle {
+namespace inference {
+namespace tensorrt {
+
+/*
+ * Get a random float value between [low, high]
+ */
+float random(float low, float high) {
+  static std::random_device rd;
+  static std::mt19937 mt(rd());
+  std::uniform_real_distribution<double> dist(1.0, 10.0);
+  return dist(mt);
+}
+
+void RandomizeTensor(framework::LoDTensor* tensor, const platform::Place& place,
+                     const platform::DeviceContext& ctx) {
+  auto dims = tensor->dims();
+  size_t num_elements = analysis::AccuDims(dims, dims.size());
+  PADDLE_ENFORCE_GT(num_elements, 0);
+  auto* data = tensor->mutable_data<float>(place);
+  for (size_t i = 0; i < num_elements; i++) {
+    *(data + i) = random(0., 1.);
+  }
+}
+
+/*
+ * Help to validate the correctness between Fluid Op and the corresponding TRT
+ * layer.
+ */
+class TRTConvertValidation {
+ public:
+  TRTConvertValidation() = delete;
+
+  TRTConvertValidation(int batch_size, int workspace_size = 1 << 10) {
+    // create engine.
+    engine_.reset(new TensorRTEngine(10, 1 << 10, &stream_));
+    engine_->InitNetwork();
+
+    PADDLE_ENFORCE_EQ(cudaStreamCreate(&stream_), 0);
+  }
+
+  // Declare a Variable as input with random initialization.
+  void DeclInputVar(const std::string& name, const nvinfer1::Dims& dims) {
+    DeclVar(name, dims);
+    // Declare TRT inputs.
+    engine_->DeclareInput(name, nvinfer1::DataType::kFLOAT, dims);
+  }
+
+  void DeclOutputVar(const std::string& name, const nvinfer1::Dims& dims) {
+    DeclVar(name, dims);
+  }
+
+  void DeclVar(const std::string& name, const nvinfer1::Dims& dims) {
+    platform::CPUPlace place;
+    platform::CPUDeviceContext ctx(place);
+
+    // Init Fluid tensor.
+    std::vector<int> dim_vec(dims.nbDims);
+    for (int i = 0; i < dims.nbDims; i++) {
+      dim_vec[i] = dims.d[i];
+    }
+    auto* x = scope_.Var(name);
+    auto* x_tensor = x->GetMutable<framework::LoDTensor>();
+    x_tensor->Resize(framework::make_ddim(dim_vec));
+    RandomizeTensor(x_tensor, place, ctx);
+  }
+
+  void SetOp(const framework::proto::OpDesc& desc) {
+    op_ = framework::OpRegistry::CreateOp(desc);
+
+    OpConverter op_converter;
+    op_converter.ConvertOp(desc, engine_.get());
+
+    engine_->FreezeNetwork();
+
+    // Declare outputs.
+    op_desc_.reset(new framework::OpDesc(desc, nullptr, nullptr));
+
+    // Set Inputs.
+    for (const auto& input : op_desc_->InputArgumentNames()) {
+      auto* var = scope_.FindVar(input);
+      PADDLE_ENFORCE(var);
+      auto tensor = var->GetMutable<framework::LoDTensor>();
+      engine_->SetInputFromCPU(
+          input, static_cast<void*>(tensor->data<float>()),
+          sizeof(float) *
+              analysis::AccuDims(tensor->dims(), tensor->dims().size()));
+    }
+  }
+
+  void Execute(int batch_size) {
+    // Execute Fluid Op
+    // Execute TRT
+    platform::CPUPlace place;
+    platform::CPUDeviceContext ctx(place);
+    engine_->Execute(batch_size);
+
+    op_->Run(scope_, place);
+
+    ASSERT_FALSE(op_desc_->OutputArgumentNames().empty());
+    for (const auto& output : op_desc_->OutputArgumentNames()) {
+      std::vector<float> fluid_out;
+      std::vector<float> trt_out(200);
+      engine_->GetOutputInCPU(output, &trt_out[0], 200 * sizeof(float));
+
+      auto* var = scope_.FindVar(output);
+      auto tensor = var->GetMutable<framework::LoDTensor>();
+      framework::TensorToVector(*tensor, ctx, &fluid_out);
+      // Compare two output
+      ASSERT_FALSE(fluid_out.empty());
+      for (size_t i = 0; i < fluid_out.size(); i++) {
+        EXPECT_LT(std::abs(fluid_out[i] - trt_out[i]), 0.001);
+      }
+    }
+  }
+
+  framework::Scope& scope() { return scope_; }
+
+ private:
+  std::unique_ptr<TensorRTEngine> engine_;
+  cudaStream_t stream_;
+  framework::Scope scope_;
+  std::unique_ptr<framework::OperatorBase> op_;
+  std::unique_ptr<framework::OpDesc> op_desc_;
+};
+
+}  // namespace tensorrt
+}  // namespace inference
+}  // namespace paddle

paddle/fluid/inference/tensorrt/engine.cc

@@ -18,6 +18,7 @@ limitations under the License. */
 #include <cuda.h>
 #include <glog/logging.h>
 #include <string>
+#include "paddle/fluid/inference/analysis/helper.h"
 #include "paddle/fluid/inference/tensorrt/helper.h"
 #include "paddle/fluid/platform/enforce.h"
 
@@ -71,9 +72,10 @@ void TensorRTEngine::FreezeNetwork() {
   for (auto& item : buffer_sizes_) {
     if (item.second == 0) {
       auto slot_offset = infer_engine_->getBindingIndex(item.first.c_str());
+      auto dims = infer_engine_->getBindingDimensions(slot_offset);
       item.second = kDataTypeSize[static_cast<int>(
                         infer_engine_->getBindingDataType(slot_offset))] *
-                    AccumDims(infer_engine_->getBindingDimensions(slot_offset));
+                    analysis::AccuDims(dims.d, dims.nbDims);
     }
     auto& buf = buffer(item.first);
     CHECK(buf.buffer == nullptr);  // buffer should be allocated only once.
@@ -85,14 +87,15 @@ void TensorRTEngine::FreezeNetwork() {
 
 nvinfer1::ITensor* TensorRTEngine::DeclareInput(const std::string& name,
                                                 nvinfer1::DataType dtype,
-                                                const nvinfer1::Dims& dim) {
+                                                const nvinfer1::Dims& dims) {
   PADDLE_ENFORCE_EQ(0, buffer_sizes_.count(name), "duplicate input name %s",
                     name);
 
   PADDLE_ENFORCE(infer_network_ != nullptr, "should initnetwork first");
-  auto* input = infer_network_->addInput(name.c_str(), dtype, dim);
+  auto* input = infer_network_->addInput(name.c_str(), dtype, dims);
   PADDLE_ENFORCE(input, "infer network add input %s failed", name);
-  buffer_sizes_[name] = kDataTypeSize[static_cast<int>(dtype)] * AccumDims(dim);
+  buffer_sizes_[name] = kDataTypeSize[static_cast<int>(dtype)] *
+                        analysis::AccuDims(dims.d, dims.nbDims);
   TensorRTEngine::SetITensor(name, input);
   return input;
 }
@@ -162,13 +165,13 @@ void TensorRTEngine::SetInputFromCPU(const std::string& name, void* data,
 void TensorRTEngine::SetITensor(const std::string& name,
                                 nvinfer1::ITensor* tensor) {
   PADDLE_ENFORCE(tensor != nullptr);
-  PADDLE_ENFORCE_EQ(0, itensor_map_.count(name), "duplicate itensor name %s",
+  PADDLE_ENFORCE_EQ(0, itensor_map_.count(name), "duplicate ITensor name %s",
                     name);
   itensor_map_[name] = tensor;
 }
 
 nvinfer1::ITensor* TensorRTEngine::GetITensor(const std::string& name) {
-  PADDLE_ENFORCE(itensor_map_.count(name), "no itensor %s", name);
+  PADDLE_ENFORCE(itensor_map_.count(name), "no ITensor %s", name);
   return itensor_map_[name];
 }
 

paddle/fluid/inference/tensorrt/helper.h

@@ -26,15 +26,6 @@ namespace tensorrt {
 
 namespace dy = paddle::platform::dynload;
 
-static size_t AccumDims(nvinfer1::Dims dims) {
-  size_t num = dims.nbDims == 0 ? 0 : 1;
-  for (int i = 0; i < dims.nbDims; i++) {
-    PADDLE_ENFORCE_GT(dims.d[i], 0);
-    num *= dims.d[i];
-  }
-  return num;
-}
-
 // TensorRT data type to size
 const int kDataTypeSize[] = {
     4,  // kFLOAT

paddle/fluid/operators/CMakeLists.txt

@@ -200,7 +200,9 @@ if(WITH_DISTRIBUTE)
     op_library(send_vars_op DEPS ${DISTRIBUTE_DEPS})
     set_source_files_properties(send_vars_op.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
     op_library(send_barrier_op DEPS ${DISTRIBUTE_DEPS})
+    op_library(fetch_barrier_op DEPS ${DISTRIBUTE_DEPS})
     set_source_files_properties(send_barrier_op.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
+    set_source_files_properties(fetch_barrier_op.cc PROPERTIES COMPILE_FLAGS ${DISTRIBUTE_COMPILE_FLAGS})
     #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 SERIAL)
@@ -214,7 +216,7 @@ if(WITH_DISTRIBUTE)
         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 gen_nccl_id_op)
+    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()
 
 op_library(cross_entropy_op DEPS cross_entropy)

paddle/fluid/operators/detail/grpc_client.cc

@@ -25,6 +25,21 @@ namespace paddle {
 namespace operators {
 namespace detail {
 
+std::once_flag RPCClient::init_flag_;
+
+std::unique_ptr<RPCClient> RPCClient::rpc_client_(nullptr);
+
+RPCClient* RPCClient::GetInstance() {
+  std::call_once(init_flag_, &RPCClient::Init);
+  return rpc_client_.get();
+}
+
+void RPCClient::Init() {
+  if (rpc_client_.get() == nullptr) {
+    rpc_client_.reset(new RPCClient());
+  }
+}
+
 bool RPCClient::AsyncSendVariable(const std::string& ep,
                                   const platform::DeviceContext& ctx,
                                   const framework::Scope& scope,
@@ -60,7 +75,6 @@ bool RPCClient::AsyncSendVariable(const std::string& ep,
     call->StartCall();
     call->Finish(&s->reply_, &s->status_, reinterpret_cast<void*>(s));
   });
-
   req_count_++;
 
   return true;
@@ -249,8 +263,9 @@ bool RPCClient::Proceed() {
   delete c;
   return true;
 }
-
 std::shared_ptr<grpc::Channel> RPCClient::GetChannel(const std::string& ep) {
+  // TODO(Yancey1989): make grpc client completely thread-safe
+  std::unique_lock<std::mutex> lock(mutex_);
   auto it = channels_.find(ep);
   if (it != channels_.end()) {
     return it->second;
@@ -263,7 +278,6 @@ std::shared_ptr<grpc::Channel> RPCClient::GetChannel(const std::string& ep) {
 
   auto ch =
       grpc::CreateCustomChannel(ep, grpc::InsecureChannelCredentials(), args);
-
   channels_[ep] = ch;
   return ch;
 }

paddle/fluid/operators/detail/grpc_client.h

@@ -21,6 +21,7 @@ limitations under the License. */
 #include <functional>
 #include <iostream>
 #include <map>
+#include <mutex>  // NOLINT
 #include <string>
 #include <vector>
 
@@ -35,6 +36,7 @@ limitations under the License. */
 #include "paddle/fluid/framework/scope.h"
 #include "paddle/fluid/framework/selected_rows.h"
 #include "paddle/fluid/operators/detail/sendrecvop_utils.h"
+#include "paddle/fluid/platform/macros.h"  // for DISABLE_COPY_AND_ASSIGN
 
 namespace paddle {
 namespace operators {
@@ -161,6 +163,10 @@ class FetchBarrierProcessor : public BaseProcessor {
 
 class RPCClient {
  public:
+  RPCClient() {}
+
+  static RPCClient* GetInstance();
+
   bool AsyncSendVariable(const std::string& ep,
                          const platform::DeviceContext& ctx,
                          const framework::Scope& scope,
@@ -191,11 +197,17 @@ class RPCClient {
  private:
   bool Proceed();
   std::shared_ptr<grpc::Channel> GetChannel(const std::string& ep);
+  // Init is called by GetInstance.
+  static void Init();
 
  private:
   grpc::CompletionQueue cq_;
   std::map<std::string, std::shared_ptr<grpc::Channel>> channels_;
-  int64_t req_count_ = 0;
+  std::atomic<int64_t> req_count_{0};
+  std::mutex mutex_;
+  static std::unique_ptr<RPCClient> rpc_client_;
+  static std::once_flag init_flag_;
+  DISABLE_COPY_AND_ASSIGN(RPCClient);
 };
 
 }  // namespace detail

paddle/fluid/operators/detail/grpc_server.cc

@@ -177,11 +177,8 @@ class RequestPrefetch final : public RequestBase {
         program_(program),
         prefetch_ctx_(prefetch_ctx),
         req_id_(req_id) {
-    if (sync_mode_) {
-      request_.reset(new VariableResponse(scope, dev_ctx_, false));
-    } else {
-      request_.reset(new VariableResponse(scope, dev_ctx_, true));
-    }
+    // prefetch always create a new sub scope
+    request_.reset(new VariableResponse(scope, dev_ctx_, true));
     int method_id = static_cast<int>(detail::GrpcMethod::kPrefetchVariable);
     service_->RequestAsyncUnary(
         method_id, &ctx_, request_.get(), &responder_, cq_, cq_,
@@ -198,10 +195,10 @@ class RequestPrefetch final : public RequestBase {
     std::string var_name = request_->OutVarname();
     VLOG(3) << "RequestPrefetch " << var_name;
     auto var_desc = program_->Block(0).FindVar(var_name);
-    framework::Scope* local_scope = &scope_->NewScope();
+    framework::Scope* local_scope = request_->GetMutableLocalScope();
     auto* var = local_scope->FindVar(var_name);
     InitializeVariable(var, var_desc->GetType());
-    executor_->RunPreparedContext(prefetch_ctx_, scope_);
+    executor_->RunPreparedContext(prefetch_ctx_, local_scope);
 
     SerializeToByteBuffer(var_name, var, *dev_ctx_, &reply_);
 

paddle/fluid/operators/detail/grpc_server_test.cc

@@ -121,10 +121,10 @@ TEST(PREFETCH, DISABLED_CPU) {
   std::string in_var_name("ids");
   std::string out_var_name("out");
 
-  detail::RPCClient client;
-  client.AsyncPrefetchVariable("127.0.0.1:8889", ctx, scope, in_var_name,
-                               out_var_name);
-  client.Wait();
+  auto client = detail::RPCClient::GetInstance();
+  client->AsyncPrefetchVariable("127.0.0.1:8889", ctx, scope, in_var_name,
+                                out_var_name);
+  client->Wait();
 
   auto var = scope.Var(out_var_name);
   auto value = var->GetMutable<framework::SelectedRows>()->value();

paddle/fluid/operators/fetch_barrier_op.cc

+/* Copyright (c) 2016 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 <future>  // NOLINT
+#include <ostream>
+
+#include "paddle/fluid/framework/data_type.h"
+#include "paddle/fluid/framework/framework.pb.h"
+#include "paddle/fluid/framework/lod_tensor.h"
+#include "paddle/fluid/framework/op_registry.h"
+
+#include "paddle/fluid/operators/detail/grpc_client.h"
+#include "paddle/fluid/platform/profiler.h"
+
+namespace paddle {
+namespace operators {
+
+class FetchBarrierOp : public framework::OperatorBase {
+ public:
+  FetchBarrierOp(const std::string& type,
+                 const framework::VariableNameMap& inputs,
+                 const framework::VariableNameMap& outputs,
+                 const framework::AttributeMap& attrs)
+      : OperatorBase(type, inputs, outputs, attrs) {}
+
+  void RunImpl(const framework::Scope& scope,
+               const platform::Place& place) const override {
+    std::vector<std::string> eps = Attr<std::vector<std::string>>("endpoints");
+
+    platform::DeviceContextPool& pool = platform::DeviceContextPool::Instance();
+    auto& ctx = *pool.Get(place);
+    // For profiling
+    platform::RecordEvent record_event(Type(), &ctx);
+
+    auto rpc_client = detail::RPCClient::GetInstance();
+
+    PADDLE_ENFORCE(rpc_client->Wait());
+
+    for (auto& ep : eps) {
+      VLOG(3) << "fetch barrier, ep: " << ep;
+      rpc_client->AsyncSendFetchBarrier(ep);
+    }
+    PADDLE_ENFORCE(rpc_client->Wait());
+  }
+};
+
+class FetchBarrierOpMaker : public framework::OpProtoAndCheckerMaker {
+ public:
+  void Make() {
+    AddComment(R"DOC(
+SendBarrier operator
+
+This operator will send a send barrier signal to list_and_serv op, so that
+the Parameter Server would knew all variables have been sent.
+)DOC");
+
+    AddAttr<std::vector<std::string>>("endpoints",
+                                      "(string vector, default 127.0.0.1:6164)"
+                                      "Server endpoints to send variables to.")
+        .SetDefault({"127.0.0.1:6164"});
+  }
+};
+
+class FetchBarrierOpShapeInference : public framework::InferShapeBase {
+ public:
+  void operator()(framework::InferShapeContext* ctx) const override {}
+};
+
+}  // namespace operators
+}  // namespace paddle
+
+namespace ops = paddle::operators;
+
+REGISTER_OPERATOR(fetch_barrier, ops::FetchBarrierOp,
+                  paddle::framework::EmptyGradOpMaker, ops::FetchBarrierOpMaker,
+                  ops::FetchBarrierOpShapeInference);

paddle/fluid/operators/listen_and_serv_op.cc

@@ -13,8 +13,9 @@ See the License for the specific language governing permissions and
 limitations under the License. */
 
 #include <stdio.h>  // for removing the port file
+#include <csignal>
+#include <cstdlib>
 #include <fstream>
-#include <ostream>
 #include <thread>  // NOLINT
 #include <vector>
 
@@ -28,7 +29,6 @@ void RunServer(std::shared_ptr<detail::AsyncGRPCServer> service) {
   service->RunSyncUpdate();
   VLOG(4) << "RunServer thread end";
 }
-
 static void split(const std::string &str, char sep,
                   std::vector<std::string> *pieces) {
   pieces->clear();
@@ -59,7 +59,7 @@ static void ParallelExecuteBlocks(
           int run_block = idx;  // thread local
           try {
             executor->RunPreparedContext(prepared[run_block].get(), scope);
-          } catch (std::exception &e) {
+          } catch (const std::exception &e) {
             LOG(ERROR) << "run sub program error " << e.what();
           }
         }));
@@ -75,8 +75,11 @@ ListenAndServOp::ListenAndServOp(const std::string &type,
                                  const framework::AttributeMap &attrs)
     : OperatorBase(type, inputs, outputs, attrs) {}
 
+ListenAndServOp::~ListenAndServOp() { Stop(); }
+
 void ListenAndServOp::Stop() {
   rpc_service_->Push(LISTEN_TERMINATE_MESSAGE);
+  rpc_service_->ShutDown();
   server_thread_->join();
   auto file_path = string::Sprintf("/tmp/paddle.%d.port", ::getpid());
   remove(file_path.c_str());
@@ -122,7 +125,7 @@ void ListenAndServOp::RunSyncLoop(framework::Executor *executor,
   // Record received sparse variables, so that
   // we could reset those after execute optimize program
   std::vector<framework::Variable *> sparse_vars;
-  while (!exit_flag) {
+  while (!exit_flag && !SignalHandler::IsProgramExit()) {
     // Get from multiple trainers, we don't care about the order in which
     // the gradients arrives, just add suffix 0~n and merge the gradient.
     rpc_service_->SetCond(0);
@@ -187,7 +190,7 @@ void ListenAndServOp::RunSyncLoop(framework::Executor *executor,
     // mini-batch.
     // TODO(Yancey1989): move the reset action into an operator, we couldn't
     // have any hide logic in the operator.
-    for (auto &var : sparse_vars) {
+    for (framework::Variable *var : sparse_vars) {
       var->GetMutable<framework::SelectedRows>()->mutable_rows()->clear();
     }
 
@@ -204,9 +207,14 @@ static void AsyncUpdateThread(
     framework::Executor *executor,
     framework::ExecutorPrepareContext *prepared) {
   VLOG(3) << "update thread for " << var_name << " started";
-  while (!exit_flag) {
+  while (!exit_flag && !SignalHandler::IsProgramExit()) {
     const detail::ReceivedMessage v = queue->Pop();
+    if (SignalHandler::IsProgramExit()) {
+      VLOG(3) << "update thread for " << var_name << " exit";
+      break;
+    }
     auto recv_var_name = v.first;
+    VLOG(4) << "async update " << recv_var_name;
     auto var = v.second->GetVar();
     if (var == nullptr) {
       LOG(ERROR) << "Can not find server side var: " << recv_var_name;
@@ -216,7 +224,7 @@ static void AsyncUpdateThread(
       try {
         executor->RunPreparedContext(prepared,
                                      v.second->GetMutableLocalScope());
-      } catch (std::exception &e) {
+      } catch (const std::exception &e) {
         LOG(ERROR) << "run sub program error " << e.what();
       }
     });
@@ -235,7 +243,7 @@ void ListenAndServOp::RunAsyncLoop(framework::Executor *executor,
 
   auto grad_to_block_id_str =
       Attr<std::vector<std::string>>("grad_to_block_id");
-  for (auto &grad_and_id : grad_to_block_id_str) {
+  for (const auto &grad_and_id : grad_to_block_id_str) {
     std::vector<std::string> pieces;
     split(grad_and_id, ':', &pieces);
     VLOG(3) << "after split, grad = " << pieces[0] << ", id=" << pieces[1];
@@ -243,7 +251,11 @@ void ListenAndServOp::RunAsyncLoop(framework::Executor *executor,
     PADDLE_ENFORCE_EQ(grad_to_block_id.count(pieces[0]), 0);
     int block_id = std::stoi(pieces[1]);
     grad_to_block_id[pieces[0]] = block_id;
-    grad_to_queue[pieces[0]] = std::make_shared<detail::ReceivedQueue>();
+    std::shared_ptr<detail::ReceivedQueue> queue =
+        std::make_shared<detail::ReceivedQueue>();
+    grad_to_queue[pieces[0]] = queue;
+    // record blocking queue in SignalHandler
+    SignalHandler::RegisterBlockingQueue(queue);
     id_to_grad[block_id] = pieces[0];
   }
   size_t num_blocks = program->Size();
@@ -275,9 +287,8 @@ void ListenAndServOp::RunAsyncLoop(framework::Executor *executor,
                         executor, grad_to_prepared_ctx[grad_name].get());
     }));
   }
-
   VLOG(3) << "RunAsyncLoop into while";
-  while (!exit_flag) {
+  while (!exit_flag && !SignalHandler::IsProgramExit()) {
     const detail::ReceivedMessage v = rpc_service_->Get();
     auto recv_var_name = v.first;
     if (recv_var_name == LISTEN_TERMINATE_MESSAGE) {
@@ -332,6 +343,10 @@ void ListenAndServOp::RunImpl(const framework::Scope &scope,
   VLOG(3) << "wait server thread to become ready...";
   rpc_service_->WaitServerReady();
 
+  // register SIGINT(from ctrl+C) and SIGTERM(from kill) signal handlers
+  signal(SIGINT, SignalHandler::StopAndExit);
+  signal(SIGTERM, SignalHandler::StopAndExit);
+
   // Write to a file of server selected port for python use.
   std::string file_path = string::Sprintf("/tmp/paddle.%d.selected_port",
                                           static_cast<int>(::getpid()));
@@ -347,12 +362,9 @@ class ListenAndServOpMaker : public framework::OpProtoAndCheckerMaker {
  public:
   void Make() {
     AddInput("X", "(Tensor) Variables that server recv.").AsDuplicable();
-    AddComment(R"DOC(
-ListenAndServ operator
-
-This operator will start a RPC server which can receive variables
-from send_op and send back variables to recv_op.
-)DOC");
+    AddComment(R"DOC(" + "ListenAndServ operator" + "\n" + "This operator" +
+" will start a RPC server which can receive variables from send_op and send" +
+"back variables to recv_op.)DOC");
     AddAttr<std::string>("endpoint",
                          "(string, default 127.0.0.1:6164)"
                          "IP address to listen on.")
@@ -373,6 +385,29 @@ from send_op and send back variables to recv_op.
   }
 };
 
+bool SignalHandler::program_exit_flag_ = false;
+
+SignalHandler::BlockingQueueSet SignalHandler::blocking_queue_set_{};
+
+void SignalHandler::StopAndExit(int signal_num) {
+  VLOG(3) << "Catch interrupt signal: " << signal_num << ", program will exit";
+
+  program_exit_flag_ = true;
+
+  // awake all blocking queues
+  for (BlockingQueueSet::iterator iter = blocking_queue_set_.begin();
+       iter != blocking_queue_set_.end(); iter++) {
+    iter->get()->Push(
+        std::make_pair(std::string(LISTEN_TERMINATE_MESSAGE), nullptr));
+  }
+
+  exit(EXIT_SUCCESS);
+}
+
+void SignalHandler::RegisterBlockingQueue(BlockingQueue &queue) {
+  blocking_queue_set_.insert(queue);
+}
+
 }  // namespace operators
 }  // namespace paddle
 

paddle/fluid/operators/listen_and_serv_op.h

@@ -16,7 +16,7 @@ limitations under the License. */
 
 #include <stdint.h>
 #include <atomic>
-#include <ostream>
+#include <set>
 #include <string>
 
 #include "paddle/fluid/framework/executor.h"
@@ -40,6 +40,8 @@ class ListenAndServOp : public framework::OperatorBase {
                   const framework::VariableNameMap& outputs,
                   const framework::AttributeMap& attrs);
 
+  virtual ~ListenAndServOp();
+
   void RunSyncLoop(framework::Executor* executor,
                    framework::ProgramDesc* program,
                    framework::Scope* recv_scope,
@@ -68,5 +70,25 @@ class ListenAndServOp : public framework::OperatorBase {
   static std::atomic_int selected_port_;
 };
 
+class SignalHandler {
+ public:
+  typedef std::shared_ptr<detail::ReceivedQueue> BlockingQueue;
+  typedef std::unordered_set<BlockingQueue> BlockingQueueSet;
+
+ public:
+  static void StopAndExit(int signal_num);
+
+  static void RegisterBlockingQueue(BlockingQueue&);
+
+  static inline bool IsProgramExit() { return program_exit_flag_; }
+
+ private:
+  static bool program_exit_flag_;
+
+  static BlockingQueueSet blocking_queue_set_;
+
+  DISABLE_COPY_AND_ASSIGN(SignalHandler);
+};
+
 }  // namespace operators
 }  // namespace paddle

paddle/fluid/operators/lookup_sparse_table_op.cc

@@ -127,7 +127,7 @@ class LookupSparseTableOpMaker : public framework::OpProtoAndCheckerMaker {
         .SetDefault(-1.0f);
     AddAttr<float>("max",
                    "(float, default 1.0) "
-                   "Maximun value of uniform random")
+                   "Maximum value of uniform random")
         .SetDefault(1.0f);
     AddAttr<int>("seed",
                  "(int, default 0) "

paddle/fluid/operators/math/cross_entropy.cc

@@ -46,7 +46,10 @@ class CrossEntropyFunctor<platform::CPUDeviceContext, T> {
 
       const int64_t* label_data = labels->data<int64_t>();
       for (int i = 0; i < batch_size; ++i) {
-        int index = i * class_num + label_data[i];
+        int lbl = label_data[i];
+        PADDLE_ENFORCE_GE(lbl, 0);
+        PADDLE_ENFORCE_LT(lbl, class_num);
+        int index = i * class_num + lbl;
         loss_data[i] = -math::TolerableValue<T>()(std::log(prob_data[index]));
       }
     }

paddle/fluid/operators/prefetch_op.cc

@@ -41,12 +41,7 @@ class PrefetchOp : public framework::OperatorBase {
     platform::DeviceContextPool& pool = platform::DeviceContextPool::Instance();
     auto& ctx = *pool.Get(place);
 
-    auto client_var_name = Output("RPCClient");
-    PADDLE_ENFORCE_NOT_NULL(scope.FindVar(client_var_name),
-                            "Can not find variable '%s' in the scope.",
-                            client_var_name);
-    auto* client_var = scope.FindVar(client_var_name);
-    detail::RPCClient* rpc_client = client_var->GetMutable<detail::RPCClient>();
+    auto rpc_client = detail::RPCClient::GetInstance();
 
     for (size_t i = 0; i < ins.size(); i++) {
       if (NeedSend(scope, ins[i])) {
@@ -66,9 +61,6 @@ class PrefetchOpMaker : public framework::OpProtoAndCheckerMaker {
  public:
   void Make() {
     AddInput("X", "(LoDTensor) Input Id variables to be sent").AsDuplicable();
-    AddOutput("RPCClient",
-              "(RPCClient) The RPC client object which will be"
-              "initialized at most once.");
     AddOutput("Out",
               "(LoDTensor) result "
               "to be fetched from parameter server")
@@ -87,17 +79,6 @@ the parameter server and fetch result back.
   }
 };
 
-class PrefetchOpVarTypeInference : public framework::VarTypeInference {
- public:
-  void operator()(const framework::OpDesc& op_desc,
-                  framework::BlockDesc* block) const override {
-    auto out_var_name = op_desc.Output("RPCClient").front();
-    auto& out_var = block->FindRecursiveOrCreateVar(out_var_name);
-    auto var_type = framework::proto::VarType::RAW;
-    out_var.SetType(var_type);
-  }
-};
-
 class PrefetchOpShapeInference : public framework::InferShapeBase {
  public:
   void operator()(framework::InferShapeContext* ctx) const override {}
@@ -110,5 +91,4 @@ namespace ops = paddle::operators;
 
 REGISTER_OPERATOR(prefetch, ops::PrefetchOp,
                   paddle::framework::EmptyGradOpMaker, ops::PrefetchOpMaker,
-                  ops::PrefetchOpVarTypeInference,
                   ops::PrefetchOpShapeInference);

paddle/fluid/operators/recv_op.cc

@@ -21,6 +21,7 @@ limitations under the License. */
 #include "paddle/fluid/framework/op_registry.h"
 
 #include "paddle/fluid/operators/detail/grpc_client.h"
+#include "paddle/fluid/platform/profiler.h"
 
 namespace paddle {
 namespace operators {
@@ -36,19 +37,23 @@ class RecvOp : public framework::OperatorBase {
                const platform::Place& place) const override {
     auto outs = Outputs("Out");
     std::vector<std::string> epmap = Attr<std::vector<std::string>>("epmap");
+    int sync_mode = Attr<int>("sync_mode");
 
     platform::DeviceContextPool& pool = platform::DeviceContextPool::Instance();
     auto& ctx = *pool.Get(place);
+    // For profiling
+    platform::RecordEvent record_event(Type(), &ctx);
+
+    auto rpc_client = detail::RPCClient::GetInstance();
 
     for (size_t i = 0; i < outs.size(); i++) {
-      VLOG(3) << "getting " << outs[i];
-      client_.AsyncGetVariable(epmap[i], ctx, scope, outs[i]);
+      VLOG(3) << "getting " << outs[i] << " from " << epmap[i];
+      rpc_client->AsyncGetVariable(epmap[i], ctx, scope, outs[i]);
+    }
+    if (sync_mode) {
+      PADDLE_ENFORCE(rpc_client->Wait());
     }
-    PADDLE_ENFORCE(client_.Wait());
   }
-
- private:
-  mutable detail::RPCClient client_;
 };
 
 class RecvOpMaker : public framework::OpProtoAndCheckerMaker {
@@ -65,6 +70,10 @@ This operator can get variables from server side.
                                       "Server endpoints in the order of input "
                                       "variables for mapping")
         .SetDefault({});
+    AddAttr<int>("sync_mode",
+                 "(int, default 0)"
+                 "sync recv or async recv.")
+        .SetDefault(0);
   }
 };
 

paddle/fluid/operators/send_barrier_op.cc

@@ -21,6 +21,7 @@ limitations under the License. */
 #include "paddle/fluid/framework/op_registry.h"
 
 #include "paddle/fluid/operators/detail/grpc_client.h"
+#include "paddle/fluid/platform/profiler.h"
 
 namespace paddle {
 namespace operators {
@@ -36,31 +37,30 @@ class SendBarrierOp : public framework::OperatorBase {
   void RunImpl(const framework::Scope& scope,
                const platform::Place& place) const override {
     std::vector<std::string> eps = Attr<std::vector<std::string>>("endpoints");
+    bool sync_mode = Attr<bool>("sync_mode");
 
-    auto client_var_name = Output("RPCClient");
-    PADDLE_ENFORCE_NOT_NULL(scope.FindVar(client_var_name),
-                            "Can not find variable '%s' in the scope.",
-                            client_var_name);
-    auto* client_var = scope.FindVar(client_var_name);
-    detail::RPCClient* rpc_client = client_var->GetMutable<detail::RPCClient>();
+    platform::DeviceContextPool& pool = platform::DeviceContextPool::Instance();
+    auto& ctx = *pool.Get(place);
+    // For profiling
+    platform::RecordEvent record_event(Type(), &ctx);
+
+    auto rpc_client = detail::RPCClient::GetInstance();
 
     // need to wait before sending send_barrier message
     PADDLE_ENFORCE(rpc_client->Wait());
-
-    for (auto& ep : eps) {
-      VLOG(3) << "send barrier, ep: " << ep;
-      rpc_client->AsyncSendBatchBarrier(ep);
+    if (sync_mode) {
+      for (auto& ep : eps) {
+        VLOG(3) << "send barrier, ep: " << ep;
+        rpc_client->AsyncSendBatchBarrier(ep);
+      }
+      PADDLE_ENFORCE(rpc_client->Wait());
     }
-    PADDLE_ENFORCE(rpc_client->Wait());
   }
 };
 
 class SendBarrierOpMaker : public framework::OpProtoAndCheckerMaker {
  public:
   void Make() {
-    AddOutput("RPCClient",
-              "(RPCClient) The RPC client object which is"
-              "initialized at most once.");
     AddComment(R"DOC(
 SendBarrier operator
 
@@ -72,17 +72,7 @@ the Parameter Server would knew all variables have been sent.
                                       "(string vector, default 127.0.0.1:6164)"
                                       "Server endpoints to send variables to.")
         .SetDefault({"127.0.0.1:6164"});
-  }
-};
-
-class SendBarrierOpVarTypeInference : public framework::VarTypeInference {
- public:
-  void operator()(const framework::OpDesc& op_desc,
-                  framework::BlockDesc* block) const override {
-    auto out_var_name = op_desc.Output("RPCClient").front();
-    auto& out_var = block->FindRecursiveOrCreateVar(out_var_name);
-    auto var_type = framework::proto::VarType::RAW;
-    out_var.SetType(var_type);
+    AddAttr<bool>("sync_mode", "work in sync_mode or not").SetDefault(true);
   }
 };
 
@@ -98,5 +88,4 @@ namespace ops = paddle::operators;
 
 REGISTER_OPERATOR(send_barrier, ops::SendBarrierOp,
                   paddle::framework::EmptyGradOpMaker, ops::SendBarrierOpMaker,
-                  ops::SendBarrierOpVarTypeInference,
                   ops::SendBarrierOpShapeInference);

paddle/fluid/operators/send_op.cc

@@ -49,12 +49,7 @@ class SendOp : public framework::OperatorBase {
     // For profiling
     platform::RecordEvent record_event(Type(), &ctx);
 
-    auto client_var_name = Output("RPCClient");
-    PADDLE_ENFORCE_NOT_NULL(scope.FindVar(client_var_name),
-                            "Can not find variable '%s' in the scope.",
-                            client_var_name);
-    auto* client_var = scope.FindVar(client_var_name);
-    detail::RPCClient* rpc_client = client_var->GetMutable<detail::RPCClient>();
+    auto rpc_client = detail::RPCClient::GetInstance();
 
     for (size_t i = 0; i < ins.size(); i++) {
       if (NeedSend(scope, ins[i])) {
@@ -96,9 +91,6 @@ class SendOpMaker : public framework::OpProtoAndCheckerMaker {
     AddInput("X", "(Tensor) Input tensor to be sent").AsDuplicable();
     AddOutput("Out", "(Tensor) Output tensor to be received from server")
         .AsDuplicable();
-    AddOutput("RPCClient",
-              "(RPCClient) The RPC client object which is"
-              "initialized at most once.");
     AddComment(R"DOC(
 Send operator
 
@@ -119,17 +111,6 @@ This operator will send tensor to recv_op at the parameter server.
   }
 };
 
-class SendOpVarTypeInference : public framework::VarTypeInference {
- public:
-  void operator()(const framework::OpDesc& op_desc,
-                  framework::BlockDesc* block) const override {
-    auto out_var_name = op_desc.Output("RPCClient").front();
-    auto& out_var = block->FindRecursiveOrCreateVar(out_var_name);
-    auto var_type = framework::proto::VarType::RAW;
-    out_var.SetType(var_type);
-  }
-};
-
 class SendOpShapeInference : public framework::InferShapeBase {
  public:
   void operator()(framework::InferShapeContext* ctx) const override {}
@@ -141,5 +122,4 @@ class SendOpShapeInference : public framework::InferShapeBase {
 namespace ops = paddle::operators;
 
 REGISTER_OPERATOR(send, ops::SendOp, paddle::framework::EmptyGradOpMaker,
-                  ops::SendOpMaker, ops::SendOpVarTypeInference,
-                  ops::SendOpShapeInference);
+                  ops::SendOpMaker, ops::SendOpShapeInference);

paddle/fluid/operators/send_recv_op_test.cc

@@ -156,6 +156,7 @@ TEST(SendRecvOp, CPUDense) {
   std::thread server_thread(StartServerNet, false, &initialized);
   while (!initialized) {
   }
+
   static_cast<paddle::operators::ListenAndServOp *>(listen_and_serv_op.get())
       ->WaitServerReady();
 
@@ -175,9 +176,10 @@ TEST(SendRecvOp, CPUDense) {
   std::string endpoint = paddle::string::Sprintf("127.0.0.1:%d", selected_port);
   attrs.insert({"endpoints", std::vector<std::string>({endpoint})});
   attrs.insert({"epmap", std::vector<std::string>({endpoint})});
-  auto send_op = f::OpRegistry::CreateOp(
-      "send", {{"X", {"x1"}}},
-      {{"Out", {"Out"}}, {"RPCClient", {"RPC_CLIENT_VAR"}}}, attrs);
+  const f::VariableNameMap &inputs = {{"X", {"x1"}}};
+  const f::VariableNameMap &outputs = {{"Out", {"Out"}}};
+
+  auto send_op = f::OpRegistry::CreateOp("send", inputs, outputs, attrs);
   send_op->Run(scope, place);
 
   auto in_var = scope.Var("x1");
@@ -220,9 +222,8 @@ TEST(SendRecvOp, CPUSparse) {
   std::string endpoint = paddle::string::Sprintf("127.0.0.1:%d", selected_port);
   attrs.insert({"endpoints", std::vector<std::string>({endpoint})});
   attrs.insert({"epmap", std::vector<std::string>({endpoint})});
-  auto send_op = f::OpRegistry::CreateOp(
-      "send", {{"X", {"x1"}}},
-      {{"Out", {"Out"}}, {"RPCClient", {"RPC_CLIENT_VAR"}}}, attrs);
+  auto send_op = f::OpRegistry::CreateOp("send", {{"X", {"x1"}}},
+                                         {{"Out", {"Out"}}}, attrs);
   send_op->Run(scope, place);
 
   auto x0 = scope.Var("x0")->GetMutable<f::SelectedRows>();

paddle/fluid/operators/send_recv_util.h

@@ -20,6 +20,9 @@ namespace operators {
 
 inline bool NeedSend(const framework::Scope& scope,
                      const std::string& varname) {
+  // dummy variable is only used in parallel executor to represent
+  // some dependency relationship, we don't need to send/recv it.
+  if (varname == "dummy") return false;
   auto* var = scope.FindVar(varname);
   PADDLE_ENFORCE_NOT_NULL(var, "Can not find variable '%s' in the send side.",
                           varname);

paddle/fluid/operators/send_vars_op.cc

@@ -20,6 +20,7 @@ limitations under the License. */
 #include "paddle/fluid/framework/op_registry.h"
 #include "paddle/fluid/operators/detail/grpc_client.h"
 #include "paddle/fluid/operators/send_recv_util.h"
+#include "paddle/fluid/platform/profiler.h"
 
 namespace paddle {
 namespace operators {
@@ -41,12 +42,10 @@ class SendVarsOp : public framework::OperatorBase {
     platform::DeviceContextPool& pool = platform::DeviceContextPool::Instance();
     auto& ctx = *pool.Get(place);
 
-    auto client_var_name = Output("RPCClient");
-    PADDLE_ENFORCE_NOT_NULL(scope.FindVar(client_var_name),
-                            "Can not find variable '%s' in the scope.",
-                            client_var_name);
-    auto* client_var = scope.FindVar(client_var_name);
-    detail::RPCClient* rpc_client = client_var->GetMutable<detail::RPCClient>();
+    // For profiling
+    platform::RecordEvent record_event(Type(), &ctx);
+
+    auto rpc_client = detail::RPCClient::GetInstance();
 
     for (size_t i = 0; i < ins.size(); i++) {
       if (NeedSend(scope, ins[i])) {
@@ -69,9 +68,6 @@ class SendVarsOpMaker : public framework::OpProtoAndCheckerMaker {
   void Make() {
     AddInput("X", "(Tensor, SelectedRows) Input variables to be sent")
         .AsDuplicable();
-    AddOutput("RPCClient",
-              "(RPCClient) The RPC client object which will be"
-              "initialized at most once.");
     AddComment(R"DOC(
 Send operator
 
@@ -89,17 +85,6 @@ This operator will send variables to listen_and_serve op at the parameter server
   }
 };
 
-class SendVarsOpVarTypeInference : public framework::VarTypeInference {
- public:
-  void operator()(const framework::OpDesc& op_desc,
-                  framework::BlockDesc* block) const override {
-    auto out_var_name = op_desc.Output("RPCClient").front();
-    auto& out_var = block->FindRecursiveOrCreateVar(out_var_name);
-    auto var_type = framework::proto::VarType::RAW;
-    out_var.SetType(var_type);
-  }
-};
-
 class SendVarsOpShapeInference : public framework::InferShapeBase {
  public:
   void operator()(framework::InferShapeContext* ctx) const override {}
@@ -112,5 +97,4 @@ namespace ops = paddle::operators;
 
 REGISTER_OPERATOR(send_vars, ops::SendVarsOp,
                   paddle::framework::EmptyGradOpMaker, ops::SendVarsOpMaker,
-                  ops::SendVarsOpVarTypeInference,
                   ops::SendVarsOpShapeInference);

paddle/fluid/operators/sgd_op.h

@@ -96,8 +96,12 @@ class SGDOpKernel : public framework::OpKernel<T> {
         return;
       }
 
-      size_t param_row_width = param.value().numel() / param.rows().size();
-      size_t grad_row_width = grad.value().numel() / grad.rows().size();
+      auto param_row_width = param.value().dims()[1];
+      auto grad_row_width = grad.value().dims()[1];
+      VLOG(4) << " param rows: " << param.rows().size()
+              << " param memory rows: " << param.value().dims()[0]
+              << " grad rows: " << grad.rows().size()
+              << " grad memory rows: " << grad.value().dims()[0];
       PADDLE_ENFORCE_EQ(param_row_width, grad_row_width,
                         "param_row should have the same size with grad_row");
 

paddle/fluid/pybind/const_value.cc

@@ -32,7 +32,8 @@ void BindConstValue(pybind11::module* m) {
       .value("Forward", framework::OpRole::kForward)
       .value("Backward", framework::OpRole::kBackward)
       .value("Optimize", framework::OpRole::kOptimize)
-      .value("Loss", framework::OpRole::kLoss);
+      .value("Loss", framework::OpRole::kLoss)
+      .value("RPC", framework::OpRole::kRPC);
 
   op_proto_and_checker_maker.def(
       "kOpRoleAttrName", framework::OpProtoAndCheckerMaker::OpRoleAttrName);

paddle/function/BlockExpandOp.cpp

@@ -33,7 +33,7 @@ namespace paddle {
  * \param outputs[0] Image data of NCHW format.
  */
 class BlockExpandFunction : public FunctionBase {
-public:
+ public:
   void init(const FuncConfig& config) override {
     // function arguments
     strides_ = config.get<std::vector<size_t>>("strides");
@@ -81,7 +81,7 @@ public:
                         (size_t)blockW()});
   }
 
-protected:
+ protected:
   std::vector<size_t> strides_;
   std::vector<size_t> paddings_;
   std::vector<size_t> blocks_;
@@ -101,7 +101,7 @@ protected:
 
 template <DeviceType Device>
 class BlockExpandForward : public BlockExpandFunction {
-public:
+ public:
   void init(const FuncConfig& config) override {
     BlockExpandFunction::init(config);
   }
@@ -149,7 +149,7 @@ public:
 
 template <DeviceType Device>
 class BlockExpandBackward : public BlockExpandFunction {
-public:
+ public:
   void init(const FuncConfig& config) override {
     BlockExpandFunction::init(config);
   }

paddle/function/BufferArg.h

@@ -63,12 +63,12 @@ enum ArgType {
   ADD_TO = 2,
 };
 class BufferArg {
-public:
+ public:
   void setArgType(ArgType argType) { argType_ = argType; }
 
   ArgType getArgType() const { return argType_; }
 
-public:
+ public:
   BufferArg(ValueType valueType,
             const TensorShape& shape,
             ArgType argType = UNSPECIFIED)
@@ -169,7 +169,7 @@ public:
   const SequenceArg& sequence() const;
   const SparseMatrixArg& sparse() const;
 
-protected:
+ protected:
   void* buf_;
   ValueType valueType_;
   TensorShape shape_;
@@ -185,7 +185,7 @@ protected:
 // valueType_ = int32
 // if a < b then value_.buf_[a] < value_.buf_[b]
 class SequenceIdArg : public BufferArg {
-public:
+ public:
   SequenceIdArg(const TensorShape& shape, ArgType argType = UNSPECIFIED)
       : BufferArg(VALUE_TYPE_INT32, shape, argType) {
     bufferType_ = TENSOR_SEQUENCE_ID;
@@ -212,7 +212,7 @@ public:
 
   size_t numSeqs() const { return numSeqs_; }
 
-private:
+ private:
   size_t numSeqs_;
 };
 
@@ -222,7 +222,7 @@ private:
 // SequenceArg can be used to represent sequences that contain multiple
 // unequal lengths.
 class SequenceArg : public BufferArg {
-public:
+ public:
   SequenceArg(ValueType valueType,
               const TensorShape& shape,
               ArgType argType = UNSPECIFIED)
@@ -255,7 +255,7 @@ public:
   SequenceIdArg& getSequenceId() { return startPositions_; }
   const SequenceIdArg& getSequenceId() const { return startPositions_; }
 
-private:
+ private:
   SequenceIdArg startPositions_;
 };
 
@@ -263,7 +263,7 @@ private:
 // valueType_ == float or double
 // shape_.ndims() == 2
 class SparseMatrixArg : public BufferArg {
-public:
+ public:
   SparseMatrixArg(void* buf,
                   ValueType valueType,
                   const TensorShape& shape,
@@ -353,7 +353,7 @@ public:
 
   SparseDataType dataType() const { return type_; }
 
-private:
+ private:
   BufferArg row_;
   BufferArg col_;
   size_t nnz_;

paddle/function/ContextProjectionOp.cpp

@@ -100,7 +100,7 @@ void ContextProjectionForward<DEVICE_TYPE_CPU>(CpuMatrix& out_mat,
  */
 template <DeviceType Device>
 class ContextProjectionForwardFunc : public FunctionBase {
-public:
+ public:
   void init(const FuncConfig& config) override {
     context_length_ = config.get<size_t>("context_length");
     context_start_ = config.get<int>("context_start");
@@ -146,7 +146,7 @@ public:
                                      begin_pad_);
   }
 
-private:
+ private:
   size_t context_length_;
   int context_start_;
   size_t begin_pad_;
@@ -223,7 +223,7 @@ void ContextProjectionBackward<DEVICE_TYPE_CPU>(const CpuMatrix& out_grad_mat,
  */
 template <DeviceType Device>
 class ContextProjectionBackwardFunc : public FunctionBase {
-public:
+ public:
   void init(const FuncConfig& config) override {
     context_length_ = config.get<size_t>("context_length");
     context_start_ = config.get<int>("context_start");
@@ -278,7 +278,7 @@ public:
                                       total_pad_);
   }
 
-private:
+ private:
   size_t context_length_;
   int context_start_;
   size_t begin_pad_;
@@ -299,7 +299,7 @@ private:
  */
 template <DeviceType Device>
 class ContextProjectionBackwardDataFunc : public FunctionBase {
-public:
+ public:
   void init(const FuncConfig& config) override {
     context_length_ = config.get<size_t>("context_length");
     context_start_ = config.get<int>("context_start");
@@ -331,7 +331,7 @@ public:
         out_grad_mat, in_grad_mat, seq_vec, context_length_, context_start_);
   }
 
-private:
+ private:
   size_t context_length_;
   int context_start_;
 };
@@ -348,7 +348,7 @@ private:
  */
 template <DeviceType Device>
 class ContextProjectionBackwardWeightFunc : public FunctionBase {
-public:
+ public:
   void init(const FuncConfig& config) override {
     context_length_ = config.get<size_t>("context_length");
     context_start_ = config.get<int>("context_start");
@@ -382,7 +382,7 @@ public:
                                             begin_pad_);
   }
 
-private:
+ private:
   size_t context_length_;
   int context_start_;
   size_t begin_pad_;

paddle/function/ConvOp.h

@@ -56,7 +56,7 @@ namespace paddle {
  *      H and W is height and width of filter.
  */
 class ConvFunctionBase : public FunctionBase {
-public:
+ public:
   void init(const FuncConfig& config) override {
     // function arguments
     strides_ = config.get<std::vector<size_t>>("strides");
@@ -101,7 +101,7 @@ public:
     }
   }
 
-protected:
+ protected:
   size_t getFilterHeight(const TensorShape& filter) const {
     return filter[filter.ndims() - 2];
   }

paddle/function/CosSimOp.cpp

@@ -97,7 +97,7 @@ class CosSimForwardFunc : public FunctionBase {
     CosSimForward<Device>(out_mat, in1_mat, in2_mat, scale_);
   }
 
-private:
+ private:
   real scale_;
 };
 
@@ -227,7 +227,7 @@ class CosSimBackwardFunc : public FunctionBase {
         out_grad, out_val, in1_val, in2_val, in1_grad, in2_grad, scale_);
   }
 
-private:
+ private:
   real scale_;
 };
 

paddle/function/CropOp.cpp

@@ -112,7 +112,7 @@ void CropGrad<DEVICE_TYPE_CPU>(const real* inGrad,
  */
 template <DeviceType Device>
 class CropFunc : public FunctionBase {
-public:
+ public:
   void init(const FuncConfig& config) override { conf_ = config; }
 
   void calc(const BufferArgs& inputs, const BufferArgs& outputs) override {
@@ -130,7 +130,7 @@ public:
                  conf_);
   }
 
-private:
+ private:
   FuncConfig conf_;
 };
 
@@ -145,7 +145,7 @@ private:
 
 template <DeviceType Device>
 class CropGradFunc : public FunctionBase {
-public:
+ public:
   void init(const FuncConfig& config) override { conf_ = config; }
 
   void calc(const BufferArgs& inputs, const BufferArgs& outputs) override {
@@ -163,7 +163,7 @@ public:
                      conf_);
   }
 
-private:
+ private:
   FuncConfig conf_;
 };
 

paddle/function/CrossMapNormalOp.cpp

@@ -160,7 +160,7 @@ void CrossMapNormalGrad<DEVICE_TYPE_CPU>(real* inputsGrad,
  */
 template <DeviceType Device>
 class CrossMapNormalFunc : public FunctionBase {
-public:
+ public:
   void init(const FuncConfig& config) override {
     // function arguments
     size_ = config.get<size_t>("size");
@@ -220,7 +220,7 @@ public:
     return ops;
   }
 
-private:
+ private:
   size_t size_;
   real scale_;
   real pow_;
@@ -260,7 +260,7 @@ private:
  */
 template <DeviceType Device>
 class CrossMapNormalGradFunc : public FunctionBase {
-public:
+ public:
   void init(const FuncConfig& config) override {
     // function arguments
     size_ = config.get<size_t>("size");
@@ -328,7 +328,7 @@ public:
     return ops;
   }
 
-private:
+ private:
   size_t size_;
   real scale_;
   real pow_;

paddle/function/DepthwiseConvOp.h

@@ -44,7 +44,7 @@ namespace paddle {
  */
 template <DeviceType Device, class T>
 class DepthwiseConvFunctor {
-public:
+ public:
   void operator()(const T* inputData,
                   const T* filterData,
                   int batchSize,
@@ -89,7 +89,7 @@ public:
  */
 template <DeviceType Device, class T>
 class DepthwiseConvGradInputFunctor {
-public:
+ public:
   void operator()(const T* outputGrad,
                   const T* filterData,
                   int batchSize,
@@ -135,7 +135,7 @@ public:
  */
 template <DeviceType Device, class T>
 class DepthwiseConvGradFilterFunctor {
-public:
+ public:
   void operator()(const T* outputGrad,
                   const T* inputData,
                   int batchSize,

paddle/function/EigenThreadDevice.h

@@ -46,7 +46,7 @@ int GetCpuCount() { return 1; }
 #endif
 
 class EigenDeviceWarpper {
-public:  // NOLINT
+ public:  // NOLINT
 #if EIGEN_USE_THREADS
   static Eigen::ThreadPoolDevice* device() {
     const int num_cpus = GetCpuCount();