fleet_base.py 14.9 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
#   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.

from __future__ import print_function
16
import paddle
17
from .role_maker import UserDefinedRoleMaker, PaddleCloudRoleMaker, RoleMakerBase
18
from .strategy_compiler import StrategyCompiler
19
from .distributed_strategy import DistributedStrategy
20 21 22
from .meta_optimizer_factory import MetaOptimizerFactory
from .runtime_factory import RuntimeFactory
from .util_factory import UtilFactory
23
from paddle.fluid.wrapped_decorator import wrap_decorator
24

25
#__all__ = ['Fleet']
26 27


28 29 30 31 32 33 34 35 36 37 38 39 40 41 42
def _inited_runtime_handler_(func):
    def __impl__(*args, **kwargs):
        cls = args[0]

        if cls._runtime_handle is None:
            raise ValueError("Fleet can not find suitable runtime handler")

        return func(*args, **kwargs)

    return __impl__


inited_runtime_handler = wrap_decorator(_inited_runtime_handler_)


43 44 45 46 47 48 49 50 51 52 53 54
class Fleet(object):
    """
    Unified API for distributed training of PaddlePaddle
    Please reference the https://github.com/PaddlePaddle/Fleet for details


    Returns:
        Fleet: A Fleet instance

    Examples:
        .. code-block:: python

55 56
            import paddle.distributed.fleet as fleet
            role = fleet.role_maker.PaddleCloudRoleMaker(is_collective=True)
57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77
            fleet.init(role)
            strategy = fleet.DistributedStrategy()
            optimizer = paddle.optimizer.SGD(learning_rate=0.001)
            optimizer = fleet.distributed_optimizer(optimizer, strategy=strategy)
            if fleet.is_first_worker():
                print("this is first worker")
            print("current node index: {}".format(fleet.worker_index()))
            print("total number of worker num: {}".format(fleet.worker_num()))
            if fleet.is_worker():
                print("this is worker")
            print("worker endpoints: {}".format(fleet.worker_endpoints(to_string=True)))
            print("server num: {}".format(fleet.server_num()))
            print("server endpoints: {}".format(fleet.server_endpoints(to_string=True)))
            if fleet.is_server():
                print("this is server")
            fleet.stop_worker()
    """

    def __init__(self):
        self._runtime_handle = None
        self._util = None
78 79
        self._role_maker = None
        self._is_collective = False
80

81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104
    def init(self, role_maker=None, is_collective=False):
        """
        Initialize role_maker in Fleet.

        This function is responsible for the distributed architecture 
        what you want to run your code behind,such as Transpiler,
        Collective in PaddleCloudRoleMaker or UserDefinedRoleMaker 
        
        """
        if isinstance(role_maker, RoleMakerBase):
            self._role_maker = role_maker
        elif role_maker == None:
            if isinstance(is_collective, bool):
                self._is_collective = is_collective
                self._role_maker = PaddleCloudRoleMaker(
                    is_collective=self._is_collective)
            else:
                raise ValueError(
                    "Something wrong occurred, please check whether is_collective is bool value"
                )
        else:
            raise ValueError(
                "Something wrong occurred, please check whether rolemaker is instance of RoleMakerBase"
            )
105
        self.strategy_compiler = StrategyCompiler()
106
        return None
107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 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 214 215 216 217 218 219 220 221 222 223 224 225 226

    def is_first_worker(self):
        """
        Check whether the node is the first instance of worker.

        Returns:
            bool: True if this is the first node of worker,
                  False if not.
        
        """
        return self._role_maker.is_first_worker()

    def worker_index(self):
        """
        Get current worker index.

        Returns:
            int: node id
        """
        return self._role_maker.worker_index()

    def worker_num(self):
        """
        Get current total worker number.

        Returns:
            int: worker numbers
        """
        return self._role_maker.worker_num()

    def is_worker(self):
        """
        Check whether the node is an instance of worker.

        Returns:
            bool: True if this is a node of worker,
                  False if not.
        """
        return self._role_maker.is_worker()

    def worker_endpoints(self, to_string=False):
        """
        Get current server endpoints, such as ["127.0.0.1:1001", "127.0.0.1:1002"].

        Returns:
            list/string: server endpoints
        """
        '''
        if to_string:
            return ",".join(self._role_maker.get_trainer_endpoints())
        else:
            return self._role_maker.get_trainer_endpoints()
        '''
        return ["127.0.0.1:1001", "127.0.0.1:1002"]

    def server_num(self):
        """
        Get current total worker number.

        Returns:
            int: server number
        """
        return len(self._role_maker.get_pserver_endpoints())

    def server_index(self):
        """
        Get current server index.

        Returns:
            int: node id
        """
        return self._role_maker.server_index()

    def server_endpoints(self, to_string=False):
        """
        Get current server endpoints, such as ["127.0.0.1:1001", "127.0.0.1:1002"].

        Returns:
            list/string: server endpoints
        """
        '''
        if to_string:
            return ",".join(self._role_maker.get_pserver_endpoints())
        else:
            return self._role_maker.get_pserver_endpoints()
        '''
        return ["127.0.0.1:1001", "127.0.0.1:1002"]

    def is_server(self):
        """
        Check whether the node is an instance of server.

        Returns:
            bool: True if this is a node of server,
                  False if not.
        """
        return self._role_maker.is_server()

    @property
    def util(self):
        """
        Utility functions that can be used under certain runtime
        return util
        """
        return self._util

    @util.setter
    def util(self, util):
        """
        Set Utility functions for userd-defined runtime
        set util
        """
        self._util = util

    def barrier_worker(self):
        """
        barrier between workers
        """
        self._role_maker.barrier_worker()

