device_worker.py 9.1 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.
14
"""Defination of device workers."""
15

H
hutuxian 已提交
16
__all__ = ['DeviceWorker', 'Hogwild', 'DownpourSGD', 'Section']
17

18 19

class DeviceWorker(object):
X
xjqbest 已提交
20
    """
21
    DeviceWorker is an abstract class, which generates worker desc.
22 23
    This class is an inner class that we do computation logics within
    the implementation. For example, execution of a program or a graph.
X
xjqbest 已提交
24
    """
25

26
    def __init__(self):
27
        """Init."""
D
dongdaxiang 已提交
28 29
        self._program = None
        self._infer = None
30

31 32 33 34 35 36 37
    def _set_infer(self, infer=False):
        """
        set inference flag for current device worker
        
        Args:
            infer(bool): whether to do inference
        """
D
dongdaxiang 已提交
38
        self._infer = infer
D
dongdaxiang 已提交
39

40
    def _set_fleet_desc(self, fleet_desc):
X
xjqbest 已提交
41 42 43 44 45 46
        """
        Set fleet desc.

        Args:
            fleet_desc(PSParameter): pslib.PSParameter object
        """
D
dongdaxiang 已提交
47
        self._fleet_desc = fleet_desc
D
dongdaxiang 已提交
48

49
    def _set_program(self, program):
X
xjqbest 已提交
50 51 52 53 54 55
        """
        Set program.

        Args:
            program(Program): a Program object
        """
D
dongdaxiang 已提交
56
        self._program = program
57

58
    def _gen_worker_desc(self, trainer_desc):
X
xjqbest 已提交
59 60 61 62 63 64 65 66 67
        """
        Generator worker desc.

        Args:
            trainer_desc(TrainerDesc): a TrainerDesc object
        """
        raise NotImplementedError(
            "DeviceWorker does not implement gen_worker_desc, "
            "please use Hogwild or DownpourSGD, etc.")
68 69 70


class Hogwild(DeviceWorker):
X
xjqbest 已提交
71 72 73 74
    """
    Hogwild is a kind of SGD algorithm.

    """
75

76
    def __init__(self):
77
        """Init."""
78 79
        super(Hogwild, self).__init__()

80
    def _gen_worker_desc(self, trainer_desc):
X
xjqbest 已提交
81 82 83 84 85 86
        """
        Generator worker desc, which device worker is HogwildWorker.

        Args:
            trainer_desc(TrainerDesc): a TrainerDesc object
        """
87
        trainer_desc.device_worker_name = "HogwildWorker"
D
dongdaxiang 已提交
88
        if self._infer:
89 90
            # just ignore feed op for inference model
            trainer_desc.hogwild_param.skip_ops.extend(["feed"])
91 92


D
dongdaxiang 已提交
93
class DownpourSGD(DeviceWorker):
X
xjqbest 已提交
94 95 96
    """
    DownpourSGD is a kind of distributed SGD algorithm.
    """
97

98
    def __init__(self):
X
xjqbest 已提交
99 100
        """
        Init.
101
        initialize downpourSGD device worker
X
xjqbest 已提交
102
        """
D
dongdaxiang 已提交
103
        super(DownpourSGD, self).__init__()
104

105
    def _gen_worker_desc(self, trainer_desc):
X
xjqbest 已提交
106 107 108 109 110 111
        """
        Generator worker desc, which device worker is DownpourWorker.

        Args:
            trainer_desc(TrainerDesc): a TrainerDesc object
        """
X
fix bug  
xjqbest 已提交
112
        dense_table_set = set()
D
dongdaxiang 已提交
113 114
        program_id = str(id(self._program))
        if self._program == None:
D
dongdaxiang 已提交
115
            print("program of current device worker is not configured")
116
            exit(-1)
D
dongdaxiang 已提交
117
        opt_info = self._program._fleet_opt
D
dongdaxiang 已提交
118
        program_configs = opt_info["program_configs"]
119
        downpour = trainer_desc.downpour_param
D
dongdaxiang 已提交
120

D
dongdaxiang 已提交
121 122
        for pid in program_configs:
            if pid == program_id:
D
dongdaxiang 已提交
123 124 125 126 127 128
                pc = downpour.program_config.add()
                pc.program_id = program_id
                for i in program_configs[program_id]["push_sparse"]:
                    pc.push_sparse_table_id.extend([i])
                for i in program_configs[program_id]["push_dense"]:
                    pc.push_dense_table_id.extend([i])
X
xjqbest 已提交
129
                    dense_table_set.add(i)
D
dongdaxiang 已提交
130 131 132 133
                for i in program_configs[program_id]["pull_sparse"]:
                    pc.pull_sparse_table_id.extend([i])
                for i in program_configs[program_id]["pull_dense"]:
                    pc.pull_dense_table_id.extend([i])
X
fix bug  
xjqbest 已提交
134
                    dense_table_set.add(i)
D
dongdaxiang 已提交
135
                break
136

137 138 139
        trainer_desc.device_worker_name = "DownpourWorker"
        pull_thread = trainer_desc.pull_dense_param
        pull_thread.device_num = trainer_desc.thread_num
