提交 fe48dfcb 编写于 作者: J JiayiFeng

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

...@@ -102,6 +102,8 @@ def split_dense_variable(var_list, ...@@ -102,6 +102,8 @@ def split_dense_variable(var_list,
the parameter server side can gain better performance. By default the parameter server side can gain better performance. By default
minimum block size is 1024. The max block size is used to prevent minimum block size is 1024. The max block size is used to prevent
very large blocks that may cause send error. very large blocks that may cause send error.
:return: A list of VarBlocks. Each VarBlock specifies a shard of
the var.
""" """
blocks = [] blocks = []
for var in var_list: for var in var_list:
...@@ -192,22 +194,24 @@ class DistributeTranspiler: ...@@ -192,22 +194,24 @@ class DistributeTranspiler:
self.trainer_id = trainer_id self.trainer_id = trainer_id
pserver_endpoints = pservers.split(",") pserver_endpoints = pservers.split(",")
# step1 # step1: For large parameters and gradients, split them into smaller
# blocks.
param_list = [pg[0] for pg in params_grads] param_list = [pg[0] for pg in params_grads]
grad_list = [pg[1] for pg in params_grads] grad_list = [pg[1] for pg in params_grads]
grad_blocks = split_dense_variable(grad_list, len(pserver_endpoints)) grad_blocks = split_dense_variable(grad_list, len(pserver_endpoints))
param_blocks = split_dense_variable(param_list, len(pserver_endpoints)) param_blocks = split_dense_variable(param_list, len(pserver_endpoints))
# step2 # step2: Create new vars for the parameters and gradients blocks and
# add ops to do the split.
grad_var_mapping = self._append_split_op(program, grad_blocks) grad_var_mapping = self._append_split_op(program, grad_blocks)
# step3 param_var_mapping = self._create_vars_from_blocklist(program,
param_blocks)
# step3: Add gradients as send op inputs and parameters as send
# op outputs.
send_inputs = [] send_inputs = []
send_outputs = [] send_outputs = []
for b in grad_blocks: # append by order for b in grad_blocks: # append by order
varname, block_id, _ = b.split(":") varname, block_id, _ = b.split(":")
send_inputs.append(grad_var_mapping[varname][int(block_id)]) send_inputs.append(grad_var_mapping[varname][int(block_id)])
param_var_mapping = self._create_vars_from_blocklist(program,
param_blocks)
for b in param_blocks: for b in param_blocks:
varname, block_id, _ = b.split(":") varname, block_id, _ = b.split(":")
send_outputs.append(param_var_mapping[varname][int(block_id)]) send_outputs.append(param_var_mapping[varname][int(block_id)])
...@@ -237,7 +241,7 @@ class DistributeTranspiler: ...@@ -237,7 +241,7 @@ class DistributeTranspiler:
"RPCClient": rpc_client_var}, "RPCClient": rpc_client_var},
attrs={"endpoints": pserver_endpoints, attrs={"endpoints": pserver_endpoints,
"epmap": eplist}) "epmap": eplist})
# step4 # step4: Concat the parameters splits together after recv.
for varname, splited_var in param_var_mapping.iteritems(): for varname, splited_var in param_var_mapping.iteritems():
if len(splited_var) <= 1: if len(splited_var) <= 1:
continue continue
...@@ -258,13 +262,14 @@ class DistributeTranspiler: ...@@ -258,13 +262,14 @@ class DistributeTranspiler:
def get_pserver_program(self, endpoint): def get_pserver_program(self, endpoint):
""" """
Get pserver side program using the endpoint. Get pserver side program using the endpoint.
TODO(panyx0718): Revisit this assumption. what if #blocks > #pservers.
NOTE: assume blocks of the same variable is not distributed NOTE: assume blocks of the same variable is not distributed
on the same pserver, only change param/grad varnames for on the same pserver, only change param/grad varnames for
trainers to fetch. trainers to fetch.
""" """
# step1 # step1
pserver_program = Program() pserver_program = Program()
# step2 # step2: Create vars to receive vars at parameter servers.
recv_inputs = [] recv_inputs = []
for v in self.param_grad_ep_mapping[endpoint]["params"]: for v in self.param_grad_ep_mapping[endpoint]["params"]:
self._clone_var(pserver_program.global_block(), v) self._clone_var(pserver_program.global_block(), v)
...@@ -278,12 +283,6 @@ class DistributeTranspiler: ...@@ -278,12 +283,6 @@ class DistributeTranspiler:
orig_var_name = v.name[:suff_idx] orig_var_name = v.name[:suff_idx]
else: else:
orig_var_name = v.name orig_var_name = v.name
single_trainer_var = pserver_program.global_block().create_var(
name=orig_var_name,
persistable=True,
type=v.type,
dtype=v.dtype,
shape=v.shape)
if self.trainers > 1: if self.trainers > 1:
for trainer_id in xrange(self.trainers): for trainer_id in xrange(self.trainers):
var = pserver_program.global_block().create_var( var = pserver_program.global_block().create_var(
...@@ -294,6 +293,12 @@ class DistributeTranspiler: ...@@ -294,6 +293,12 @@ class DistributeTranspiler:
shape=v.shape) shape=v.shape)
recv_inputs.append(var) recv_inputs.append(var)
else: else:
single_trainer_var = pserver_program.global_block().create_var(
name=orig_var_name,
persistable=True,
type=v.type,
dtype=v.dtype,
shape=v.shape)
recv_inputs.append(single_trainer_var) recv_inputs.append(single_trainer_var)
# step3 # step3
...@@ -344,7 +349,7 @@ class DistributeTranspiler: ...@@ -344,7 +349,7 @@ class DistributeTranspiler:
self._append_pserver_non_opt_ops(block, op) self._append_pserver_non_opt_ops(block, op)
append_block = optimize_block append_block = optimize_block
# append lr decay ops to the child block if exits # append lr decay ops to the child block if exists
lr_ops = self._get_lr_ops() lr_ops = self._get_lr_ops()
if len(lr_ops) > 0: if len(lr_ops) > 0:
for _, op in enumerate(lr_ops): for _, op in enumerate(lr_ops):
...@@ -447,8 +452,10 @@ class DistributeTranspiler: ...@@ -447,8 +452,10 @@ class DistributeTranspiler:
block_list, block_list,
add_trainer_suffix=False): add_trainer_suffix=False):
""" """
Create vars for each split.
NOTE: only grads need to be named for different trainers, use NOTE: only grads need to be named for different trainers, use
add_trainer_suffix to rename the grad vars. add_trainer_suffix to rename the grad vars.
:return: A dict mapping from original var name to each var split.
""" """
block_map = dict() block_map = dict()
var_mapping = dict() var_mapping = dict()
...@@ -615,6 +622,7 @@ class DistributeTranspiler: ...@@ -615,6 +622,7 @@ class DistributeTranspiler:
type="sum", type="sum",
inputs={"X": vars2merge}, inputs={"X": vars2merge},
outputs={"Out": merged_var}) outputs={"Out": merged_var})
# TODO(panyx0718): What if it's SELECTED_ROWS.
if not merged_var.type == core.VarDesc.VarType.SELECTED_ROWS: if not merged_var.type == core.VarDesc.VarType.SELECTED_ROWS:
optimize_block.append_op( optimize_block.append_op(
type="scale", type="scale",
...@@ -638,7 +646,7 @@ class DistributeTranspiler: ...@@ -638,7 +646,7 @@ class DistributeTranspiler:
shape=param_block.shape) shape=param_block.shape)
new_inputs[key] = tmpvar new_inputs[key] = tmpvar
elif key == "LearningRate": elif key == "LearningRate":
# leraning rate variable has already be created by non-optimize op, # learning rate variable has already be created by non-optimize op,
# don't create it once again. # don't create it once again.
lr_varname = opt_op.input(key)[0] lr_varname = opt_op.input(key)[0]
if pserver_block.vars.has_key(lr_varname): if pserver_block.vars.has_key(lr_varname):
...@@ -773,6 +781,7 @@ class DistributeTranspiler: ...@@ -773,6 +781,7 @@ class DistributeTranspiler:
return False return False
def _get_input_map_from_op(self, varmap, op): def _get_input_map_from_op(self, varmap, op):
"""Returns a dict from op input name to the vars in varmap."""
iomap = dict() iomap = dict()
for key in op.input_names: for key in op.input_names:
vars = [] vars = []
...@@ -785,6 +794,7 @@ class DistributeTranspiler: ...@@ -785,6 +794,7 @@ class DistributeTranspiler:
return iomap return iomap
def _get_output_map_from_op(self, varmap, op): def _get_output_map_from_op(self, varmap, op):
"""Returns a dict from op output name to the vars in varmap."""
iomap = dict() iomap = dict()
for key in op.output_names: for key in op.output_names:
vars = [] vars = []
...@@ -812,6 +822,7 @@ class DistributeTranspiler: ...@@ -812,6 +822,7 @@ class DistributeTranspiler:
find_ops.append(op) find_ops.append(op)
# make a union find struct by the ops in default_main_program # make a union find struct by the ops in default_main_program
ufind = UnionFind(block.ops) ufind = UnionFind(block.ops)
for op1 in block.ops: for op1 in block.ops:
for op2 in block.ops: for op2 in block.ops:
# NOTE: we need to skip all optimize ops, since it is connected # NOTE: we need to skip all optimize ops, since it is connected
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册