227
    @inited_runtime_handler
228 229 230 231 232 233
    def init_worker(self):
        """
        init worker
        """
        self._runtime_handle._init_worker()

234
    @inited_runtime_handler
235
    def init_server(self, *args, **kwargs):
236 237 238
        """
        init server
        """
239
        self._runtime_handle._init_server(*args, **kwargs)
240

241
    @inited_runtime_handler
242 243 244 245 246 247
    def run_server(self):
        """
        run server
        """
        self._runtime_handle._run_server()

248
    @inited_runtime_handler
249 250 251 252 253 254
    def stop_worker(self):
        """
        stop worker
        """
        self._runtime_handle._stop_worker()

255 256 257 258 259 260 261 262 263 264 265 266 267 268
    def save_inference_model(self,
                             executor,
                             dirname,
                             feeded_var_names,
                             target_vars,
                             main_program=None,
                             export_for_deployment=True):
        self._runtime_handle._save_inference_model(
            executor, dirname, feeded_var_names, target_vars, main_program,
            export_for_deployment)

    def save_persistables(self, executor, dirname, main_program=None):
        self._runtime_handle._save_persistables(executor, dirname, main_program)

269
    def distributed_optimizer(self, optimizer, strategy=None):
270 271 272 273 274 275 276
        """
        distirbuted_optimizer
        Returns:
            Fleet instance with minimize interface like optimizers

        Examples:
            .. code-block:: python
277 278
            import paddle.distributed.fleet as fleet
            role = fleet.role_maker.PaddleCloudRoleMaker(is_collective=True)
279 280 281 282 283 284
            fleet.init(role)
            strategy = fleet.DistributedStrategy()
            optimizer = paddle.optimizer.SGD(learning_rate=0.001)
            optimizer = fleet.distributed_optimizer(optimizer, strategy=strategy)
        """
        self.user_defined_optimizer = optimizer
285 286
        if strategy == None:
            strategy = DistributedStrategy()
287
        self.user_defined_strategy = strategy
D
Dong Daxiang 已提交
288
        self.valid_strategy = None
289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319
        return self

    def minimize(self,
                 loss,
                 startup_program=None,
                 parameter_list=None,
                 no_grad_set=None):
        """
        Add distributed operations to minimize ``loss`` by updating ``parameter_list``.

        Args:
            loss (Variable): A ``Variable`` containing the value to minimize.
            startup_program (Program, optional): :ref:`api_fluid_Program` for
                initializing parameters in ``parameter_list``. The default value
                is None, at this time :ref:`api_fluid_default_startup_program` will be used.
            parameter_list (Iterable, optional): Iterable of ``Variable`` or ``Variable.name`` to update
                to minimize ``loss``. The default value is None, at this time all parameters
                will be updated.
            no_grad_set (set, optional): Set of ``Variable``  or ``Variable.name`` that don't need
                to be updated. The default value is None.

        Returns:
            tuple: tuple (optimize_ops, params_grads), A list of operators appended
            by minimize and a list of (param, grad) variable pairs, param is
            ``Parameter``, grad is the gradient value corresponding to the parameter.
            The returned tuple can be passed to ``fetch_list`` in ``Executor.run()`` to 
            indicate program pruning. If so, the program will be pruned by ``feed`` and 
            ``fetch_list`` before run, see details in ``Executor``.

        Examples:
            import paddle
320
            import paddle.distributed.fleet as fleet
321 322 323 324 325 326 327

            fc_1 = paddle.layers.fc(input=input_x, size=hid_dim, act='tanh')
            fc_2 = paddlen.layers.fc(input=fc_1, size=hid_dim, act='tanh')
            prediction = paddle.layers.fc(input=[fc_2], size=label_dim, act='softmax')
            cost = paddle.layers.cross_entropy(input=prediction, label=input_y)
            avg_cost = paddle.layers.mean(x=cost)

328
            role = fleet.role_maker.PaddleCloudRoleMaker(is_collective=True)
329 330 331 332 333 334 335 336 337
            fleet.init(role)
            strategy = fleet.DistributedStrategy()
            optimizer = paddle.optimizer.SGD(learning_rate=0.001)
            optimizer = fleet.distributed_optimizer(optimizer, strategy=strategy)
            optimizer.minimize(avg_cost)

            # for more examples, please reference https://github.com/PaddlePaddle/Fleet

        """
