# Copyright (c) 2020 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. import os import warnings from paddle import base from paddle.base import core from paddle.base.compiler import CompiledProgram from paddle.base.executor import Executor from paddle.base.framework import Program from ..base.private_helper_function import wait_server_ready from .runtime_base import RuntimeBase __all__ = [] def conv_indent(indent): return "".join([" "] * indent) PSERVER_SAVE_SUFFIX = ".shard" def parse_table_class(varname, o_main_program): from paddle.incubate.distributed.fleet.parameter_server.ir.public import ( is_distributed_sparse_op, is_sparse_op, ) for op in o_main_program.global_block().ops: if not is_distributed_sparse_op(op) and not is_sparse_op(op): continue param_name = op.input("W")[0] if ( param_name == varname and op.type == "lookup_table" or op.type == "lookup_table_v2" ): if op.has_attr('table_class') and op.attr("table_class") != "none": return op.attr('table_class') else: return "MemorySparseTable" def get_default_accessor_proto(accessor, varname, o_main_program): embedding_dim = 0 for var in o_main_program.list_vars(): if var.name == varname: embedding_dim = var.shape[1] break if not accessor.HasField("accessor_class"): accessor.accessor_class = "CtrCommonAccessor" if not accessor.HasField("fea_dim"): accessor.fea_dim = embedding_dim if not accessor.HasField("embedx_dim"): accessor.embedx_dim = embedding_dim - 3 if not accessor.HasField("embedx_threshold"): accessor.embedx_threshold = 0 ctr_accessor_param = accessor.ctr_accessor_param if not ctr_accessor_param.HasField("nonclk_coeff"): ctr_accessor_param.nonclk_coeff = 0.1 if not ctr_accessor_param.HasField("click_coeff"): ctr_accessor_param.click_coeff = 1.0 if not ctr_accessor_param.HasField("base_threshold"): ctr_accessor_param.base_threshold = 0 if not ctr_accessor_param.HasField("delta_threshold"): ctr_accessor_param.delta_threshold = 0 if not ctr_accessor_param.HasField("delta_keep_days"): ctr_accessor_param.delta_keep_days = 16 if not ctr_accessor_param.HasField("show_click_decay_rate"): ctr_accessor_param.show_click_decay_rate = 1 if not ctr_accessor_param.HasField("delete_threshold"): ctr_accessor_param.delete_threshold = 0 if not ctr_accessor_param.HasField("delete_after_unseen_days"): ctr_accessor_param.delete_after_unseen_days = 30 if not ctr_accessor_param.HasField("ssd_unseenday_threshold"): ctr_accessor_param.ssd_unseenday_threshold = 1 for sgd_param in [accessor.embed_sgd_param, accessor.embedx_sgd_param]: if not sgd_param.HasField("name"): sgd_param.name = "SparseAdaGradSGDRule" if ( sgd_param.name == "SparseAdaGradSGDRule" or sgd_param.name == "StdAdaGradSGDRule" ): if not sgd_param.adagrad.HasField("learning_rate"): sgd_param.adagrad.learning_rate = 0.05 if not sgd_param.adagrad.HasField("initial_g2sum"): sgd_param.adagrad.initial_g2sum = 3.0 if not sgd_param.adagrad.HasField("initial_range"): sgd_param.adagrad.initial_range = 0.0001 if len(sgd_param.adagrad.weight_bounds) == 0: sgd_param.adagrad.weight_bounds.extend([-10.0, 10.0]) if sgd_param.name == "SparseNaiveSGDRule": if not sgd_param.naive.HasField("learning_rate"): sgd_param.naive.learning_rate = 0.05 if not sgd_param.naive.HasField("initial_range"): sgd_param.naive.initial_range = 0.0001 if len(sgd_param.naive.weight_bounds) == 0: sgd_param.naive.weight_bounds.extend([-10.0, 10.0]) if sgd_param.name == "SparseAdamSGDRule": if not sgd_param.adam.HasField("learning_rate"): sgd_param.adam.learning_rate = 0.001 if not sgd_param.adam.HasField("initial_range"): sgd_param.adam.initial_range = 0.0001 if not sgd_param.adam.HasField("beta1_decay_rate"): sgd_param.adam.beta1_decay_rate = 0.9 if not sgd_param.adam.HasField("beta2_decay_rate"): sgd_param.adam.beta2_decay_rate = 0.999 if not sgd_param.adam.HasField("ada_epsilon"): sgd_param.adam.ada_epsilon = 1e-08 if len(sgd_param.adam.weight_bounds) == 0: sgd_param.adam.weight_bounds.extend([-10.0, 10.0]) def check_embedding_dim(accessor, varname, o_main_program): embedding_dim = 0 for var in o_main_program.list_vars(): if var.name == varname: embedding_dim = var.shape[1] break fea_dim = accessor.fea_dim if fea_dim != embedding_dim: raise ValueError( f"The fea_dim is wrong, it will be sparse_embedding_dim: {embedding_dim}, but got {fea_dim}" ) embedx_dim = accessor.embedx_dim if embedx_dim != embedding_dim - 3: raise ValueError( f"The embedx_dim is wrong, it will be sparse_embedding_dim - 3: {embedding_dim - 3}, but got {embedx_dim}" ) class Accessor: def __init__(self): self.accessor_class = "" self.optimizer = None self.feature_dim = -1 self.embedding_dim = -1 self.optimizer = None def to_string(self, indent): accessor_str = "{}accessor {{{}\n{}}}" attrs = "" attrs += f'accessor_class: "{self.accessor_class}" ' attrs += f"fea_dim: {self.feature_dim} " attrs += f"embedx_dim: {self.embedding_dim} " attrs += "\n" if self.optimizer is not None: attrs += self.optimizer.to_string(indent) return accessor_str.format( conv_indent(indent), attrs, conv_indent(indent) ) class CommonAccessor: def __init__(self): self.accessor_class = "" self.table_name = None self.entry = None self.attrs = [] self.params = [] self.dims = [] self.trainer_num = 0 self.sync = "false" self.table_num = None self.table_dim = None self.initializers = [] self.opt_input_map = {} self.opt_attr_map = {} self.opt_init_map = {} self.define_optimize_map() def define_optimize_map(self): opt_input_map = {} opt_input_map["sgd"] = [("Param", None), ("LearningRate", 1)] opt_input_map["adam"] = [ ("Param", None), ("Moment1", None), ("Moment2", None), ("Beta1Pow", 1), ("Beta2Pow", 1), ("LearningRate", 1), ] opt_input_map["adam_d2sum"] = [ ("Param", None), ("D2Sum", None), ("G2Sum", None), ("Moment", None), ("MomentDecayRate", 1), ("AdaDecayRate", 1), ("AdaEpsilon", 1), ("LearningRate", 1), ] opt_input_map["sum"] = [("Param", None)] opt_input_map["naive_adagrad"] = [ ("Param", None), ("G2Sum", 1), ("LearningRate", 1), ] opt_attr_map = {} opt_attr_map["sgd"] = [] opt_attr_map["sum"] = [] opt_attr_map["naive_adagrad"] = [] opt_attr_map["adam"] = [ ("beta1", "f"), ("beta2", "f"), ("epsilon", "f"), ] opt_attr_map["adam_d2sum"] = [ ("beta1", "f"), ("beta2", "f"), ("epsilon", "f"), ] opt_init_map = {} opt_init_map["gaussian_random"] = ["seed", "mean", "std"] opt_init_map["fill_constant"] = ["value"] opt_init_map["uniform_random"] = ["seed", "min", "max"] opt_init_map["truncated_gaussian_random"] = ["seed", "mean", "std"] self.opt_attr_map = opt_attr_map self.opt_input_map = opt_input_map self.opt_init_map = opt_init_map def parse_entry(self, varname, o_main_program): from paddle.incubate.distributed.fleet.parameter_server.ir.public import ( is_distributed_sparse_op, is_sparse_op, ) for op in o_main_program.global_block().ops: if not is_distributed_sparse_op(op) and not is_sparse_op(op): continue param_name = op.input("W")[0] if param_name == varname and op.type == "lookup_table": self.entry = op.attr('entry') break if param_name == varname and op.type == "lookup_table_v2": self.entry = "none" break def get_shard(self, total_dim, shard_num, pserver_id): # remainder = total_dim % shard_num blocksize = int(total_dim / shard_num + 1) if blocksize * (pserver_id + 1) <= total_dim: return blocksize else: if blocksize * pserver_id < total_dim: return total_dim - blocksize * pserver_id else: return 0 def get_initializer_attr(self, value_name, o_startup_program): l_in = "&" attr_str = "" origin_var_name = value_name for op in o_startup_program.global_block().ops: if ( op.type in self.opt_init_map.keys() and origin_var_name == op.output("Out")[0] ): init_attr = [op.type] for attr in self.opt_init_map[op.type]: init_attr.append(str(op.attr(attr))) attr_str = l_in.join(init_attr) break return attr_str def parse_by_optimizer( self, grad_name, is_sparse, size, single_dim, compiled_strategy, adam_d2sum, ): from paddle.incubate.distributed.fleet.parameter_server.ir.public import ( _get_optimize_ops, ) param_name = compiled_strategy.grad_name_to_param_name[grad_name] main_program, startup_program = compiled_strategy.get_origin_programs() pserver_id = compiled_strategy.get_role_id() pserver_num = len(compiled_strategy.get_ps_endpoints()) optimizer_ops = _get_optimize_ops(main_program) oop = None for op in optimizer_ops: if ("Param" in op.input_names) and ( op.input("Param")[0] == param_name ): oop = op break if oop is None: raise ValueError(f"can not find optimizer for {grad_name}") params = [] dims = [] attrs = [] initializers = [] self.trainer_num = compiled_strategy.get_trainers() self.table_num = size self.table_dim = single_dim if oop.type != 'adam' and adam_d2sum: print('optimization algorithm is not adam, set adam_d2sum False') adam_d2sum = False print("adam_d2sum:", adam_d2sum) if compiled_strategy.is_geo_mode(): param_varnames = self.opt_input_map["sum"] attr_varnames = self.opt_attr_map["sum"] self.accessor_class = "sum" elif compiled_strategy.use_ps_gpu and is_sparse: param_varnames = self.opt_input_map["naive_adagrad"] attr_varnames = self.opt_attr_map["naive_adagrad"] self.accessor_class = "sgd" elif adam_d2sum and not is_sparse: param_varnames = self.opt_input_map["adam_d2sum"] attr_varnames = self.opt_attr_map["adam_d2sum"] self.accessor_class = "adam_d2sum" else: param_varnames = self.opt_input_map[oop.type] attr_varnames = self.opt_attr_map[oop.type] self.accessor_class = oop.type for formal_name, shape in param_varnames: params.append(formal_name) if self.accessor_class == "adam_d2sum": # for dims if shape is None: if is_sparse: shape = single_dim else: shape = self.get_shard(size, pserver_num, pserver_id) dims.append(shape) # for initializers if formal_name == "Param" or formal_name == "LearningRate": param = main_program.global_block().vars[ oop.input(formal_name)[0] ] # TODO: for dense learning_rate, can be different from sparse lr if ( formal_name == "LearningRate" and param.name != "learning_rate_0" ): warnings.warn("will support decay soon") param = main_program.global_block().vars[ "learning_rate_0" ] initializer = self.get_initializer_attr( param.name, startup_program ) elif formal_name == "MomentDecayRate": initializer = "fill_constant&0.99" elif formal_name == "AdaDecayRate": initializer = "fill_constant&0.9999" elif formal_name == "AdaEpsilon": initializer = "fill_constant&1.0e-8" else: initializer = "fill_constant&0" initializers.append(initializer) else: if formal_name == "G2Sum": dims.append(1) initializer = "fill_constant&0" initializers.append(initializer) else: param = main_program.global_block().vars[ oop.input(formal_name)[0] ] if ( formal_name == "LearningRate" and param.name != "learning_rate_0" ): warnings.warn("will support decay soon") param = main_program.global_block().vars[ "learning_rate_0" ] if shape is None: if is_sparse: shape = single_dim else: shape = self.get_shard( size, pserver_num, pserver_id ) dims.append(shape) initializer = self.get_initializer_attr( param.name, startup_program ) initializers.append(initializer) for attr_varname, type_ in attr_varnames: value = oop.attr(attr_varname) attrs.append("&".join([attr_varname, type_, str(value)])) self.params = params self.dims = dims self.initializers = initializers self.attrs = attrs def to_string(self, indent): accessor_str = "{}common {{{}\n{}}}" attrs = "" attrs += f'name: "{self.accessor_class}" ' if self.table_name: attrs += f'table_name: "{self.table_name}" ' if self.entry: attrs += f'entry: "{self.entry}" ' attrs += f"trainer_num: {self.trainer_num} " attrs += f"sync: {self.sync} " if self.table_num: attrs += f"table_num: {self.table_num} " if self.table_dim: attrs += f"table_dim: {self.table_dim} " for param in self.params: attrs += f'params: "{param}" ' for dim in self.dims: attrs += f"dims: {dim} " for initializer in self.initializers: attrs += f'initializers: "{initializer}" ' attrs += "\n" return accessor_str.format( conv_indent(indent), attrs, conv_indent(indent) ) class Tensor: def __init__(self): self.main_program_id = None self.startup_program_id = None self.feed_var_name = None self.fetch_var_name = None self.tensor_table_class = False def to_string(self, indent): program_str = "{}tensor {{{}\n{}}}" attrs = "" attrs += f'feed_var_name: "{self.feed_var_name}" ' attrs += f'fetch_var_name: "{self.fetch_var_name}" ' attrs += f"startup_program_id: {self.startup_program_id} " attrs += f"main_program_id: {self.main_program_id} " attrs += f'tensor_table_class: "{self.tensor_table_class}" ' attrs += "\n" return program_str.format( conv_indent(indent), attrs, conv_indent(indent) ) class Table: def __init__(self): self.id = -1 self.table_class = None self.shard_num = -1 self.type = None self.accessor = None self.common = None self.tensor = None self.accessor_proto = None def to_string(self, indent): # if self.id == 1: # proto_txt = '' # with open('./sparse_table.prototxt') as f: # proto_txt = f.read() # return proto_txt table_str = "{}downpour_table_param {{{}\n{}}}" attrs = "" attrs += f"table_id: {self.id} " attrs += f'table_class: "{self.table_class}" ' attrs += f"shard_num: {self.shard_num} " attrs += f"type: {self.type}" attrs += "\n" indent += 2 if self.accessor_proto is not None: accessor_str = "{}accessor {{{}\n{}}}" accessor_str = accessor_str.format( conv_indent(indent), self.accessor_proto, conv_indent(indent) ) attrs += accessor_str + "\n" elif self.accessor is not None: attrs += self.accessor.to_string(indent) attrs += "\n" if self.tensor is not None: attrs += self.tensor.to_string(indent) attrs += "\n" if self.common is not None: attrs += self.common.to_string(indent) attrs += "\n" return table_str.format(conv_indent(indent), attrs, conv_indent(indent)) class Service: def __init__(self): self.server_class = "BrpcPsServer" self.client_class = "BrpcPsClient" self.service_class = "BrpcPsService" self.start_server_port = 0 self.server_thread_num = 12 def to_string(self, indent): service_str = "{}service_param {{{}\n{}}}" attrs = "" attrs += f'server_class: "{self.server_class}" ' attrs += f'client_class: "{self.client_class}" ' attrs += f'service_class: "{self.service_class}" ' attrs += f"start_server_port: {self.start_server_port} " attrs += f"server_thread_num: {self.server_thread_num} " return service_str.format( conv_indent(indent), attrs, conv_indent(indent) ) class DownpourServer: def __init__(self): self.service = None self.tables = [] def set_service_param(self, service): self.service = service def append_tables(self, table): if not isinstance(table, Table): raise ValueError("only support instance Table") self.tables.append(table) def to_string(self, indent): server_str = "{}downpour_server_param {{{}\n{}}}" table_strs = "" indent += 2 table_strs += "\n" table_strs += self.service.to_string(indent) for table in self.tables: table_strs += "\n" table_strs += table.to_string(indent) return server_str.format( conv_indent(indent), table_strs, conv_indent(indent) ) class Server: def __init__(self): self.servers = [] def add_server(self, server): if not isinstance(server, DownpourServer): raise ValueError("only support instance DownpourServer") self.servers.append(server) def __str__(self): server_str = "server_param {{{}\n}}" indent = 2 servers_str = "" for server in self.servers: servers_str += "\n" servers_str += server.to_string(indent) return server_str.format(servers_str) class DownpourWorker: def __init__(self): self.tables = [] def append_tables(self, table): if not isinstance(table, Table): raise ValueError("only support instance Table") self.tables.append(table) def to_string(self, indent): worker_str = "{}downpour_worker_param {{{}\n{}}}" table_strs = "" indent += 2 for table in self.tables: table_strs += "\n" table_strs += table.to_string(indent) return worker_str.format( conv_indent(indent), table_strs, conv_indent(indent) ) class Worker: def __init__(self): self.workers = [] def add_worker(self, worker): if not isinstance(worker, DownpourWorker): raise ValueError("only support instance DownpourWorker") self.workers.append(worker) def __str__(self): worker_str = "worker_param {{{}\n}}" indent = 2 workers_str = "" for worker in self.workers: workers_str += "\n" workers_str += worker.to_string(indent) return worker_str.format(workers_str) class fsClient: def __init__(self, proto): self.proto = proto self.uri = proto.uri self.user = proto.user self.passwd = proto.passwd self.hadoop_bin = proto.hadoop_bin def to_string(self): from google.protobuf import text_format proto_txt = text_format.MessageToString(self.proto) if proto_txt: fs_str = "fs_client_param {{\n{}}}" return fs_str.format(proto_txt) else: return "" class TheOnePSRuntime(RuntimeBase): def __init__(self): super().__init__() self._communicator = None self._server = None self._worker = base.core.DistFleetWrapper() self._server_sub_program = [] self._heter_client = None def _set_basic_info(self, context): self.context = context self.role_maker = context["role_maker"] self.origin_main_program = context["origin_main_program"] self.origin_startup_program = context["origin_startup_program"] self.async_strategy = self._get_distributed_strategy() self.compiled_strategy = self.build_compiled_strategy() def _get_distributed_strategy(self): strategy = None from paddle.incubate.distributed.fleet.parameter_server.distribute_transpiler.distributed_strategy import ( StrategyFactory, ) dist_strategy = self.context["valid_strategy"] k_steps = dist_strategy.a_sync_configs["k_steps"] if not dist_strategy.a_sync and k_steps == 0: strategy = StrategyFactory.create_sync_strategy() if dist_strategy.a_sync and k_steps == 0: strategy = StrategyFactory.create_async_strategy() if dist_strategy.a_sync and k_steps > 0: strategy = StrategyFactory.create_geo_strategy(k_steps) if not strategy: raise ValueError("k_steps must be invalid value, please check") if dist_strategy.a_sync_configs["use_ps_gpu"]: strategy.use_ps_gpu = True return strategy def build_compiled_strategy(self): from paddle.incubate.distributed.fleet.parameter_server.ir.public import ( CompileTimeStrategy, ) compiled_config = CompileTimeStrategy( self.origin_main_program, self.origin_main_program, self.async_strategy, self.role_maker, ) if self.async_strategy.use_ps_gpu: compiled_config.use_ps_gpu = True return compiled_config def _init_worker(self): from paddle.incubate.distributed.fleet.parameter_server.distribute_transpiler.distributed_strategy import ( SyncStrategy, ) is_sync = self.compiled_strategy.is_sync_mode() worker = self._get_fleet_proto(is_server=False, is_sync=is_sync) server = self._get_fleet_proto(is_server=True, is_sync=is_sync) dist_strategy = self.context["valid_strategy"] use_ps_gpu = dist_strategy.a_sync_configs["use_ps_gpu"] if use_ps_gpu: main_program = self.context['loss'].block.program if not main_program._fleet_opt: main_program._fleet_opt = {} main_program._fleet_opt["use_ps_gpu"] = True gpus_env = os.getenv("FLAGS_selected_gpus") main_program._fleet_opt["worker_places"] = [ int(s) for s in gpus_env.split(",") ] def sync_strategy_envs(): kwargs = {} kwargs["pserver_endpoints"] = ( self.role_maker._get_pserver_endpoints() ) kwargs["trainer_id"] = self.role_maker._worker_index() return kwargs proto_txt = str(worker) + "\n" + str(server) with open('proto_txt', 'w') as f: f.write(proto_txt) debug = bool(int(os.getenv("PSERVER_DEBUG", "0"))) if debug: print(f"worker: \n{proto_txt}") endpoints = self.compiled_strategy.get_ps_endpoints() string_hosts = [] for idx, ep in enumerate(endpoints): host, port = ep.split(":") pshost = base.core.PSHost(host, int(port), idx) string_hosts.append(pshost.serialize_to_string()) dense_map = self.compiled_strategy.get_the_one_recv_context( split_dense_table=self.role_maker._is_heter_parameter_server_mode ) send_ctx = self.compiled_strategy.get_the_one_send_context( split_dense_table=self.role_maker._is_heter_parameter_server_mode, use_origin_program=self.role_maker._is_heter_parameter_server_mode, ep_list=endpoints, ) trainer_config = self.async_strategy.get_trainer_runtime_config() debug = bool(int(os.getenv("PSERVER_DEBUG", "0"))) if debug: print(f"worker: \n{proto_txt}") print("communicator send_ctx:") for key in send_ctx: print(f"{key}: {send_ctx[key]}") for key in dense_map: print(f"{key}: {dense_map[key]}") kwargs = {} kwargs['need_global_step'] = "0" kwargs["trainer_id"] = self.role_maker._role_id() kwargs["trainers"] = self.role_maker._worker_num() # if self.role_maker._is_heter_worker(): # kwargs["trainer_id"] += kwargs["trainers"] for table in server.servers[0].tables: if table.table_class == "BarrierTable": kwargs["barrier_table_id"] = table.id break if isinstance(self.async_strategy, SyncStrategy): sync_kwargs = sync_strategy_envs() kwargs.update(sync_kwargs) from paddle.distributed.communicator import Communicator, HeterClient self._communicator = Communicator( trainer_config.mode, kwargs, trainer_config.get_communicator_flags() ) self._communicator.init_with_ctx( send_ctx, dense_map, proto_txt, string_hosts, base.global_scope() ) from paddle.distributed import fleet fleet.util.barrier() info = self._communicator.get_client_info() if isinstance(info, list) and len(info) > 0: all_info = self.role_maker._all_gather(info[0]) # for unittest if not isinstance(all_info, list): warnings.warn("gloo may not initialize correctly") all_info = [all_info] self._communicator.set_clients(all_info) # create_c2c_connection default param: # pserver_timeout_ms=500000 # pserver_connect_timeout_ms=10000 # max_retry=3 self._communicator.create_client_to_client_connection() print('create c2c connection done') else: print('cannot create c2c connection') dist_strategy = self.context["valid_strategy"] is_test = bool(int(os.getenv("TEST_MODE", "0"))) if ( self.role_maker._is_first_worker() and self.role_maker._is_heter_parameter_server_mode ): # for ps-heter mode load all parameters on first_worker init_params = self.compiled_strategy.get_the_one_recv_context( split_dense_table=True, use_origin_program=True ) else: init_params = dense_map if not is_test: self._communicator.init_params(init_params) fleet.util.barrier() self._communicator.pull_dense(init_params) fleet.util.barrier() if not self._communicator.is_running(): self._communicator.start() else: warnings.warn("communicator has been initialized, skip") launch_barrier = dist_strategy.a_sync_configs["launch_barrier"] launch_barrier_flag = int(os.getenv("FLAGS_LAUNCH_BARRIER", "1")) if launch_barrier and launch_barrier_flag: # for trainer wait server ready wait_server_ready(self.role_maker._get_pserver_endpoints()) if ( self.role_maker._is_heter_parameter_server_mode and self.role_maker._get_next_trainers() != [] ): wait_server_ready(self.role_maker._get_next_trainers()) if self.role_maker._is_heter_parameter_server_mode: previous_trainers = [] if self.role_maker._get_previous_trainers() != []: previous_trainers = self.role_maker._get_previous_trainers() next_trainers = [] if self.role_maker._get_next_trainers() != []: next_trainers = self.role_maker._get_next_trainers() self._heter_client = HeterClient( next_trainers, previous_trainers, self.role_maker._role_id() ) def _push_sparse_param( self, var_name, table_id=-1, scope=base.global_scope() ): self._communicator.push_sparse_param(var_name, table_id, scope) def _get_executor(self): executor = base.Executor(base.CPUPlace()) if self.role_maker._is_heter_parameter_server_mode: if self.role_maker._is_heter_worker(): heter_device_type = self.role_maker._heter_device_type().upper() if heter_device_type not in ["GPU", "XPU", "CPU"]: raise ValueError( f"Heter Worker Not Support Device {heter_device_type}" ) if heter_device_type == "GPU": executor = Executor( base.CUDAPlace( int(os.getenv("FLAGS_selected_gpus", "0")) ) ) elif heter_device_type == "XPU": executor = Executor( base.XPUPlace( int(os.getenv("FLAGS_selected_xpus", "0")) ) ) return executor def _get_fleet_proto(self, is_server, is_sync, **kwargs): def _build_merge_accessor(ctx): accessor = Accessor() accessor.accessor_class = "CommMergeAccessor" accessor.optimizer = None if ctx.is_sparse(): accessor.feature_dim = ctx.sections()[0] accessor.embedding_dim = ctx.sections()[1] else: accessor.feature_dim = ctx.sections()[0] accessor.embedding_dim = 1 return accessor def _build_barrier_table(idx): table = Table() table.id = idx table.type = "PS_OTHER_TABLE" table.table_class = "BarrierTable" table.shard_num = 256 accessor = Accessor() accessor.accessor_class = "CommMergeAccessor" accessor.optimizer = None accessor.feature_dim = 0 accessor.embedding_dim = 0 table.accessor = accessor common = CommonAccessor() common.table_name = "barrier_table" trainer_num = self.compiled_strategy.get_trainers() if self.role_maker._is_heter_parameter_server_mode: trainer_num += len( self.role_maker._get_heter_worker_endpoints() ) common.trainer_num = trainer_num common.attrs = "" common.dims = [] common.params = [] table.common = common return table def _build_tensor_table(idx, tensor_dict): table = Table() table.id = idx table.type = "PS_OTHER_TABLE" table.table_class = tensor_dict["tensor_table_class"] table.shard_num = 256 accessor = Accessor() accessor.accessor_class = "CommMergeAccessor" accessor.optimizer = None accessor.feature_dim = 0 accessor.embedding_dim = 0 table.accessor = accessor common = CommonAccessor() common.table_name = tensor_dict["feed_var_name"] common.trainer_num = self.compiled_strategy.get_trainers() common.attrs = "" common.dims = [] common.params = [] table.common = common tensor = Tensor() tensor.main_program_id = tensor_dict["main_program_id"] tensor.startup_program_id = tensor_dict["startup_program_id"] tensor.feed_var_name = tensor_dict["feed_var_name"] tensor.fetch_var_name = tensor_dict["fetch_var_name"] tensor.tensor_table_class = tensor_dict["tensor_table_class"] table.tensor = tensor return table def _add_tensor_table(tables): tensor_table_dict = self.compiled_strategy.get_tensor_table_dict() program_idx = 0 for table_name in tensor_table_dict: if tensor_table_dict[table_name]["startup_program"] is not None: tensor_table_dict[table_name]["startup_program_id"] = ( program_idx ) self._server_sub_program.append( tensor_table_dict[table_name]["startup_program"].desc ) program_idx += 1 if tensor_table_dict[table_name]["main_program"] is not None: tensor_table_dict[table_name]["main_program_id"] = ( program_idx ) self._server_sub_program.append( tensor_table_dict[table_name]["main_program"].desc ) program_idx += 1 # Todo: Hard code for lr_decay table apply table id new_table = _build_tensor_table( len(tables), tensor_table_dict[table_name] ) tables.append(new_table) return tables def _get_tables(): send_ctx = self.compiled_strategy.get_the_one_send_context( use_origin_program=True, split_dense_table=self.role_maker._is_heter_parameter_server_mode, ) tables = [] for idx, (name, ctx) in enumerate(send_ctx.items()): if ctx.is_tensor_table() or len(ctx.origin_varnames()) < 1: continue table = Table() table.id = ctx.table_id() common = CommonAccessor() if ctx.is_sparse(): table.type = "PS_SPARSE_TABLE" table.shard_num = 256 common.table_name = ( self.compiled_strategy.grad_name_to_param_name[ ctx.origin_varnames()[0] ] ) if self.compiled_strategy.is_geo_mode(): table.table_class = "MemorySparseGeoTable" else: all_table_proto = self.context[ "user_defined_strategy" ].sparse_table_configs table_proto = all_table_proto.add() for proto in all_table_proto: if proto.table_name == common.table_name: table_proto = proto break if table_proto.HasField("table_class"): table.table_class = table_proto.table_class else: table.table_class = parse_table_class( common.table_name, self.origin_main_program ) if table.table_class != 'MemorySparseTable': table.table_class = 'MemorySparseTable' warnings.warn( "The PS mode must use MemorySparseTable." ) if table_proto.HasField("shard_num"): table.shard_num = table_proto.shard_num else: table.shard_num = 1000 warnings.warn( "The shard_num of sparse table is not set, use default value 1000." ) if table_proto.accessor.ByteSize() == 0: warnings.warn( "The accessor of sparse table is not set, use default value." ) get_default_accessor_proto( table_proto.accessor, common.table_name, self.origin_main_program, ) check_embedding_dim( table_proto.accessor, common.table_name, self.origin_main_program, ) from google.protobuf import text_format table.accessor_proto = text_format.MessageToString( table_proto.accessor ) else: table.type = "PS_DENSE_TABLE" table.table_class = "MemoryDenseTable" table.shard_num = 256 common.table_name = "MergedDense" adam_d2sum = self.context["user_defined_strategy"].adam_d2sum common.parse_by_optimizer( ctx.origin_varnames()[0], ctx.is_sparse(), ctx.sections()[0], ctx.sections()[1] if ctx.is_sparse() else 1, self.compiled_strategy, adam_d2sum, ) if ctx.is_sparse(): common.parse_entry( common.table_name, self.origin_main_program ) if is_sync: common.sync = "true" else: common.sync = "false" table.common = common if table.table_class != 'MemorySparseTable': accessor = _build_merge_accessor(ctx) table.accessor = accessor tables.append(table) tensor_table_dict = self.compiled_strategy.get_tensor_table_dict() if len(tensor_table_dict) > 0: tables = _add_tensor_table(tables) else: empty_program = Program() self._server_sub_program.append(empty_program.desc) barrier_table = _build_barrier_table(len(tables)) tables.append(barrier_table) return tables if is_server: server = Server() downpour_server = DownpourServer() service = Service() dist_strategy = self.context["valid_strategy"] use_ps_gpu = dist_strategy.a_sync_configs["use_ps_gpu"] if use_ps_gpu: service.server_class = "PsLocalServer" service.client_class = "PsLocalClient" downpour_server.set_service_param(service) tables = _get_tables() downpour_server.tables = tables server.add_server(downpour_server) return server else: worker = Worker() downpour_worker = DownpourWorker() tables = _get_tables() downpour_worker.tables = tables worker.add_worker(downpour_worker) return worker def _init_server(self, dirname=None, var_names=None, **kwargs): role_id = self.compiled_strategy.get_role_id() endpoints = self.compiled_strategy.get_ps_endpoints() is_sync = self.compiled_strategy.is_sync_mode() trainers = self.compiled_strategy.get_trainers() if self.role_maker._is_heter_parameter_server_mode: trainers += len(self.role_maker._get_heter_worker_endpoints()) server = self._get_fleet_proto(is_server=True, is_sync=is_sync) proto_txt = str(server) fs_client = fsClient( self.context["user_defined_strategy"].fs_client_param ) proto_txt = proto_txt + "\n" + fs_client.to_string() debug = bool(int(os.getenv("PSERVER_DEBUG", "0"))) if debug: print(f"server: \n{proto_txt}") string_hosts = [] for idx, ep in enumerate(endpoints): host, port = ep.split(":") pshost = base.core.PSHost(host, int(port), idx) string_hosts.append(pshost.serialize_to_string()) self._server = base.core.DistFleetWrapper() self._server.init_server( proto_txt, string_hosts, role_id, trainers, self._server_sub_program ) from paddle.incubate.distributed.fleet.parameter_server.ir.public import ( get_sparse_tablenames, ) dist_varnames = get_sparse_tablenames(self.origin_main_program, True) sparse_varnames = get_sparse_tablenames(self.origin_main_program, False) distributed_varnames = dist_varnames + sparse_varnames if var_names is None: load_varnames = distributed_varnames else: for var_name in var_names: if var_name not in distributed_varnames: raise ValueError( f"fleet.init server can only load sparse variables in {distributed_varnames}" ) load_varnames = var_names if dirname is None or not load_varnames: return sparse_table_maps = {} for table in server.servers[0].tables: if table.type == "PS_SPARSE_TABLE" and table.common is not None: sparse_table_maps[table.common.table_name] = table.id dirname = os.path.normpath(dirname) pserver_id = self.role_maker._role_id() for var_name in load_varnames: table_id = sparse_table_maps[var_name] # path = os.path.join(dirname, var_name + PSERVER_SAVE_SUFFIX, # "{}.block{}.txt".format(var_name, pserver_id)) # meta = os.path.join(dirname, var_name + PSERVER_SAVE_SUFFIX, # "{}.block{}.meta".format(var_name, pserver_id)) self._server.load_sparse(dirname, "0", table_id) def _run_server(self): ep = self.compiled_strategy.get_ps_endpoint() host, port = ep.split(":") self._server.run_server(host, int(port)) def _stop_worker(self): self._communicator.stop() if self.role_maker._is_heter_parameter_server_mode: assert self._heter_client is not None, ( "heter client should not be None in heterps mode" ) self._heter_client.stop() # executor = self._get_executor() # executor.close() @staticmethod def __exclude_vars(exclude_var_names=[]): def is_valid(var): if var.name in exclude_var_names: return False from paddle.incubate.distributed.fleet.parameter_server.ir.public import ( _get_varname_parts, ) origin_varname, _, _ = _get_varname_parts(var.name) if origin_varname.endswith("@GRAD"): return False if origin_varname == "learning_rate_0": return False if ( var.desc.type() == core.VarDesc.VarType.FEED_MINIBATCH or var.desc.type() == core.VarDesc.VarType.FETCH_LIST or var.desc.type() == core.VarDesc.VarType.READER ): return False return var.persistable return is_valid def _get_inference_model_path(self, dirname): if dirname.startswith("afs:") or dirname.startswith("hdfs:"): model_path = "./dnn_plugin" else: model_path = os.path.join(dirname, "dnn_plugin") return model_path def _save_sparse_params( self, executor, dirname, context, main_program, mode ): from paddle.incubate.distributed.fleet.parameter_server.ir.public import ( get_sparse_tablenames, ) distributed_varnames = get_sparse_tablenames( self.compiled_strategy.origin_main_program, True ) values = [] model_path = self._get_inference_model_path(dirname) for id, names in context.items(): if names[0] not in distributed_varnames: # only save sparse param to local try: self._worker.recv_and_save_model(id, model_path) except: pass # save sparse & distributed param on server self._worker.save_one_model(id, dirname, mode) values.extend(names) # self._worker.save_all_model(dirname, mode) return values def _save_distributed_persistables( self, executor, dirname, main_program, mode=0 ): denses = self.compiled_strategy.get_the_one_recv_context( is_dense=True, split_dense_table=self.role_maker._is_heter_parameter_server_mode, use_origin_program=True, ) sparses = self.compiled_strategy.get_the_one_recv_context( is_dense=False, split_dense_table=self.role_maker._is_heter_parameter_server_mode, use_origin_program=True, ) sparse_varnames = self._save_sparse_params( executor, dirname, sparses, main_program, mode ) recv_dense_varnames = [] for id, names in denses.items(): recv_dense_varnames.extend(names) self._communicator.pull_dense(denses) saved_varnames = sparse_varnames remaining_vars = list( filter( TheOnePSRuntime.__exclude_vars(saved_varnames), main_program.list_vars(), ) ) import paddle for var in remaining_vars: # if var.name not in recv_dense_varnames: # continue tensor = var.get_value() paddle.save( tensor, os.path.join(dirname, var.name), use_binary_format=True ) def _ps_inference_save_persistables( self, executor, dirname, main_program=None, mode=0, **kwargs ): """ This function filters out all variables with `persistable==True` from the give `main_program` and then saves these variables to the folder `dirname` or file `filename`. The `dirname` is used to specify the folder where persistable variables are going to be saved. If you would like to save variables in separate files, set `filename` None; if you would like to save all variables in a single file, use `filename` to specify the file name. """ if not isinstance(executor, Executor): raise TypeError( "in fleet.save() function, executor must be as Executor type" ) if main_program is None: main_program = self.compiled_strategy.get_origin_ps_main_program() if isinstance(main_program, CompiledProgram): raise TypeError( "in fleet.save() function, main_program must be as Program type, CompiledProgram is not allowed" ) # Todo(MrChengmo): Save optimizer status # self._save_distributed_persistables(executor, dirname, main_program, # mode) self._worker.save_all_model(dirname, mode) def _ps_inference_save_inference_model( self, executor, dirname, feeded_var_names, target_vars, main_program=None, export_for_deployment=True, mode=0, ): """ Prune the given `main_program` to build a new program especially for inference, and then save it and all related parameters to given `dirname` by the `executor`. """ if not isinstance(executor, Executor): raise TypeError( "in fleet.save() function, executor must be as Executor type" ) import paddle program = ( self.origin_main_program if main_program is None else main_program ) if isinstance(program, CompiledProgram): raise TypeError( "in fleet.save() function, main_program must be as Program type, CompiledProgram is not allowed" ) feed_vars = [ program.global_block().var(name) for name in feeded_var_names ] infer_program = paddle.static.normalize_program( program, feed_vars, target_vars ) infer_program._copy_dist_param_info_from(program) model_path = self._get_inference_model_path(dirname) model_basename = "__model__" model_basename = os.path.join(model_path, model_basename) paddle.save(infer_program, model_basename) sparses = self.compiled_strategy.get_the_one_recv_context( is_dense=False, split_dense_table=self.role_maker._is_heter_parameter_server_mode, use_origin_program=True, ) sparse_names = self._save_sparse_params( executor, dirname, sparses, main_program, mode ) denses = self.compiled_strategy.get_the_one_recv_context( is_dense=True, split_dense_table=self.role_maker._is_heter_parameter_server_mode, use_origin_program=True, ) # TODO(zhaocaibei123): for GEO: should call GeoCommunicator::RecvDense self._communicator.pull_dense(denses) generate_vars = self.context[ "user_defined_strategy" ].trainer_desc_configs["stat_var_names"] generate_vars = list(generate_vars) remaining_vars = list( filter( TheOnePSRuntime.__exclude_vars(sparse_names), infer_program.list_vars(), ) ) for var in remaining_vars: tensor = var.get_value() paddle.save( tensor, os.path.join(model_path, var.name), use_binary_format=True, ) def _save_inference_model(self, *args, **kwargs): self._ps_inference_save_inference_model(*args, **kwargs) def _save_persistables(self, *args, **kwargs): self._ps_inference_save_persistables(*args, **kwargs) def _load_sparse_params(self, dirname, context, main_program, mode): from paddle.incubate.distributed.fleet.parameter_server.ir.public import ( get_sparse_tablenames, ) distributed_varnames = get_sparse_tablenames( self.compiled_strategy.origin_main_program, True ) values = [] for id, names in context.items(): if names[0] not in distributed_varnames: # TODO: only load sparse param from local warnings.warn("varname is not in distributed_varnames, pass") # load sparse & distributed param on server self._worker.load_one_table(id, dirname, mode) values.extend(names) return values def _ps_inference_load_inference_model( self, dirname, mode=0, main_program=None ): if main_program is None: main_program = self.compiled_strategy.get_origin_ps_main_program() if isinstance(main_program, CompiledProgram): raise TypeError( "in fleet.save() function, main_program must be as Program type, CompiledProgram is not allowed" ) denses = self.compiled_strategy.get_the_one_recv_context( is_dense=True, split_dense_table=self.role_maker._is_heter_parameter_server_mode, use_origin_program=True, ) sparses = self.compiled_strategy.get_the_one_recv_context( is_dense=False, split_dense_table=self.role_maker._is_heter_parameter_server_mode, use_origin_program=True, ) sparse_varnames = self._load_sparse_params( dirname, sparses, main_program, mode ) recv_dense_varnames = [] for id, names in denses.items(): recv_dense_varnames.extend(names) loaded_varnames = sparse_varnames remaining_vars = list( filter( TheOnePSRuntime.__exclude_vars(loaded_varnames), main_program.list_vars(), ) ) if dirname.startswith("afs:") or dirname.startswith("hdfs:"): model_path = "./dnn_plugin" else: model_path = os.path.join(dirname, "dnn_plugin") import paddle for var in remaining_vars: if var.name not in recv_dense_varnames: continue tensor = paddle.load(os.path.join(model_path, var.name)) var.set_value(tensor) self._communicator.init_params(denses) def _load_distributed_persistables(self, path, mode): self._worker.load_model(path, mode) def load_model(self, path, mode): if mode == 0 or mode == 3: self._load_distributed_persistables(path, mode) else: self._ps_inference_load_inference_model(path, mode) # self._load_distributed_persistables(path, mode=mode) def _shrink(self, threshold=None): if threshold is not None: warnings.warn( "The param threshold is not used in MemorySparseTable, if you need to shrink, please set the config of accessor" ) else: threshold = 0 from paddle.distributed import fleet fleet.util.barrier() if self.role_maker._is_first_worker(): sparses = self.compiled_strategy.get_the_one_recv_context( is_dense=False, split_dense_table=self.role_maker._is_heter_parameter_server_mode, use_origin_program=True, ) for id, names in sparses.items(): self._worker.shrink_sparse_table(id, threshold) fleet.util.barrier()