140 141 142 143 144 145 146 147
        if opt_info.get("program_id_to_worker") is None:
            raise ValueError("opt_info must have program_id_to_worker")
        prog_id_to_worker = opt_info["program_id_to_worker"]
        if prog_id_to_worker.get(program_id) is None:
            raise ValueError("%s not found in program_id_to_worker" %
                             program_id)
        worker = opt_info["program_id_to_worker"][program_id]
        for i in worker.get_desc().dense_table:
148 149
            if i.table_id in dense_table_set:
                dense_table = pull_thread.dense_table.add()
150
                dense_table.dense_value_name.extend(i.dense_variable_name)
151 152
                dense_table.table_id = \
                    i.table_id
153
        sparse_len = len(worker.get_desc().sparse_table)
154 155
        for i in range(sparse_len):
            sparse_table = downpour.sparse_table.add()
156 157 158 159 160 161 162
            sparse_table.table_id = worker.get_desc().sparse_table[i].table_id
            sparse_table.sparse_key_name.extend(worker.get_desc().sparse_table[
                i].slot_key)
            sparse_table.sparse_value_name.extend(worker.get_desc()
                                                  .sparse_table[i].slot_value)
            sparse_table.sparse_grad_name.extend(worker.get_desc().sparse_table[
                i].slot_gradient)
163 164
            if opt_info["use_cvm"] or "no_cvm" in opt_info and opt_info[
                    "no_cvm"] == True:
165 166 167 168 169 170 171 172 173 174 175
                sparse_table.emb_dim = \
                    self._fleet_desc.server_param.downpour_server_param.downpour_table_param[
                    i].accessor.fea_dim
                sparse_table.fea_dim = sparse_table.emb_dim
            else:
                sparse_table.emb_dim = \
                    self._fleet_desc.server_param.downpour_server_param.downpour_table_param[
                    i].accessor.fea_dim - 2
                sparse_table.fea_dim = sparse_table.emb_dim + 2
            # TODO(guru4elephant): hard code here, need to improve
            sparse_table.label_var_name = "click"
176 177 178
        if opt_info["stat_var_names"]:
            for i in opt_info["stat_var_names"]:
                downpour.stat_var_names.extend([i])
179

180
        for i in worker.get_desc().dense_table:
X
fix bug  
xjqbest 已提交
181 182 183
            if i.table_id in dense_table_set:
                dense_table = downpour.dense_table.add()
                dense_table.table_id = i.table_id
184
                dense_table.dense_value_name.extend(i.dense_variable_name)
X
fix bug  
xjqbest 已提交
185 186
                dense_table.dense_grad_name.extend(
                    i.dense_gradient_variable_name)
X
xujiaqi01 已提交
187
        downpour.skip_ops.extend(worker.get_desc().skip_op)
D
dongdaxiang 已提交
188
        if self._infer:
189 190
            downpour.push_dense = False
            downpour.push_sparse = False
X
fix bug  
xjqbest 已提交
191

192

H
hutuxian 已提交
193
class Section(DeviceWorker):
194
    """SectionWorker."""
H
hutuxian 已提交
195 196

    def __init__(self):
197
        """Init."""
H
hutuxian 已提交
198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240
        super(Section, self).__init__()

    def _gen_worker_desc(self, trainer_desc):
        """
        Generator worker desc, which device worker is SectionWorker.
        Args:
            trainer_desc(TrainerDesc): a TrainerDesc object
        """
        from google.protobuf import text_format
        from . import core
        trainer_desc.device_worker_name = "SectionWorker"
        pipeline_opt = self._program._pipeline_opt
        section_param = trainer_desc.section_param
        section_param.queue_size = pipeline_opt["queue_size"]
        section_param.sync_steps = pipeline_opt["sync_steps"]
        section_param.start_cpu_core_id = pipeline_opt["start_cpu_core_id"]
        for e in pipeline_opt["param_need_sync"]:
            section_param.param_need_sync.append(e)
        for i, program in enumerate(pipeline_opt["section_program_list"]):
            cfg = section_param.section_config.add()
            cfg.program_desc.ParseFromString(program["program"]._get_desc()
                                             .serialize_to_string())
            # TODO: why does not work
            #cfg.program_desc.CopyFrom(program.program._get_desc())
            place = pipeline_opt["place_list"][i]
            if isinstance(place, core.CPUPlace):
                cfg.place = cfg.CPUPlace
            elif isinstance(place, core.CUDAPlace):
                cfg.place = cfg.CUDAPlace
            elif isinstance(place, core.CUDAPinnedPlace):
                cfg.place = cfg.CUDAPinnedPlace
            else:
                raise NotImplementedError(
                    "SectionWorker only supports CPUPlace, CUDAPlace and CUDAPinnedPlace now."
                )

            cfg.concurrency = pipeline_opt["concurrency_list"][i]
            for var in program["input_set"]:
                cfg.section_in_var_names.append(var)
            for var in program["output_set"]:
                cfg.section_out_var_names.append(var)


241
class DeviceWorkerFactory(object):
242
    def _create_device_worker(self, worker_type):
243 244
        classname = worker_type.capitalize()
        return globals()[classname]()