338
        context = {}
339 340
        # cache original feed forward program
        self.origin_main_program = loss.block.program
341 342
        context["origin_main_program"] = self.origin_main_program
        context["loss"] = loss
343 344
        if startup_program == None:
            self.origin_startup_program = \
345 346
                paddle.static.default_startup_program().clone(for_test=False)
            startup_program = paddle.static.default_startup_program()
347 348 349
        else:
            self.origin_startup_program = \
                startup_program.clone(for_test=False)
350

351 352
        context["origin_startup_program"] = startup_program
        context["role_maker"] = self._role_maker
353 354 355 356 357

        # compile time
        distributed_optimizer_list = \
            MetaOptimizerFactory()._get_valid_meta_optimizers(
                self.user_defined_optimizer)
D
Dong Daxiang 已提交
358

359 360
        valid_optimizer_list = []
        valid_graph_optimizer_list = []
D
Dong Daxiang 已提交
361
        can_not_apply_optimizer_list = []
362 363 364 365 366 367 368
        # recall meta optimizers for ranking
        for opt in distributed_optimizer_list:
            opt._set_basic_info(loss, self._role_maker,
                                self.user_defined_optimizer,
                                self.user_defined_strategy)
            if opt._can_apply() and not opt._is_graph_out():
                valid_optimizer_list.append(opt)
D
Dong Daxiang 已提交
369
            elif opt._can_apply() and opt._is_graph_out():
370
                valid_graph_optimizer_list.append(opt)
D
Dong Daxiang 已提交
371 372
            else:
                can_not_apply_optimizer_list.append(opt)
373
        # combine recalled meta optimizers to be a valid meta optimizer
D
Dong Daxiang 已提交
374
        meta_optimizer, graph_optimizer = \
375 376 377 378
                self.strategy_compiler.generate_optimizer(
                    loss, self._role_maker, self.user_defined_optimizer,
                    self.user_defined_strategy, valid_optimizer_list,
                    valid_graph_optimizer_list)
D
Dong Daxiang 已提交
379

D
Dong Daxiang 已提交
380 381
        valid_strategy = self.strategy_compiler._get_valid_strategy(
            self.user_defined_strategy, can_not_apply_optimizer_list)
382 383 384

        context["valid_strategy"] = valid_strategy

D
Dong Daxiang 已提交
385
        self.valid_strategy = valid_strategy
386
        self.valid_strategy._enable_env()
D
Dong Daxiang 已提交
387

388 389
        optimize_ops = []
        params_grads = []
390

391 392 393 394 395 396
        if meta_optimizer:
            optimize_ops, params_grads = meta_optimizer.minimize(
                loss,
                startup_program=startup_program,
                parameter_list=parameter_list,
                no_grad_set=no_grad_set)
397

398
            default_program = paddle.static.default_main_program()
399 400 401 402

            if id(default_program) != id(loss.block.program):
                paddle.fluid.framework.switch_main_program(loss.block.program)

403 404 405 406 407 408
        else:
            optimize_ops, params_grads = self.user_defined_optimizer.minimize(
                loss,
                startup_program=startup_program,
                parameter_list=parameter_list,
                no_grad_set=no_grad_set)
409

410 411
        context["program_optimize_ops"] = optimize_ops
        context["program_params_grads"] = params_grads
412

413
        if graph_optimizer:
D
Dong Daxiang 已提交
414
            optimize_ops, params_grads = graph_optimizer.minimize(
415 416 417 418 419 420 421 422
                loss,
                startup_program=startup_program,
                parameter_list=parameter_list,
                no_grad_set=no_grad_set)
            # since we do not encourage users to use graph operations
            # if a graph optimizer takes effect, mostly
            # optimizers_ops and params_grads are None
            # i.e. users can not modify current computation graph anymore
423 424 425
            context["graph_optimize_ops"] = optimize_ops
            context["graph_optimize_grads"] = params_grads

426
        if self._runtime_handle is None:
427
            self._runtime_handle = RuntimeFactory()._create_runtime(context)
428 429

        if self._util is None:
430
            self._util = UtilFactory()._create_util(context)
431 432

        return optimize_ops, params_grads