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

Author: NHZlX

benchmark/cluster/vgg16/vgg16_fluid.py

@@ -240,8 +240,6 @@ def train_loop(exe, trainer_prog):
 
         t = fluid.DistributeTranspiler()
         t.transpile(
-            optimize_ops,
-            params_grads,
             trainer_id=args.task_index,
             pservers=args.ps_hosts,
             trainers=trainers)

doc/v2/howto/cluster/multi_cluster/index_en.rst

@@ -8,28 +8,28 @@ The user's cluster environment is not the same. To facilitate everyone's deploym
 ..  toctree::
   :maxdepth: 1
 
-  k8s_cn.md
-  k8s_distributed_cn.md
+  k8s_en.md
+  k8s_distributed_en.md
 
 `OpenMPI <https://www.open-mpi.org>`_ is a mature high-performance parallel computing framework, which is widely used in the field of HPC. The following guide describes how to use OpenMPI to build PaddlePaddle's cluster training task:
 
 ..  toctree::
   :maxdepth: 1
 
-  openmpi_cn.md
+  openmpi_en.md
 
 `Fabric <http://www.fabfile.org>`_ is a convenient tool for program deployment and management. We provide a way to deploy and manage with Fabric. If you want to know more about it, please read the following guidelines:
 
 ..  toctree::
   :maxdepth: 1
 
-  fabric_cn.md
+  fabric_en.md
 
 We also support the deployment of PaddlePaddle on AWS. Learn more about:
 
 ..  toctree::
   :maxdepth: 1
 
-  k8s_aws_cn.md
+  k8s_aws_en.md
 
-The examples can be found under `cluster_train_v2 <https://github.com/PaddlePaddle/Paddle/tree/develop/paddle/scripts/cluster_train_v2>`_ .
+The examples can be found under `cluster_train_v2 <https://github.com/PaddlePaddle/Paddle/tree/develop/paddle/scripts/cluster_train_v2>`_ .

python/paddle/fluid/__init__.py

@@ -21,14 +21,15 @@
 from executor import *
 
 import trainer
-from trainer import *
+from trainer import Trainer
+from trainer import BeginEpochEvent
+from trainer import EndEpochEvent
+from trainer import BeginStepEvent
+from trainer import EndStepEvent
 
 import inferencer
 from inferencer import Inferencer
 
-import params
-from params import Params
-
 import io
 import evaluator
 import initializer
@@ -57,7 +58,7 @@
 Tensor = LoDTensor
 
 __all__ = framework.__all__ + executor.__all__ + concurrency.__all__ +\
