Merge branch 'guru4elephnat_for_pslib' into develop

python/paddle/fluid/async_executor.py

@@ -150,8 +150,13 @@ class AsyncExecutor(object):
                                      data_feed.desc(), filelist, thread_num,
                                      fetch_var_names, debug)
 
-    def config_ps(self, dist_desc, host_sign_list, node_num, index):
-        self.executor.config_pslib(dist_desc, host_sign_list, node_num, index)
+    def config_distributed_nodes(self, dist_opt):
+        # get total rank
+        # get rank index
+        # get iplists
+        # get hadoop info
+        return
+
 
     def start_server(self):
         self.executor.start_server()

python/paddle/fluid/distribute_lookup_table.py

@@ -15,6 +15,26 @@
 LOOKUP_TABLE_TYPE = "lookup_table"
 
 
+def find_distributed_lookup_table_inputs(program, table_name):
+    local_vars = program.current_block().vars
+    inputs = []
+    for op in program.global_block().ops:
+        if op.type == LOOKUP_TABLE_TYPE:
+            if table_name == op.input("W")[0]:
+                inputs.extend(
+                    [local_vars[name] for name in op.input("Ids")])
+    return inputs
+
+def find_distributed_lookup_table_outputs(program, table_name):
+    local_vars = program.current_block().vars
+    outputs = []
+    for op in program.global_block().ops:
+        if op.type == LOOKUP_TABLE_TYPE:
+            if table_name == op.input("W")[0]:
+                outputs.extend(
+                    [local_vars[name] for name in op.output("Out")])
+    return outputs
+
 def find_distributed_lookup_table(program):
     """
     Find distribute lookup table in program.

python/paddle/fluid/distributed/downpour.py

@@ -3,30 +3,62 @@ from .node import DownpourWorker
 from ..backward import append_backward
 import ps_pb2 as pslib
 from paddle.fluid.distribute_lookup_table import find_distributed_lookup_table
+from paddle.fluid.distribute_lookup_table import find_distributed_lookup_table_inputs
+from paddle.fluid.distribute_lookup_table import find_distributed_lookup_table_outputs
 from google.protobuf import text_format
 
 class DownpourSGD(object):
+    """
+    Distributed optimizer of downpour stochastic gradient descent
+    Standard implementation of Google's Downpour SGD
+    in Large Scale Distributed Deep Networks
+
+    Args:
+        learning_rate (float): the learning rate used to update parameters. \
+        Can be a float value
+    Examples:
+        .. code-block:: python
+    
+             downpour_sgd = fluid.distributed.DownpourSGD(learning_rate=0.2)
+             downpour_sgd.minimize(cost)
+    """
     def __init__(self, learning_rate=0.001, window=1):
-        # todo(guru4elephant): if optimizer is not None, will warning here
+        # todo(guru4elephant): add more optimizers here as argument
+        # todo(guru4elephant): make learning_rate as a variable
         self.learning_rate_ = learning_rate
         self.window_ = window
+        self.type = "downpour"
     
     def minimize(self, loss, startup_program=None,
-                 parameter_list=None, no_grad_set=None,
-                 prefetch_slots=None, prefetch_slots_emb=None):
-        params_grads = sorted(append_backward(loss), key=lambda x:x[0].name)
+                 parameter_list=None, no_grad_set=None):
+        params_grads = sorted(append_backward(
+            loss, parameter_list, no_grad_set), key=lambda x:x[0].name)
         table_name = find_distributed_lookup_table(loss.block.program)
+        prefetch_slots = find_distributed_lookup_table_inputs(
+            loss.block.program, table_name)
+        prefetch_slots_emb = find_distributed_lookup_table_outputs(
+            loss.block.program, table_name)
         server = DownpourServer()
+        # window is communication strategy
         worker = DownpourWorker(self.window_)
-        server.add_sparse_table(0, learning_rate,
+        # Todo(guru4elephant): support multiple tables definitions
+        # currently support one big sparse table
+        sparse_table_index = 0
+        # currently merge all dense parameters into one dense table
+        dense_table_index = 1
+        server.add_sparse_table(sparse_table_index, self.learning_rate_,
                                 prefetch_slots, prefetch_slots_emb)
-        server.add_dense_table(1, learning_rate, params, grads)
-        worker.add_sparse_table(0, learning_rate,
+        server.add_dense_table(dense_table_index, self.learning_rate_, 
+                               params_grads[0], params_grads[1])
+        worker.add_sparse_table(sparse_table_index, self.learning_rate_,
                                 prefetch_slots, prefetch_slots_emb)
-        worker.add_dense_table(1, learning_rate, params, grads)
+        worker.add_dense_table(dense_table_index, self.learning_rate_, 
+                               params_grads[0], params_grads[1])
         ps_param = pslib.PSParameter()
         ps_param.server_param.CopyFrom(server.get_desc())
-        #ps_param.worker_param.CopyFrom(worker.get_desc())
+        ps_param.worker_param.CopyFrom(worker.get_desc())
+        # Todo(guru4elephant): figure out how to support more sparse parameters
+        # currently only support lookup_table
         worker_skipped_ops = ["lookup_table", "lookup_table_grad"]
         ps_param_str = text_format.MessageToString(ps_param)
         return [ps_param_str, worker_skipped_ops]

python/paddle/fluid/distributed/helper.py

 from mpi4py import MPI
 
+class FileSystem(object):
+    def __init__(self, fs_type="afs",
+                 uri="afs://tianqi.afs.baidu.com:9902",
+                 user=None,
+                 passwd=None,
+                 hadoop_bin="",
+                 afs_conf=None):
+        assert user not None
+        assert passwd not None
+        assert hadoop_bin not None
+        fs_client = pslib.FsClientParameter()
+        if fs_type == "afs":
+            fs_client.fs_type = pslib.FsApiType.AFS
+        else:
+            fs_client.fs_type = pslib.FsApiType.HDFS
+        fs_client.uri = uri
+        fs_client.user = user
+        fs_client.passwd = passwd
+        fs_client.buffer_size = 0
+        fs_client.afs_conf = afs_conf if not afs_conf else ""
+
+
 class MPIHelper(object):
     def __init__(self):
         self.comm = MPI.COMM_WORLD
@@ -18,3 +40,5 @@ class MPIHelper(object):
     def get_hostname(self):
         import socket
         return socket.gethostname()
+    
+        

python/paddle/fluid/distributed/node.py

@@ -12,29 +12,29 @@ class Worker(object):
 
 class DownpourServer(Server):
     def __init__(self):
-        #self.server_ = pslib.ServerParameter().downpour_server_param
         self.server_ = pslib.ServerParameter()
 
     def add_sparse_table(self, table_id, learning_rate,
-                         slot_key, slot_value_var, slot_grad_var):
-        #table = self.server_.downpour_table_param.add()
+                         slot_key_vars, slot_value_var):
         table = self.server_.downpour_server_param.downpour_table_param.add()
         table.table_id = table_id
-        table.type = PS_SPARSE_TABLE
+        table.type = pslib.PS_SPARSE_TABLE
         table.accessor.accessor_class = "DownpourFeatureValueAccessor"
         table.accessor.dense_sgd_param.adam.learning_rate = learning_rate
-        table.accessor.fea_dim = slot_value_var[0].shape[1]
+        table.accessor.fea_dim = abs(reduce(lambda x, y: x * y, 
+                                            slot_value_var[0].shape, 1))
 
     def add_dense_table(self, table_id, learning_rate, 
                         param_var, grad_var):
-        #table = self.server_.downpour_table_param.add()
         table = self.server_.downpour_server_param.downpour_table_param.add()
         table.table_id = table_id
-        table.type = PS_DENSE_TABLE
+        table.type = pslib.PS_DENSE_TABLE
         table.accessor.accessor_class = "DownpourDenseValueAccessor"
         table.accessor.sparse_sgd_param.learning_rate = learning_rate
-        table.accessor.fea_dim = 1
-        #table.accessor.fea_dim = reduce(lambda x, y: x.shape, 1 for x in param_var)
+        fea_dim = 0
+        for param in param_var:
+            fea_dim += reduce(lambda x, y: x * y, param.shape, 1)
+        table.accessor.fea_dim = fea_dim
 
     def get_desc(self):
         return self.server_
@@ -43,28 +43,24 @@ class DownpourServer(Server):
 class DownpourWorker(Worker):
     def __init__(self, window):
         self.window = window
-        #self.worker_ = pslib.WorkerParameter().downpour_worker_param
-        #self.worker_ = pslib.WorkerParameter()
         self.worker_ = pslib.DownpourTrainerParameter()
-        #self.worker_.pull_dense_per_batch = window
-        #self.worker_.push_dense_per_batch = window
-        #self.worker_.downpour_worker_param.pull_dense_per_batch = window
-        #self.worker_.downpour_worker_param.push_dense_per_batch = window
         self.worker_.pull_dense_per_batch = window
         self.worker_.push_dense_per_batch = window
-        print(self.worker_)
 
-    def add_sparse_table(self, table_id, 
-                         slot_keys, slot_value_vars, slot_grad_vars):
-        #table = self.worker_.sparse_table.add()
-        table = self.worker_.downpour_worker_param.sparse_table.add()
+    def add_sparse_table(self, table_id, learning_rate,
+                         slot_key_vars, slot_value_vars):
+        table = self.worker_.sparse_table.add()
         table.table_id = table_id
-        table.slot.extend(slot_keys)
-        self.worker_.extend([grad.name for grad in slot_grad_vars])
+        table.slot_key.extend(
+            [var.name for var in slot_key_vars])
+        table.slot_value.extend(
+            [var.name for var in slot_value_vars])
+        table.slot_gradient.extend(
+            [var.name + "@GRAD" for var in slot_value_vars])
 
-    def add_dense_table(self, table_id, param_vars, grad_vars):
-        #table = self.worker_.dense_table.add()
-        table = self.worker_.downpour_worker_param.dense_table.add()
+    def add_dense_table(self, table_id, learning_rate, 
+                        param_vars, grad_vars):
+        table = self.worker_.dense_table.add()
         table.table_id = table_id
         table.dense_variable_name.extend([p.name for p in param_vars])
         table.dense_gradient_variable_name.extend([g.name for g in grad_vars])