diff --git a/test/deprecated/ps/config_gpubox.yaml b/test/deprecated/ps/config_gpubox.yaml deleted file mode 100755 index aeea3bddeabe1..0000000000000 --- a/test/deprecated/ps/config_gpubox.yaml +++ /dev/null @@ -1,55 +0,0 @@ -# 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. - -# workspace -#workspace: "models/rank/dnn" - - -runner: - train_data_dir: "data/sample_data/train" - train_reader_path: "criteo_reader" # importlib format - use_gpu: True - use_auc: False - train_batch_size: 32 - epochs: 3 - print_interval: 10 - model_save_path: "output_model_dnn_queue" - - sync_mode: "gpubox" - thread_num: 30 - reader_type: "InmemoryDataset" # DataLoader / QueueDataset / RecDataset / InmemoryDataset - pipe_command: "python dataset_generator_criteo.py" - dataset_debug: False - split_file_list: False - - infer_batch_size: 2 - infer_reader_path: "criteo_reader" # importlib format - test_data_dir: "data/sample_data/train" - infer_load_path: "output_model_dnn_queue" - infer_start_epoch: 0 - infer_end_epoch: 3 -# hyper parameters of user-defined network -hyper_parameters: - # optimizer config - optimizer: - class: Adam - learning_rate: 0.001 - strategy: async - # user-defined pairs - sparse_inputs_slots: 27 - sparse_feature_number: 1024 - sparse_feature_dim: 9 - dense_input_dim: 13 - fc_sizes: [512, 256, 128, 32] - distributed_embedding: 0 diff --git a/test/deprecated/ps/cpu_async_ps_config.yaml b/test/deprecated/ps/cpu_async_ps_config.yaml deleted file mode 100755 index 6a843865a609a..0000000000000 --- a/test/deprecated/ps/cpu_async_ps_config.yaml +++ /dev/null @@ -1,33 +0,0 @@ -# Copyright (c) 2022 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. - -# refer to PaddleRec/models/rank/dnn/benchmark.yaml - -hyper_parameters: - optimizer: - class: Adam - learning_rate: 0.0001 - adam_lazy_mode: True - sparse_inputs_slots: 27 - sparse_feature_number: 1000001 - sparse_feature_dim: 10 - dense_input_dim: 13 - fc_sizes: [400, 400, 400] - -runner: - sync_mode: "async" # sync / async / geo / heter - thread_num: 16 - use_gpu: 0 - - model_path: "../ps_dnn_model.py" diff --git a/test/deprecated/ps/cpu_geo_ps_config.yaml b/test/deprecated/ps/cpu_geo_ps_config.yaml deleted file mode 100644 index f6864cacd6265..0000000000000 --- a/test/deprecated/ps/cpu_geo_ps_config.yaml +++ /dev/null @@ -1,34 +0,0 @@ -# Copyright (c) 2022 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. - -# refer to PaddleRec/models/rank/dnn/benchmark.yaml - -hyper_parameters: - optimizer: - class: Adam - learning_rate: 0.0001 - adam_lazy_mode: True - sparse_inputs_slots: 27 - sparse_feature_number: 1000001 - sparse_feature_dim: 10 - dense_input_dim: 13 - fc_sizes: [400, 400, 400] - -runner: - geo_step: 400 - sync_mode: "geo" - thread_num: 16 - use_gpu: 0 - - model_path: "../ps_dnn_model.py" diff --git a/test/deprecated/ps/cpu_sync_ps_config.yaml b/test/deprecated/ps/cpu_sync_ps_config.yaml deleted file mode 100644 index 449856fe12eee..0000000000000 --- a/test/deprecated/ps/cpu_sync_ps_config.yaml +++ /dev/null @@ -1,33 +0,0 @@ -# Copyright (c) 2022 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. - -# refer to PaddleRec/models/rank/dnn/benchmark.yaml - -hyper_parameters: - optimizer: - class: Adam - learning_rate: 0.0001 - adam_lazy_mode: True - sparse_inputs_slots: 27 - sparse_feature_number: 1000001 - sparse_feature_dim: 10 - dense_input_dim: 13 - fc_sizes: [400, 400, 400] - -runner: - sync_mode: "sync" - thread_num: 16 - use_gpu: 0 - - model_path: "../ps_dnn_model.py" diff --git a/test/deprecated/ps/fl_async_ps_config.yaml b/test/deprecated/ps/fl_async_ps_config.yaml deleted file mode 100755 index 99f4b9f938c4d..0000000000000 --- a/test/deprecated/ps/fl_async_ps_config.yaml +++ /dev/null @@ -1,37 +0,0 @@ -# Copyright (c) 2022 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. - -# refer to PaddleRec/models/rank/dnn/benchmark.yaml - -hyper_parameters: - optimizer: - class: Adam - learning_rate: 0.0001 - adam_lazy_mode: True - sparse_inputs_slots: 27 - sparse_feature_number: 1000001 - sparse_feature_dim: 10 - dense_input_dim: 13 - fc_sizes: [] - -runner: - sync_mode: "async" # sync / async / geo / heter - is_fl_ps_mode: 1 - reader_thread_num: 16 - use_gpu: 0 - batch_size: 2 - train_files_path: "./train_data" - epoch_num: 4 - - model_path: "../ps_dnn_model.py" diff --git a/test/deprecated/ps/gpu_ps_config.yaml b/test/deprecated/ps/gpu_ps_config.yaml deleted file mode 100644 index aef4c1eb4ba2c..0000000000000 --- a/test/deprecated/ps/gpu_ps_config.yaml +++ /dev/null @@ -1,35 +0,0 @@ -# Copyright (c) 2022 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. - -# refer to PaddleRec/models/rank/dnn/config_gpubox.yaml - -hyper_parameters: - optimizer: - class: Adam - learning_rate: 0.001 - strategy: async - sparse_inputs_slots: 27 - sparse_feature_number: 1024 - sparse_feature_dim: 11 - dense_input_dim: 13 - fc_sizes: [512, 256, 128, 32] - distributed_embedding: 0 - -runner: - geo_step: 400 - sync_mode: "gpubox" - thread_num: 16 - use_gpu: 1 - - model_path: "../ps_dnn_model.py" diff --git a/test/deprecated/ps/heter_ps_config.yaml b/test/deprecated/ps/heter_ps_config.yaml deleted file mode 100644 index 0cab383a08815..0000000000000 --- a/test/deprecated/ps/heter_ps_config.yaml +++ /dev/null @@ -1,34 +0,0 @@ -# Copyright (c) 2022 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. - - -hyper_parameters: - optimizer: - class: Adam - learning_rate: 0.0001 - strategy: async # 有用 - sparse_inputs_slots: 27 - sparse_feature_number: 1024 - sparse_feature_dim: 11 - dense_input_dim: 13 - fc_sizes: [512, 256, 128, 32] - distributed_embedding: 0 - -runner: - sync_mode: "heter" - thread_num: 8 - micro_num: 8 # micro batch num for each thread - pipeline: True - - model_path: "../ps_dnn_model.py" diff --git a/test/deprecated/ps/ps_dnn_model.py b/test/deprecated/ps/ps_dnn_model.py deleted file mode 100755 index 625d106c1f3e8..0000000000000 --- a/test/deprecated/ps/ps_dnn_model.py +++ /dev/null @@ -1,389 +0,0 @@ -# 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 math - -import paddle -from paddle import nn - - -class DNNLayer(nn.Layer): - def __init__( - self, - sparse_feature_number, - sparse_feature_dim, - dense_feature_dim, - num_field, - layer_sizes, - sync_mode=None, - ): - super().__init__() - self.sync_mode = sync_mode - self.sparse_feature_number = sparse_feature_number - self.sparse_feature_dim = sparse_feature_dim - self.dense_feature_dim = dense_feature_dim - self.num_field = num_field - self.layer_sizes = layer_sizes - - self.embedding = paddle.nn.Embedding( - self.sparse_feature_number, - self.sparse_feature_dim, - sparse=True, - weight_attr=paddle.ParamAttr( - name="SparseFeatFactors", - initializer=paddle.nn.initializer.Uniform(), - ), - ) - - sizes = ( - [sparse_feature_dim * num_field + dense_feature_dim] - + self.layer_sizes - + [2] - ) - acts = ["relu" for _ in range(len(self.layer_sizes))] + [None] - self._mlp_layers = [] - for i in range(len(layer_sizes) + 1): - linear = paddle.nn.Linear( - in_features=sizes[i], - out_features=sizes[i + 1], - weight_attr=paddle.ParamAttr( - initializer=paddle.nn.initializer.Normal( - std=1.0 / math.sqrt(sizes[i]) - ) - ), - ) - self.add_sublayer('linear_%d' % i, linear) - self._mlp_layers.append(linear) - if acts[i] == 'relu': - act = paddle.nn.ReLU() - self.add_sublayer('act_%d' % i, act) - self._mlp_layers.append(act) - - def forward(self, sparse_inputs, dense_inputs): - sparse_embs = [] - for s_input in sparse_inputs: - if self.sync_mode == "gpubox": - emb = paddle.static.nn.sparse_embedding( - input=s_input, - size=[self.sparse_feature_number, self.sparse_feature_dim], - param_attr=paddle.ParamAttr(name="embedding"), - ) - else: - emb = self.embedding(s_input) - emb = paddle.reshape(emb, shape=[-1, self.sparse_feature_dim]) - # emb.stop_gradient = True - sparse_embs.append(emb) - - y_dnn = paddle.concat(x=sparse_embs + [dense_inputs], axis=1) - - if self.sync_mode == 'heter': - with paddle.base.device_guard('gpu'): - for n_layer in self._mlp_layers: - y_dnn = n_layer(y_dnn) - else: - for n_layer in self._mlp_layers: - y_dnn = n_layer(y_dnn) - - return y_dnn - - -class FlDNNLayer(nn.Layer): - def __init__( - self, - sparse_feature_number, - sparse_feature_dim, - dense_feature_dim, - sparse_number, - sync_mode=None, - ): - super().__init__() - - self.PART_A_DEVICE_FlAG = 'gpu:0' - self.PART_A_JOINT_OP_DEVICE_FlAG = 'gpu:2' - self.PART_B_DEVICE_FlAG = 'gpu:1' - self.PART_B_JOINT_OP_DEVICE_FlAG = 'gpu:3' - - self.sync_mode = sync_mode - self.sparse_feature_number = sparse_feature_number - self.sparse_feature_dim = sparse_feature_dim - self.slot_num = sparse_number - self.dense_feature_dim = dense_feature_dim - - layer_sizes_a = [ - self.slot_num * self.sparse_feature_dim, - 5, - 7, - ] # for test - layer_sizes_b = [self.dense_feature_dim, 6, 7] - layer_sizes_top = [7, 2] - - self.embedding = paddle.nn.Embedding( - self.sparse_feature_number, - self.sparse_feature_dim, - sparse=True, - weight_attr=paddle.ParamAttr( - name="SparseFeatFactors", - initializer=paddle.nn.initializer.Uniform(), - ), - ) - - # part_a fc - acts = ["relu" for _ in range(len(layer_sizes_a))] - self._mlp_layers_a = [] - for i in range(len(layer_sizes_a) - 1): - linear = paddle.nn.Linear( - in_features=layer_sizes_a[i], - out_features=layer_sizes_a[i + 1], - weight_attr=paddle.ParamAttr( - initializer=paddle.nn.initializer.Normal( - std=1.0 / math.sqrt(layer_sizes_a[i]) - ) - ), - ) - self.add_sublayer('linear_%d' % i, linear) - self._mlp_layers_a.append(linear) - act = paddle.nn.ReLU() - self.add_sublayer('act_%d' % i, act) - self._mlp_layers_a.append(act) - - # part_b fc - acts = ["relu" for _ in range(len(layer_sizes_b))] - self._mlp_layers_b = [] - for i in range(len(layer_sizes_b) - 1): - linear = paddle.nn.Linear( - in_features=layer_sizes_b[i], - out_features=layer_sizes_b[i + 1], - weight_attr=paddle.ParamAttr( - initializer=paddle.nn.initializer.Normal( - std=1.0 / math.sqrt(layer_sizes_b[i]) - ) - ), - ) - self.add_sublayer('linear_%d' % i, linear) - self._mlp_layers_b.append(linear) - act = paddle.nn.ReLU() - self.add_sublayer('act_%d' % i, act) - self._mlp_layers_b.append(act) - - # top fc - acts = ["relu" for _ in range(len(layer_sizes_top))] - self._mlp_layers_top = [] - for i in range(len(layer_sizes_top) - 1): - linear = paddle.nn.Linear( - in_features=layer_sizes_top[i], - out_features=layer_sizes_top[i + 1], - weight_attr=paddle.ParamAttr( - initializer=paddle.nn.initializer.Normal( - std=1.0 / math.sqrt(layer_sizes_top[i]) - ) - ), - ) - self.add_sublayer('linear_%d' % i, linear) - self._mlp_layers_top.append(linear) - act = paddle.nn.ReLU() - self.add_sublayer('act_%d' % i, act) - self._mlp_layers_top.append(act) - - def bottom_a_layer(self, sparse_inputs): - with paddle.base.device_guard(self.PART_A_DEVICE_FlAG): - sparse_embs = [] - for s_input in sparse_inputs: - emb = self.embedding(s_input) - emb = paddle.reshape(emb, shape=[-1, self.sparse_feature_dim]) - sparse_embs.append(emb) - - y = paddle.concat(x=sparse_embs, axis=1) - y = self._mlp_layers_a[0](y) - y = self._mlp_layers_a[1](y) - - y = self._mlp_layers_a[2](y) - with paddle.base.device_guard( - self.PART_A_JOINT_OP_DEVICE_FlAG - ): # joint point - bottom_a = self._mlp_layers_a[3](y) - - return bottom_a - - def bottom_b_layer(self, dense_inputs): - with paddle.base.device_guard(self.PART_B_DEVICE_FlAG): - y = self._mlp_layers_b[0](dense_inputs) - y = self._mlp_layers_b[1](y) - - y = self._mlp_layers_b[2](y) - bottom_b = self._mlp_layers_b[3](y) - - return bottom_b - - def interactive_layer(self, bottom_a, bottom_b): - with paddle.base.device_guard( - self.PART_B_JOINT_OP_DEVICE_FlAG - ): # joint point - interactive = paddle.add(bottom_a, bottom_b) - return interactive - - def top_layer(self, interactive, label_input): - with paddle.base.device_guard(self.PART_B_DEVICE_FlAG): - y = self._mlp_layers_top[0](interactive) - y_top = self._mlp_layers_top[1](y) - predict_2d = paddle.nn.functional.softmax(y_top) - ( - auc, - batch_auc, - [ - self.batch_stat_pos, - self.batch_stat_neg, - self.stat_pos, - self.stat_neg, - ], - ) = paddle.static.auc( - input=predict_2d, - label=label_input, - num_thresholds=2**12, - slide_steps=20, - ) - - cost = paddle.nn.functional.cross_entropy( - input=y_top, label=label_input - ) - avg_cost = paddle.mean(x=cost) - - return auc, avg_cost - - def forward(self, sparse_inputs, dense_inputs, label_input): - bottom_a = self.bottom_a_layer(sparse_inputs) - - bottom_b = self.bottom_b_layer(dense_inputs) - - interactive = self.interactive_layer(bottom_a, bottom_b) - - auc, avg_cost = self.top_layer(interactive, label_input) - - return auc, avg_cost - - -class StaticModel: - def __init__(self, config): - self.cost = None - self.infer_target_var = None - self.config = config - self._init_hyper_parameters() - self.sync_mode = config.get("runner.sync_mode") - - def _init_hyper_parameters(self): - self.is_distributed = False - self.distributed_embedding = False - - if self.config.get("hyper_parameters.distributed_embedding", 0) == 1: - self.distributed_embedding = True - - self.sparse_feature_number = self.config.get( - "hyper_parameters.sparse_feature_number" - ) - self.sparse_feature_dim = self.config.get( - "hyper_parameters.sparse_feature_dim" - ) - self.sparse_inputs_slots = self.config.get( - "hyper_parameters.sparse_inputs_slots" - ) - self.dense_input_dim = self.config.get( - "hyper_parameters.dense_input_dim" - ) - self.learning_rate = self.config.get( - "hyper_parameters.optimizer.learning_rate" - ) - self.fc_sizes = self.config.get("hyper_parameters.fc_sizes") - - def create_feeds(self, is_infer=False): - dense_input = paddle.static.data( - name="dense_input", - shape=[None, self.dense_input_dim], - dtype="float32", - ) - - sparse_input_ids = [ - paddle.static.data(name=str(i), shape=[None, 1], dtype="int64") - for i in range(1, self.sparse_inputs_slots) - ] - - label = paddle.static.data(name="label", shape=[None, 1], dtype="int64") - - feeds_list = [label] + sparse_input_ids + [dense_input] - return feeds_list - - def net(self, input, is_infer=False): - self.label_input = input[0] - self.sparse_inputs = input[1 : self.sparse_inputs_slots] - self.dense_input = input[-1] - sparse_number = self.sparse_inputs_slots - 1 - - dnn_model = DNNLayer( - self.sparse_feature_number, - self.sparse_feature_dim, - self.dense_input_dim, - sparse_number, - self.fc_sizes, - sync_mode=self.sync_mode, - ) - raw_predict_2d = dnn_model.forward(self.sparse_inputs, self.dense_input) - predict_2d = paddle.nn.functional.softmax(raw_predict_2d) - self.predict = predict_2d - ( - auc, - batch_auc, - [ - self.batch_stat_pos, - self.batch_stat_neg, - self.stat_pos, - self.stat_neg, - ], - ) = paddle.static.auc( - input=self.predict, - label=self.label_input, - num_thresholds=2**12, - slide_steps=20, - ) - self.inference_target_var = auc - if is_infer: - fetch_dict = {'auc': auc} - return fetch_dict - - cost = paddle.nn.functional.cross_entropy( - input=raw_predict_2d, label=self.label_input - ) - avg_cost = paddle.mean(x=cost) - self._cost = avg_cost - - fetch_dict = {'cost': avg_cost, 'auc': auc} - return fetch_dict - - def fl_net(self, input, is_infer=False): - self.label_input = input[0] - self.sparse_inputs = input[1 : self.sparse_inputs_slots] - self.dense_input = input[-1] - self.sparse_number = self.sparse_inputs_slots - 1 - - fl_dnn_model = FlDNNLayer( - self.sparse_feature_number, - self.sparse_feature_dim, - self.dense_input_dim, - self.sparse_number, - sync_mode=self.sync_mode, - ) - - auc, avg_cost = fl_dnn_model.forward( - self.sparse_inputs, self.dense_input, self.label_input - ) - fetch_dict = {'cost': avg_cost, 'auc': auc} - self._cost = avg_cost - return fetch_dict diff --git a/test/deprecated/ps/ps_dnn_trainer.py b/test/deprecated/ps/ps_dnn_trainer.py deleted file mode 100755 index ad8996efd87fc..0000000000000 --- a/test/deprecated/ps/ps_dnn_trainer.py +++ /dev/null @@ -1,598 +0,0 @@ -# Copyright (c) 2022 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 argparse -import ast -import copy -import os -import struct -import sys - -import numpy as np -import yaml -from ps_dnn_model import StaticModel - -import paddle -from paddle.distributed import fleet -from paddle.distributed.fleet.base import role_maker -from paddle.distributed.ps.utils.ps_program_builder import ( - debug_program, - logger, - new_pass, - ps_log_root_dir, -) - -__dir__ = os.path.dirname(os.path.abspath(__file__)) -sys.path.append(os.path.abspath(os.path.join(__dir__, '..'))) - - -def is_distributed_env(): - node_role = os.getenv("TRAINING_ROLE") - print(f"-- Role: {node_role} --") - if node_role is None: - return False - else: - return True - - -class YamlHelper: - def load_yaml(self, yaml_file, other_part=None): - part_list = ["runner", "hyper_parameters"] - if other_part: - part_list += other_part - running_config = self.get_all_inters_from_yaml(yaml_file, part_list) - running_config = self.workspace_adapter(running_config) - return running_config - - def print_yaml(self, config): - print(self.pretty_print_envs(config)) - - def parse_yaml(self, config): - vs = [int(i) for i in yaml.__version__.split(".")] - if vs[0] < 5: - use_full_loader = False - elif vs[0] > 5: - use_full_loader = True - else: - if vs[1] >= 1: - use_full_loader = True - else: - use_full_loader = False - - if os.path.isfile(config): - with open(config, 'r', encoding="utf-8") as rb: - if use_full_loader: - _config = yaml.load(rb.read(), Loader=yaml.FullLoader) - else: - _config = yaml.load(rb.read()) - return _config - else: - raise ValueError(f"config {config} can not be supported") - - def get_all_inters_from_yaml(self, file, filters): - _envs = self.parse_yaml(file) - all_flattens = {} - - def fatten_env_namespace(namespace_nests, local_envs): - for k, v in local_envs.items(): - if isinstance(v, dict): - nests = copy.deepcopy(namespace_nests) - nests.append(k) - fatten_env_namespace(nests, v) - else: - global_k = ".".join(namespace_nests + [k]) - all_flattens[global_k] = v - - fatten_env_namespace([], _envs) - ret = {} - for k, v in all_flattens.items(): - for f in filters: - if k.startswith(f): - ret[k] = v - return ret - - def workspace_adapter(self, config): - workspace = config.get("workspace") - for k, v in config.items(): - if isinstance(v, str) and "{workspace}" in v: - config[k] = v.replace("{workspace}", workspace) - return config - - def pretty_print_envs(self, envs, header=None): - spacing = 2 - max_k = 40 - max_v = 45 - - for k, v in envs.items(): - max_k = max(max_k, len(k)) - - h_format = " " + "|{{:>{}s}}{}{{:^{}s}}|\n".format( - max_k, " " * spacing, max_v - ) - l_format = " " + f"|{{:>{max_k}s}}{{}}{{:^{max_v}s}}|\n" - length = max_k + max_v + spacing - - border = " +" + "".join(["="] * length) + "+" - line = " +" + "".join(["-"] * length) + "+" - - draws = "" - draws += border + "\n" - - if header: - draws += h_format.format(header[0], header[1]) - else: - draws += h_format.format("Ps Benchmark Envs", "Value") - - draws += line + "\n" - - for k, v in sorted(envs.items()): - if isinstance(v, str) and len(v) >= max_v: - str_v = "... " + v[-41:] - else: - str_v = v - - draws += l_format.format(k, " " * spacing, str(str_v)) - - draws += border - - _str = f"\n{draws}\n" - return _str - - -def get_user_defined_strategy(config): - if not is_distributed_env(): - logger.warn( - "Not Find Distributed env, Change To local train mode. If you want train with fleet, please use [fleetrun] command." - ) - # return None - sync_mode = config.get("runner.sync_mode") - assert sync_mode in ["async", "sync", "geo", "heter", "gpubox"] - if sync_mode == "sync": - strategy = paddle.distributed.fleet.DistributedStrategy() - strategy.a_sync = False - elif sync_mode == "async": - strategy = paddle.distributed.fleet.DistributedStrategy() - strategy.a_sync = True - strategy.is_fl_ps_mode = ( - True if config.get("runner.is_fl_ps_mode") == 1 else False - ) - if strategy.is_fl_ps_mode: - strategy.pipeline = False - micro_num = 1 - strategy.pipeline_configs = { - "accumulate_steps": micro_num - } # num_microbatches - elif sync_mode == "geo": - strategy = paddle.distributed.fleet.DistributedStrategy() - strategy.a_sync = True - strategy.a_sync_configs = {"k_steps": config.get("runner.geo_step")} - elif sync_mode == "heter": - strategy = paddle.distributed.fleet.DistributedStrategy() - strategy.a_sync = True - strategy.a_sync_configs = {"heter_worker_device_guard": "gpu"} - strategy.pipeline = True - strategy.pipeline_configs = { - "accumulate_steps": config.get('runner.micro_num') - } - elif sync_mode == "gpubox": - print(f"sync_mode = {sync_mode}") - strategy = paddle.distributed.fleet.DistributedStrategy() - strategy.a_sync = True - strategy.a_sync_configs = {"use_ps_gpu": 1} - - strategy.trainer_desc_configs = { - "dump_fields_path": config.get("runner.dump_fields_path", ""), - "dump_fields": config.get("runner.dump_fields", []), - "dump_param": config.get("runner.dump_param", []), - "stat_var_names": config.get("stat_var_names", []), - "local_sparse": config.get("runner.local_sparse", []), - "remote_sparse": config.get("runner.remote_sparse", []), - } - print("strategy:", strategy.trainer_desc_configs) - - if config.get("runner.fs_client.uri") is not None: - strategy.fs_client_param = { - "uri": config.get("runner.fs_client.uri", ""), - "user": config.get("runner.fs_client.user", ""), - "passwd": config.get("runner.fs_client.passwd", ""), - "hadoop_bin": config.get("runner.fs_client.hadoop_bin", "hadoop"), - } - print("strategy:", strategy.fs_client_param) - - strategy.adam_d2sum = config.get("hyper_parameters.adam_d2sum", True) - table_config = {} - for x in config: - if x.startswith("table_parameters"): - table_name = x.split('.')[1] - if table_name not in table_config: - table_config[table_name] = {} - table_config[table_name][x] = config[x] - print("table_config:", table_config) - strategy.sparse_table_configs = table_config - print("strategy table config:", strategy.sparse_table_configs) - a_sync_configs = strategy.a_sync_configs - a_sync_configs["launch_barrier"] = False - # a_sync_configs["launch_barrier"] = True - strategy.a_sync_configs = a_sync_configs - print("launch_barrier: ", strategy.a_sync_configs["launch_barrier"]) - - return strategy - - -def get_distributed_strategy(user_defined_strategy): # pslib - from paddle.incubate.distributed.fleet.parameter_server.distribute_transpiler.distributed_strategy import ( - StrategyFactory, - ) - - k_steps = user_defined_strategy.a_sync_configs["k_steps"] - strategy = None - - if not user_defined_strategy.a_sync and k_steps == 0: - strategy = StrategyFactory.create_sync_strategy() - - if user_defined_strategy.a_sync and k_steps == 0: - strategy = StrategyFactory.create_async_strategy() - - if user_defined_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") - - return strategy - - -def get_model(config): - abs_dir = config['config_abs_dir'] - sys.path.append(abs_dir) - static_model = StaticModel(config) - return static_model - - -def parse_args(): - parser = argparse.ArgumentParser("PsTest train script") - parser.add_argument( - '-m', '--config_yaml', type=str, required=True, help='config file path' - ) - parser.add_argument( - '-bf16', - '--pure_bf16', - type=ast.literal_eval, - default=False, - help="whether use bf16", - ) - - parser.add_argument( - '--run_minimize', type=int, default=0, help="test single pass" - ) - parser.add_argument( - '--run_single_pass', type=int, default=0, help="test single pass" - ) - parser.add_argument( - '--run_the_one_ps', type=int, default=0, help="test the_one_ps" - ) - parser.add_argument( - '--debug_new_minimize', type=int, default=0, help="test single pass" - ) - parser.add_argument( - '--debug_new_pass', type=int, default=0, help="test single pass" - ) - parser.add_argument( - '--applied_pass_name', type=str, default="", help="test single pass" - ) - parser.add_argument( - '--debug_the_one_ps', type=int, default=0, help="test the_one_ps" - ) - - args = parser.parse_args() - args.abs_dir = os.path.dirname(os.path.abspath(args.config_yaml)) - yaml_helper = YamlHelper() - config = yaml_helper.load_yaml(args.config_yaml) - config["yaml_path"] = args.config_yaml - config["config_abs_dir"] = args.abs_dir - config["pure_bf16"] = args.pure_bf16 - config['run_minimize'] = args.run_minimize - config['run_single_pass'] = args.run_single_pass - config['run_the_one_ps'] = args.run_the_one_ps - config['debug_new_minimize'] = args.debug_new_minimize - config['debug_new_pass'] = args.debug_new_pass - config['applied_pass_name'] = args.applied_pass_name - config['debug_the_one_ps'] = args.debug_the_one_ps - yaml_helper.print_yaml(config) - return config - - -def bf16_to_fp32(val): - return np.float32(struct.unpack('>>>>>>>>> python process started") - os.environ["CPU_NUM"] = str(config.get("runner.thread_num")) - benchmark_main = DnnTrainer(config) - if config['run_single_pass'] == 1: - benchmark_main.run_single_pass() - elif config['run_minimize'] == 1: - benchmark_main.run_minimize() - elif config['run_the_one_ps'] == 1: - benchmark_main.run_the_one_ps()