Quickstart (#544)

* upload enhowto

* upload cluster quick start en

* review commit

* recover original multi_node.rst index

* fix index of cluster_howto

* fix index

* fix bugs

doc/fluid/user_guides/howto/training/cluster_howto_en.rst

+.. _cluster_howto_en:
+
+Manual for Distributed Training with Fluid
+==========================================
+
+Basic Idea Of Distributed Training
+-------------------------------------
+
+Distributed deep learning training is usually divided into two parallelization methods: data parallelism, model parallelism. Refer to the following figure:
+
+.. image:: src/parallelism.png
+
+In the model parallelism mode, the layers and parameters of the model will be distributed on multiple nodes. The model will go through multiple communications across nodes in the feeding forward and back propagation training of a mini-batch. Each node only saves a part of the entire model; 
+
+In data parallelism mode, each node holds the complete layers and parameters of the model, each node performs feeding forward and back propagation calculations on its own, and then conducts the aggregation of the gradients and updates the parameters on all nodes synchronously. 
+
+Current version of Fluid only provides data parallelism mode. In addition, implementations of special cases in model parallelism mode (e.g. large sparse model training ) will be explained in subsequent documents.
+
+In the training of data parallelism mode, Fluid uses two communication modes to deal with the requirements of distributed training for different training tasks, namely RPC Communication and Collective Communication. The RPC communication method uses `gRPC <https://github.com/grpc/grpc/>`_ , Collective communication method uses `NCCL2 <https://developer.nvidia.com/nccl>`_ . 
+
+.. csv-table:: The table above is a horizontal comparison of RPC communication and Collective communication
+	:header: "Feature", "Collective", "RPC"
+
+	"Ring-Based Communication", "Yes", "No"
+	"Asynchronous Training", "Yes", "Yes"
+	"Distributed Model", "No", "Yes"
+	"Fault-tolerant Training", "No", "Yes"
+	"Performance", "Faster", "Fast"
+
+- Structure of RPC Communication Method:
+
+  .. image:: src/dist_train_pserver.png
+
+  Data-parallelised distributed training in RPC communication mode will start multiple pserver processes and multiple trainer processes, each pserver process will save a part of the model parameters and be responsible for receiving the gradients sent from the trainers and updating these model parameters; Each trainer process will save a copy of the complete model, and use a part of the data to train, then send the gradients to the pservers, finally pull the updated parameters from the pserver.
+
+  The pserver process can be on a compute node that is completely different from the trainer, or it can share the same node with a trainer. The number of pserver processes required for a distributed task usually needs to be adjusted according to the actual situation to achieve the best performance. However, usually pserver processes are no more than trainer processes.
+
+  When using GPU training, the pserver can choose to use the GPU or only use the CPU. If the pserver also uses the GPU, it will result in the extra overhead of copying the gradient data received from the CPU to the GPU. In some cases, the overall training performance will be degraded.
+
+- Structure of NCCL2 communication method:
+
+  .. image:: src/dist_train_nccl2.png
+
+NCCL2 (Collective communication method) for distributed training avoids the need of pserver processes. Each trainer process holds a complete set of model parameters. After the calculation of the gradient, the trainer, through mutual communications, "Reduce" the gradient data to all devices of all nodes and then each node completes parameter updates of its own.
+
+Training in the Parameter Server Manner 
+----------------------------------------------
+
+Use the :code:`transpiler` API to quickly convert a program that can be executed on a single machine into a program that can be executed in a distributed manner. On different server nodes, pass values to corresponding arguments at :code:`transpiler` to get the :code:`Program` which current node is to execute:
+
+
+.. csv-table:: required configuration parameters
+   :header: "parameter", "description"
+
+   "role", "\ **required**\  distinguishes whether to start as pserver or trainer, this arugument is not passed into ``transpile`` , you can also use other variable names or environment variables"
+   "trainer_id", "\ **required**\  If it is a trainer process, it is used to specify the unique id of the current trainer in the task, starting from 0, and must be guaranteed not to be repeated in one task"
+   "pservers", "\ **required**\ ip:port list string of all pservers in current task, for example: 127.0.0.1:6170,127.0.0.1:6171"
+   "trainers", "\ **required**\  the number of trainer nodes"
+   "sync_mode", "\ **optional**\  True for synchronous mode, False for asynchronous mode"
+   "startup_program", "\ **optional**\  If startup_program is not the default fluid.default_startup_program(), this parameter needs to be passed in"
+   "current_endpoint", "\ **optional**\  This parameter is only required for NCCL2 mode"
+
+For example, suppose there are two nodes, namely :code:`192.168.1.1` and :code:`192.168.1.2`, use port 6170 to start 4 trainers.
+Then the code can be written as:
+
+.. code-block:: python
+
+	role = "PSERVER"
+	trainer_id = 0 # get actual trainer id from cluster
+	pserver_endpoints = "192.168.1.1:6170,192.168.1.2:6170"
+	current_endpoint = "192.168.1.1:6170" # get actual current endpoint
+	trainers = 4
+	t = fluid.DistributeTranspiler()
+	t.transpile(trainer_id, pservers=pserver_endpoints, trainers=trainers)
+	if role == "PSERVER":
+		    pserver_prog = t.get_pserver_program(current_endpoint)
+		    pserver_startup = t.get_startup_program(current_endpoint,Pserver_prog)
+		    exe.run(pserver_startup)
+		    exe.run(pserver_prog)
+	elif role == "TRAINER":
+		train_loop(t.get_trainer_program())
+
+
+Choose Synchronous Or Asynchronous Training
++++++++++++++++++++++++++++++++++++++++++++++
+
+Fluid distributed tasks support synchronous training or asynchronous training. 
+
+In the synchronous training mode, all trainer nodes will merge the gradient data of all nodes synchronously per mini-batch and send them to the parameter server to complete the update. 
+
+In the asynchronous mode, each trainer does not wait for each other, and independently update the parameters on the parameter server. 
+
+In general, using the asynchronous training method can have a higher overall throughput than the synchronous training mode when there are more trainer nodes.
+
+When the :code:`transpile` function is called, the distributed training program is generated by default. The asynchronous training program can be generated by specifying the :code:`sync_mode=False` parameter:
+
+.. code-block:: python
+
+	t.transpile(trainer_id, pservers=pserver_endpoints, trainers=trainers, sync_mode=False)
+
+
+
+Whether To Use The Distributed Embedding Table For Training
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+Embedding is widely used in various network structures, especially text processing related models.
+In some scenarios, such as recommendation systems or search engines, the number of feature ids of embedding may be very large. When it reaches a certain number, the embedding parameter will become very large.
+On the one hand, the memory of the single machine may not be competent, resulting in the inability to train.
+On the other hand, the normal training mode needs to synchronize the complete set of parameters for each iteration. If the parameter is too large, the communication will become very slow, which will affect the training speed.
+
+Fluid supports the training of very large scale sparse features embedding at hundred billion level. The embedding parameter is only saved on the parameter server. The parameter prefetch and gradient sparse update method greatly reduce the traffic and improve the communication speed.
+
+This feature is only valid for distributed training and cannot be used on a single machine. Need to be used with sparse updates.
+
+Usage: When configuring embedding, add the parameters :code:`is_distributed=True` and :code:`is_sparse=True`.
+Parameters :code:`dict_size` Defines the total number of ids in the data. The id can be any value in the int64 range. As long as the total number of ids is less than or equal to dict_size, it can be supported.
+So before you configure, you need to estimate the total number of feature ids in the data.
+
+.. code-block:: python
+
+	emb = fluid.layers.embedding(
+		is_distributed=True,
+		input=input,
+		size=[dict_size, embedding_width],
+		is_sparse=True)
+
+
+Select Parameter Distribution Method
+++++++++++++++++++++++++++++++++++++++
+
+Parameter :code:`split_method` can specify how the parameters are distributed on the parameter servers.
+
+Fluid uses `RoundRobin <https://en.wikipedia.org/wiki/Round-robin_scheduling>`_ by default to scatter parameters to multiple parameter servers. 
+In this case, the parameters are evenly distributed on all parameter servers in the case where the parameter segmentation is not turned off by default. 
+If you need to use something else, you can pass in other methods. The currently available methods are: :code:`RoundRobin` and :code:`HashName` . You can also use a customized distribution method, just refer to `here <https://github.com/PaddlePaddle/Paddle/blob/develop/python/paddle/fluid/transpiler/ps_dispatcher.py#L44>`_
+to write customized distribution function
+
+
+Turn Off the slice-up of Parameters 
+++++++++++++++++++++++++++++++++++++++
+
+Parameter :code:`slice_var_up` specifies whether to split large (more than 8192 elements) parameters into multiple parameter servers to balance the computational load. The default is on.
+
+When the sizes of the trainable parameters in the model are relatively uniform or a customized parameter distribution method is used, which evenly distributes the parameters on multiple parameter servers, you can choose to turn off the slice-up function, which reduces the computational and copying overhead of slicing and reorganization:
+
+.. code-block:: python
+
+	t.transpile(trainer_id, pservers=pserver_endpoints, trainers=trainers, slice_var_up=False)
+
+
+Turn On Memory Optimization
+++++++++++++++++++++++++++++++
+
+In the parameter server distributed training mode, to enable memory optimization :code:`memory_optimize` , compared with a single machine, you need to pay attention to the following rules:
+
+- On the pserver side, **don't** execute :code:`memory_optimize`
+- On the trainer side, execute :code:`fluid.memory_optimize` and then execute :code:`t.transpile()`
+- On the trainer side, calling :code:`memory_optimize` needs to add :code:`skip_grads=True` to ensure the gradient sent is not renamed : :code:`fluid.memory_optimize(input_program, skip_grads=True)`
+
+Example:
+
+.. code-block:: python
+
+	if role == "TRAINER":
+		fluid.memory_optimize(fluid.default_main_program(), skip_grads=True)
+	t = fluid.DistributeTranspiler()
+	t.transpile(trainer_id, pservers=pserver_endpoints, trainers=trainers)
+	if role == "PSERVER":
+		# start pserver here
+	elif role == "TRAINER":
+		# start trainer here
+
+
+Training Using NCCL2 Communication
+--------------------
+
+Distributed training in NCCL2 mode, because there is no parameter server role, the trainers directly communicate with each other. Pay attention to the following tips:
+
+* Configure :code:`mode="nccl2"` in :code:`fluid.DistributeTranspilerConfig` .
+* When calling :code:`transpile`, :code:`trainers` is fed with the endpoints of all trainer nodes, and passed with the argument :code:`current_endpoint` .
+* Initialize :code:`ParallelExecutor` with :code:`num_trainers` and :code:`trainer_id` .
+
+For example:
+
+.. code-block:: python
+
+	trainer_id = 0 # get actual trainer id here
+	trainers = "192.168.1.1:6170,192.168.1.2:6170"
+	current_endpoint = "192.168.1.1:6170"
+	config = fluid.DistributeTranspilerConfig()
+	config.mode = "nccl2"
+	t = fluid.DistributeTranspiler(config=config)
+	t.transpile(trainer_id, trainers=trainers, current_endpoint=current_endpoint)
+	txe = fluid.ParallelExecutor(use_cuda,
+		loss_name=loss_name, num_trainers=len(trainers.split(",")), trainer_id=trainer_id)
+	...
+
+.. csv-table:: Description of the necessary parameters for NCCL2 mode
+	:header: "parameter", "description"
+
+	"trainer_id", "The unique ID of each trainer node in the task, starting at 0, there cannot be any duplication"
+	"trainers", "endpoints of all trainer nodes in the task, used to broadcast NCCL IDs when NCCL2 is initialized"
+	"current_endpoint", "endpoint of current node"
+
+Currently, distributed training using NCCL2 only supports synchronous training. The distributed training using NCCL2 mode is more suitable for the model which is relatively large and needs \
+synchronous training and GPU training. If the hardware device supports RDMA and GPU Direct, this can achieve high distributed training performance.
+
+Important Notes on NCCL2 Distributed Training
+++++++++++++++++++++++++++++++++++++++++++++++
+
+**Note** : Please ensure each node has the same amount of data to train in NCCL2 mode distributed training, which prevents
+exit at the final iteration. There are two common ways:
+
+- Randomly sample some data to complement nodes where less data are distributed. (We recommend this method for sake of a complete dataset to be trained)
+- Each node only trains fixed number of batches per pass, which is controlled by python codes. If a node has more data than this fixed amount, then these 
+  marginal data will not be trained.
+
+**Note** :  If there are multiple network devices in the system, you need to manually specify the devices used by NCCL2.
+
+Assuming you need to use :code:`eth2` as the communication device, you need to set the following environment variables:
+
+.. code-block:: bash
+
+    export NCCL_SOCKET_IFNAME=eth2
+
+In addition, NCCL2 provides other switch environment variables, such as whether to enable GPU Direct, whether to use RDMA, etc. For details, please refer to
+`ncclknobs <https://docs.nvidia.com/deeplearning/sdk/nccl-developer-guide/index.html#ncclknobs>`_ .

doc/fluid/user_guides/howto/training/cluster_quick_start_en.rst

+..  _cluster_quick_start_en:
+
+Quick Start with Distributed Training
+==========================
+
+Preparation
+--------------------
+In this article, we'll show you how to quickly start a PaddlePaddle distributed training task in a cluster. Before you start, do some preparatory work as follows:
+
+1. Prepare a connected training cluster. Here we use 4 training nodes with format ``*.paddlepaddle.com`` to represent the host name of the node. You can modify it according to the actual situation.
+
+2. Make sure you have read :ref:`install_steps` before you start and can run PaddlePaddle on all nodes of the cluster.
+
+Example code
+-------------
+
+Let's use a very simple linear regression model as an example to explain how to start a distributed training task with 2 pserver server nodes and 2 trainer nodes. You can save this code as ``dist_train.py`` .
+
+.. code:: python
+
+
+    import os
+    import paddle
+    import paddle.fluid as fluid
+
+    # train reader
+    BATCH_SIZE = 20
+    EPOCH_NUM = 30
+    BATCH_SIZE = 8
+
+    train_reader = paddle.batch(
+        paddle.reader.shuffle(
+            paddle.dataset.uci_housing.train(), buf_size=500),
+        batch_size=BATCH_SIZE)
+
+    def train():
+        y = fluid.layers.data(name='y', shape=[1], dtype='float32')
+        x = fluid.layers.data(name='x', shape=[13], dtype='float32')
+        y_predict = fluid.layers.fc(input=x, size=1, act=None)
+
+        loss = fluid.layers.square_error_cost(input=y_predict, label=y)
+        avg_loss = fluid.layers.mean(loss)
+        opt = fluid.optimizer.SGD(learning_rate=0.001)
+        opt.minimize(avg_loss)
+
+        place = fluid.CPUPlace()
+        feeder = fluid.DataFeeder(place=place, feed_list=[x, y])
+        exe = fluid.Executor(place)
+
+        # fetch distributed training environment setting
+        training_role = os.getenv("PADDLE_TRAINING_ROLE", None)
+        port = os.getenv("PADDLE_PSERVER_PORT", "6174")
+        pserver_ips = os.getenv("PADDLE_PSERVER_IPS", "")
+        trainer_id = int(os.getenv("PADDLE_TRAINER_ID", "0"))
+        eplist = []
+        for ip in pserver_ips.split(","):
+            eplist.append(':'.join([ip, port]))
+        pserver_endpoints = ",".join(eplist)
+        trainers = int(os.getenv("PADDLE_TRAINERS"))
+        current_endpoint = os.getenv("PADDLE_CURRENT_IP", "") + ":" + port
+
+        t = fluid.DistributeTranspiler()
+        t.transpile(
+            trainer_id = trainer_id,
+            pservers = pserver_endpoints,
+            trainers = trainers)
+
+        if training_role == "PSERVER":
+            pserver_prog = t.get_pserver_program(current_endpoint)
+            startup_prog = t.get_startup_program(current_endpoint, pserver_prog)
+            exe.run(startup_prog)
+            exe.run(pserver_prog)
+        elif training_role == "TRAINER":
+            trainer_prog = t.get_trainer_program()
+            exe.run(fluid.default_startup_program())
+
+            for epoch in range(EPOCH_NUM):
+                for batch_id, batch_data in enumerate(train_reader()):
+                    avg_loss_value, = exe.run(trainer_prog,
+                                          feed=feeder.feed(batch_data),
+                                          fetch_list=[avg_loss])
+                    if (batch_id + 1) % 10 == 0:
+                        print("Epoch: {0}, Batch: {1}, loss: {2}".format(
+                            epoch, batch_id, avg_loss_value[0]))
+            # destory the resource of current trainer node in pserver server node
+            exe.close()
+        else:
+            raise AssertionError("PADDLE_TRAINING_ROLE should be one of [TRAINER, PSERVER]")
+
+    train()
+
+
+Environment Variables
+------------------------------------
+
+When starting a distributed training task, different environment variables are used to represent different node roles, details as follows:
+
+.. list-table::
+  :header-rows: 1
+
+  * - Environment Variable
+    - Data Type 
+    - Example 
+    - Description
+  * - :code:`PADDLE_TRAINING_ROLE`
+    - str 
+    - :code:`PSERVER,TRANERR`
+    - role of current training node
+  * - :code:`PADDLE_PSERVER_IPS`
+    - str 
+    - :code:`ps0.paddlepaddle.com, ps1.paddlepaddle.com`
+    - The IP addresses or hostnames of all pserver nodes in the distributed training task, separated by ","
+  * - :code:`PADDLE_PSERVER_PORT`
+    - int 
+    - 6174 
+    - port that the pserver process listens to
+  * - :code:`PADDLE_TRAINERS`
+    - int
+    - 2 
+    - Number of trainer nodes in a distributed training task
+  * - :code:`PADDLE_CURRENT_IP`
+    - str 
+    - :code:`ps0.paddlepaddle.com`
+    - IP address or hostname of the current pserver node
+  * - :code:`PADDLE_TRAINER_ID`
+    - str 
+    - 0 
+    - ID of the current trainer node (unique), in the range of [0, PADDLE_TRAINERS)
+
+**Note:** Environment variables are just a way to get runtime information. In practical tasks, you can use command line parameters to obtain runtime information.
+
+API related to Distributed Training
+---------------------------------
+
+DistributeTranspiler
+~~~~~~~~~~~~~~~~~~~~~~
+
+The machines in distributed training tasks based on the pserver-trainer architecture are divided into two roles: Parameter Server (pserver) and trainer. In Fluid, users only need to configure the network configuration required for single node training. The ``DistributeTranspiler`` module automatically modifies the single-node network settings into settings on which pserver and trainer needs to run based on the role of current training node:
+
+.. code:: python
+
+  t = fluid.DistributeTranspiler()
+  t.transpile(
+    trainer_id = trainer_id,
+    pservers = pserver_endpoints,
+    trainers = trainers)
+  if PADDLE_TRAINING_ROLE == "TRAINER":
+    # fetch the trainer program and execute it
+    trainer_prog = t.get_trainer_program()
+    ...
+
+  elif PADDLE_TRAINER_ROLE == "PSERVER":
+    # fetch the pserver program and execute it
+    pserver_prog = t.get_pserver_program(current_endpoint)
+    ...
+
+
+Exe.close()
+~~~~~~~~~~~~~~
+
+
+The status information of all trainer nodes is saved in the pserver node. When trainer finishes training, ``exe.close()`` should be called to notify all PServer nodes to release the resources of the current Trainer nodes:
+
+.. code:: python
+
+  exe = fluid.Executor(fluid.CPUPlace())
+  # training process ...
+  exe.close() # notify PServer to destory the resource
+
+
+Start a Distributed Training Task
+----------------------------------
+
+.. list-table::
+   :header-rows: 1
+
+
+   * - Start Node 
+     - Start Command 
+     - Description
+   * - ps0.paddlepaddle.com 
+     - :code:`PADDLE_TRAINING_ROLE=PSERVER PADDLE_CURRENT_IP=ps0.paddlepaddle.com PADDLE_PSERVER_IPS=ps0.paddlepaddle.com, ps1.paddlepaddle.com PADDLE_TRAINERS=2 PADDLE_PSERVER_PORT=6174 python fluid_dist.py`
+     - Start pserver node
+   * - ps1.paddlepaddle.com
+     - :code:`PADDLE_TRAINING_ROLE=PSERVER PADDLE_CURRENT_IP=ps1.paddlepaddle.com PADDLE_PSERVER_IPS=ps0.paddlepaddle.com, ps1.paddlepaddle.com PADDLE_TRAINERS=2 PADDLE_PSERVER_PORT=6174 python fluid_dist.py`
+     - Start pserver node
+   * - trainer0.paddlepaddle.com       
+     - :code:`PADDLE_TRAINING_ROLE=TRAINER PADDLE_PSERVER_IPS=ps0.paddlepaddle.com, ps1.paddlepaddle.com PADDLE_TRAINERS=2 PADDLE_TRAINER_ID=0 PADDLE_PSERVER_PORT=6174 python fluid_dist.py`
+     - Start the number 0 Trainer Node 
+   * - trainer1.paddlepaddle.com       
+     - :code:`PADDLE_TRAINING_ROLE=TRAINER PADDLE_PSERVER_IPS=ps0.paddlepaddle.com, ps1.paddlepaddle.com PADDLE_TRAINERS=2 PADDLE_TRAINER_ID=1 PADDLE_PSERVER_PORT=6174 python fluid_dist.py`
+     - Start the number 1 trainer node