-          trainer.__all__ + inferencer.__all__ + params.__all__ + [
+          trainer.__all__ + inferencer.__all__ + [
     'io',
     'initializer',
     'layers',

python/paddle/fluid/distribute_transpiler.py

@@ -137,8 +137,6 @@ def split_dense_variable(var_list,
 
 class DistributeTranspiler:
     def transpile(self,
-                  optimize_ops,
-                  params_grads,
                   trainer_id,
                   program=None,
                   pservers="127.0.0.1:6174",
@@ -169,11 +167,6 @@ def transpile(self,
             4. append ops that should run on current server instance.
             5. add listen_and_serv op
 
-            :param optimize_ops: op list of optimization, should be the
-                                    return value of Optimizer.minimize
-            :type optimize_ops: list
-            :param params_grads: list of tuple(weight, gradient)
-            :type params_grads: list
             :param trainer_id: one unique id for each trainer in a job.
             :type trainer_id: int
             :param program: program to transpile, default is default_main_program
@@ -194,14 +187,14 @@ def transpile(self,
             program = default_main_program()
         self.origin_program = program
         self.trainer_num = trainers
-        self.optimize_ops = optimize_ops
         self.sync_mode = sync_mode
         # TODO(typhoonzero): currently trainer_id is fetched from cluster system
         # like Kubernetes, we should port this to use etcd later when developing
         # fluid distributed training with fault-tolerance.
         self.trainer_id = trainer_id
         pserver_endpoints = pservers.split(",")
         self.pserver_endpoints = pserver_endpoints
+        self.optimize_ops, params_grads = self._get_optimize_pass()
 
         # process lookup_table_op
         # 1. check all lookup_table_op is distributed
@@ -408,11 +401,8 @@ def get_pserver_program(self, endpoint):
         # HACK: optimization global ops only used to scale beta1 and beta2
         # replace it with dependency engine.
         for op in self.optimize_ops:
-            if op.type == "scale":
-                for in_name in op.input_arg_names:
-                    if in_name.startswith("beta1_pow_acc") or \
-                            in_name.startswith("beta2_pow_acc"):
-                        global_ops.append(op)
+            if self._is_adam_connected_op(op):
+                global_ops.append(op)
 
         def __append_optimize_op__(op, block, grad_to_block_id):
             if self._is_opt_op(op):
@@ -1147,3 +1137,32 @@ def _get_lr_ops(self):
                     # we only need to append op for once
                     break
         return lr_ops
+
+    def _get_optimize_pass(self):
+        block = self.origin_program.global_block()
+        opt_ops = []
+        params_grads = []
+        for op in block.ops:
+            if self._is_opt_op(op):
+                opt_ops.append(op)
+                params_grads.append((self.origin_program.global_block().var(
+                    op.input("Param")[0]),
+                                     self.origin_program.global_block().var(
+                                         op.input("Grad")[0])))
+            elif self._is_adam_connected_op(op):
+                opt_ops.append(op)
+            else:
+                pass
+        return opt_ops, params_grads
+
+    def _is_adam_connected_op(self, op):
+        """
+        A hack function to determinate whether the input operator
+        is connected to optimize operator.
+        """
+        if op.type == "scale":
+            for in_name in op.input_arg_names:
+                if in_name.startswith("beta1_pow_acc") or \
+                        in_name.startswith("beta2_pow_acc"):
+                    return True
+        return False

python/paddle/fluid/inferencer.py

@@ -12,18 +12,22 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import core
+
 __all__ = ['Inferencer', ]
 
 
 class Inferencer(object):
-    def __init__(self, network_func, params, place=None):
+    def __init__(self, network_func, param_path=None, place=None):
         # 1. we need to generate a framework.Program by calling
         # network_func. Reference: fluid.program_guard in test_word2vec.py
 
         # 2. move the default_main_program to self.program.
 
         # 3. run the default_startup program.
-        self.params = params
+
+        # 4. load params from param_path into scope
+        self.scope = core.Scope()
         self.place = place
 
     def infer(self, inputs):

python/paddle/fluid/params.py

@@ -80,12 +80,7 @@ def train_loop(main_program):
         trainer_id = int(os.getenv("PADDLE_INIT_TRAINER_ID"))
         training_role = os.getenv("TRAINING_ROLE", "TRAINER")
         t = fluid.DistributeTranspiler()
-        t.transpile(
-            optimize_ops,
-            params_grads,
-            trainer_id,
-            pservers=pserver_endpoints,
-            trainers=trainers)
+        t.transpile(trainer_id, pservers=pserver_endpoints, trainers=trainers)
         if training_role == "PSERVER":
             pserver_prog = t.get_pserver_program(current_endpoint)
             pserver_startup = t.get_startup_program(current_endpoint,

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

@@ -189,12 +189,7 @@ def train_loop(main_program):
         trainer_id = int(os.getenv("PADDLE_INIT_TRAINER_ID"))
         training_role = os.getenv("TRAINING_ROLE", "TRAINER")
         t = fluid.DistributeTranspiler()
-        t.transpile(
-            optimize_ops,
-            params_grads,
-            trainer_id,
-            pservers=pserver_endpoints,
-            trainers=trainers)
+        t.transpile(trainer_id, pservers=pserver_endpoints, trainers=trainers)
         if training_role == "PSERVER":
             pserver_prog = t.get_pserver_program(current_endpoint)
             pserver_startup = t.get_startup_program(current_endpoint,

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

@@ -259,12 +259,7 @@ def train_loop(main_program):
         trainer_id = int(os.getenv("PADDLE_INIT_TRAINER_ID"))
         training_role = os.getenv("TRAINING_ROLE", "TRAINER")
         t = fluid.DistributeTranspiler()
-        t.transpile(
-            optimize_ops,
-            params_grads,
-            trainer_id,
-            pservers=pserver_endpoints,
-            trainers=trainers)
+        t.transpile(trainer_id, pservers=pserver_endpoints, trainers=trainers)
         if training_role == "PSERVER":
             pserver_prog = t.get_pserver_program(current_endpoint)
             pserver_startup = t.get_startup_program(current_endpoint,

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

@@ -231,12 +231,7 @@ def train_loop(main_program):
         trainer_id = int(os.getenv("PADDLE_INIT_TRAINER_ID"))
         training_role = os.getenv("TRAINING_ROLE", "TRAINER")
         t = fluid.DistributeTranspiler()
-        t.transpile(
-            optimize_ops,
-            params_grads,
-            trainer_id,
-            pservers=pserver_endpoints,
-            trainers=trainers)
+        t.transpile(trainer_id, pservers=pserver_endpoints, trainers=trainers)
         if training_role == "PSERVER":
             pserver_prog = t.get_pserver_program(current_endpoint)
             pserver_startup = t.get_startup_program(current_endpoint,