trainer_desc.py 10.4 KB
Newer Older
1
#   Copyright (c) 2019 PaddlePaddle Authors. All Rights Reserved.
2 3 4 5 6 7 8 9 10 11 12 13
#
# 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.
X
xujiaqi01 已提交
14
"""Defination of trainers."""
15

16
import sys
17
import os
18
__all__ = ['TrainerDesc', 'MultiTrainer', 'DistMultiTrainer', 'PipelineTrainer']
19 20 21


class TrainerDesc(object):
22 23 24 25 26
    '''
    Set proto from python to c++.
    Can be initialized from train_desc.
    '''

27 28 29 30 31 32
    def __init__(self):
        '''
        self.proto_desc = data_feed_pb2.DataFeedDesc()
        with open(proto_file, 'r') as f:
            text_format.Parse(f.read(), self.proto_desc)
        '''
33 34
        # Workaround for relative import in protobuf under python3
        # TODO: should be fixed
35
        cur_path = os.path.dirname(__file__)
36 37 38 39
        if cur_path not in sys.path:
            sys.path.append(cur_path)
        if cur_path + "/proto" not in sys.path:
            sys.path.append(cur_path + "/proto")
40

41
        from proto import trainer_desc_pb2
42
        self.proto_desc = trainer_desc_pb2.TrainerDesc()
43 44 45
        import multiprocessing as mp
        # set default thread num == cpu count
        self.proto_desc.thread_num = mp.cpu_count()
46 47 48 49
        self._fleet_desc = None
        self._device_worker = None
        self._program = None
        self._infer = False
50

51
    def _set_fetch_var_and_info(self, fetch_vars, fetch_info, print_period):
52 53
        # convert fetch_info to list
        fetch_info = list(fetch_info)
D
dongdaxiang 已提交
54 55 56 57 58
        for i, v in enumerate(fetch_vars):
            self.proto_desc.fetch_config.fetch_var_names.extend([v.name])
            self.proto_desc.fetch_config.fetch_var_str_format.extend(
                [fetch_info[i]])
        self.proto_desc.fetch_config.print_period = print_period
D
dongdaxiang 已提交
59

60
    def _set_debug(self, debug):
61 62
        self.proto_desc.debug = debug

63
    def _set_thread(self, thread_num):
64 65
        self.proto_desc.thread_num = thread_num

66
    def _set_device_worker(self, device_worker):
67
        self._device_worker = device_worker
68

69
    def _set_infer(self, infer):
70
        self._infer = infer
71

72
    def _set_fleet_desc(self, fleet_desc):
73
        self._fleet_desc = fleet_desc
74

75
    def _gen_trainer_desc(self):
76 77
        pass

78
    def _set_program(self, program):
79
        self._program = program
80

81 82 83
    def _set_use_cvm(self, use_cvm=False):
        self.proto_desc.use_cvm = use_cvm

84 85 86
    def _set_no_cvm(self, no_cvm=False):
        self.proto_desc.no_cvm = no_cvm

87 88 89
    def _set_scale_datanorm(self, scale_datanorm=-1):
        self.proto_desc.scale_datanorm = scale_datanorm

90 91 92
    def _set_dump_slot(self, dump_slot):
        self.proto_desc.dump_slot = dump_slot

93 94 95
    def _set_mpi_rank(self, mpi_rank):
        self.proto_desc.mpi_rank = mpi_rank

96 97 98
    def _set_mpi_size(self, mpi_size):
        self.proto_desc.mpi_size = mpi_size

99 100 101 102 103 104 105
    def _set_dump_fields(self, dump_fields):
        for field in dump_fields:
            self.proto_desc.dump_fields.append(field)

    def _set_dump_fields_path(self, path):
        self.proto_desc.dump_fields_path = path

106 107 108
    def _set_dump_file_num(self, dump_file_num):
        self.proto_desc.dump_file_num = dump_file_num

109 110 111
    def _set_user_define_dump_filename(self, user_define_dump_filename):
        self.proto_desc.user_define_dump_filename = user_define_dump_filename

112 113 114
    def _set_dump_converter(self, converter):
        self.proto_desc.dump_converter = converter

115 116 117 118
    def _set_dump_param(self, dump_param):
        for param in dump_param:
            self.proto_desc.dump_param.append(param)

119 120 121
    def _set_thread_barrier(self, thread_barrier):
        self.proto_desc.thread_barrier = thread_barrier

122 123 124 125
    def _set_check_nan_var_names(self, check_nan_var_names):
        for var in check_nan_var_names:
            self.proto_desc.check_nan_var_names.append(var)

126 127 128 129
    def _set_loss_names(self, loss_names):
        for loss in loss_names:
            self.proto_desc.loss_names.append(loss)

130 131 132 133 134 135 136 137 138 139 140 141
    def _set_adjust_ins_weight(self, config_dict):
        self.proto_desc.adjust_ins_weight_config.need_adjust = \
                config_dict.get("need_adjust", False)
        self.proto_desc.adjust_ins_weight_config.nid_slot = \
                config_dict.get("nid_slot", "")
        self.proto_desc.adjust_ins_weight_config.nid_adjw_threshold = \
                config_dict.get("nid_adjw_threshold", 0.0)
        self.proto_desc.adjust_ins_weight_config.nid_adjw_ratio = \
                config_dict.get("nid_adjw_ratio", 0.0)
        self.proto_desc.adjust_ins_weight_config.ins_weight_slot = \
                config_dict.get("ins_weight_slot", "")

X
xujiaqi01 已提交
142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213
    def _set_copy_table_config(self, config_dict):
        config = self.proto_desc.copy_table_config
        config.need_copy = config_dict.get("need_copy", False)
        config.batch_num = config_dict.get("batch_num", 100)

        src_sparse_tables = config_dict.get("src_sparse_tables", [])
        if not isinstance(src_sparse_tables, list):
            src_sparse_tables = [src_sparse_tables]
        dest_sparse_tables = config_dict.get("dest_sparse_tables", [])
        if not isinstance(dest_sparse_tables, list):
            dest_sparse_tables = [dest_sparse_tables]
        if len(src_sparse_tables) != len(dest_sparse_tables):
            raise ValueError(
                "len(src_sparse_tables) != len(dest_sparse_tables)," \
                " %s vs %s" % (len(src_sparse_tables), \
                len(dest_sparse_tables)))
        for i in src_sparse_tables:
            config.src_sparse_tables.append(i)
        for i in dest_sparse_tables:
            config.dest_sparse_tables.append(i)

        src_dense_tables = config_dict.get("src_dense_tables", [])
        if not isinstance(src_dense_tables, list):
            src_dense_tables = [src_dense_tables]
        dest_dense_tables = config_dict.get("dest_dense_tables", [])
        if not isinstance(dest_dense_tables, list):
            dest_dense_tables = [dest_dense_tables]
        if len(src_dense_tables) != len(dest_dense_tables):
            raise ValueError(
                "len(src_dense_tables) != len(dest_dense_tables)," \
                " %s vs %s" % (len(src_dense_tables), \
                len(dest_dense_tables)))
        for i in src_dense_tables:
            config.src_dense_tables.append(i)
        for i in dest_dense_tables:
            config.dest_dense_tables.append(i)

        # user can also specify dense variables to copy,
        # instead of copy dense table
        src_var_list = config_dict.get("src_var_list", [])
        if not isinstance(src_var_list, list):
            src_var_list = [src_var_list]
        dest_var_list = config_dict.get("dest_var_list", [])
        if not isinstance(dest_var_list, list):
            dest_var_list = [dest_var_list]
        if len(src_var_list) != len(dest_var_list):
            raise ValueError(
                "len(src_var_list) != len(dest_var_list), %s vs" \
                " %s" % (len(src_var_list), len(dest_var_list)))
        for i in src_var_list:
            config.src_var_list.append(i)
        for i in dest_var_list:
            config.dest_var_list.append(i)

        dependency_map = config_dict.get("dependency_map", {})
        for key in dependency_map:
            m = config.table_denpendency_map.add()
            m.key = key
            values = dependency_map[key]
            if not isinstance(values, list):
                values = [values]
            if len(values) != 1:
                raise ValueError("dependency len %s != 1" % len(values))
            for value in values:
                m.values.append(value)
        config.dense_pull_after_copy = \
            config_dict.get("dense_pull_after_copy", True)
        config.enable_dependency = \
            config_dict.get("enable_dependency", False)
        config.sparse_copy_by_feasign = \
            config_dict.get("sparse_copy_by_feasign", True)

214
    def _desc(self):
215
        from google.protobuf import text_format
216
        return self.proto_desc.SerializeToString()
217

218
    def __str__(self):
219 220
        from google.protobuf import text_format
        return text_format.MessageToString(self.proto_desc)
221

222 223

class MultiTrainer(TrainerDesc):
224 225 226 227 228
    '''
    Implement of MultiTrainer.
    Can be init from TrainerDesc.
    '''

229
    def __init__(self):
230
        super(MultiTrainer, self).__init__()
231
        pass
232

233
    def _set_program(self, program):
234
        super(MultiTrainer, self)._set_program(program)
235
        self._program = program
236

237
    def _gen_trainer_desc(self):
238
        super(MultiTrainer, self)._gen_trainer_desc()
239
        self.proto_desc.class_name = "MultiTrainer"
D
fix bug  
dongdaxiang 已提交
240
        self._device_worker._set_infer(self._infer)
241
        self._device_worker._set_program(self._program)
242
        self._device_worker._gen_worker_desc(self.proto_desc)
243

244 245

class DistMultiTrainer(TrainerDesc):
X
xujiaqi01 已提交
246 247 248 249 250
    """
    Implement of DistMultiTrainer.
    It's for Distributed training.
    """

251
    def __init__(self):
252
        super(DistMultiTrainer, self).__init__()
253
        pass
254

255
    def _set_program(self, program):
256
        super(DistMultiTrainer, self)._set_program(program)
257
        self._program = program
258

259
    def _gen_trainer_desc(self):
260
        super(DistMultiTrainer, self)._gen_trainer_desc()
261
        self.proto_desc.class_name = "DistMultiTrainer"
262
        if self._program == None:
263
            raise RuntimeError("None Program")
D
fix bug  
dongdaxiang 已提交
264 265
        self._device_worker._set_infer(self._infer)
        self._device_worker._set_program(self._program)
266
        self._device_worker._gen_worker_desc(self.proto_desc)
267 268 269


class PipelineTrainer(TrainerDesc):
X
xujiaqi01 已提交
270 271 272 273 274
    """
    Implement of PipelineTrainer.
    It's for Pipeline.
    """

275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290
    def __init__(self):
        super(PipelineTrainer, self).__init__()
        pass

    def _set_program(self, program):
        super(PipelineTrainer, self)._set_program(program)
        self._program = program

    def _gen_trainer_desc(self):
        super(PipelineTrainer, self)._gen_trainer_desc()
        self.proto_desc.class_name = "PipelineTrainer"
        if self._program == None:
            raise RuntimeError("None Program")
        self._device_worker._set_infer(self._infer)
        self._device_worker._set_program(self._program)
        self._device_worker._gen_worker_desc(self.proto_desc)
反馈
建议
客服 返回
顶部