fleet_base.py 11.0 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18
#   Copyright (c) 2019 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

import abc

19
import paddle.fluid as fluid
T
tangwei12 已提交
20
from paddle.fluid.executor import Executor
21
from paddle.fluid.optimizer import SGD
M
MRXLT 已提交
22
from paddle.optimizer import SGD as SGD_v2
23

C
Chengmo 已提交
24
from paddle.fluid.incubate.fleet.base.mode import Mode
25
from paddle.distributed.fleet.base.role_maker import RoleMakerBase
Z
Zhen Wang 已提交
26
from paddle.fluid.contrib.mixed_precision.decorator import OptimizerWithMixedPrecision
C
Chengmo 已提交
27
from . import mode
28

C
Chengmo 已提交
29 30
__all__ = ['Fleet', 'DistributedOptimizer']
__all__ += mode.__all__
31 32 33 34 35 36 37 38 39 40 41 42 43 44 45


class Fleet(object):
    """
    Fleet is the base class, transpiler and pslib are implementation of Fleet.

    Args:
        mode(Mode): the implementation of Fleet's mode.

    Returns:
        None
    """
    __metaclass__ = abc.ABCMeta

    def __init__(self, mode):
T
tangwei12 已提交
46 47 48 49 50
        self._is_initialized = False
        self._mode = mode
        self._optimizer = None
        self._role_maker = None
        self._executor = None
51 52 53 54 55 56 57 58 59

    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.
        """
T
tangwei12 已提交
60
        return self._role_maker.is_first_worker()
61

T
tangwei12 已提交
62
    def worker_index(self):
63
        """
T
tangwei12 已提交
64
        Get current worker index.
65 66 67 68

        Returns:
            int: node id
        """
T
tangwei12 已提交
69
        return self._role_maker.worker_index()
70

T
tangwei12 已提交
71
    def worker_num(self):
72 73 74 75
        """
        Get current total worker number.

        Returns:
76
            int: worker numbers
77
        """
78
        return self._role_maker.worker_num()
79 80 81 82 83 84 85 86 87

    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.
        """
T
tangwei12 已提交
88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 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
        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()

    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()
133 134 135 136 137 138 139

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

        Returns:
            bool: True if this is a node of server,
140
                  False if not
141
        """
T
tangwei12 已提交
142
        return self._role_maker.is_server()
143

T
Thunderbrook 已提交
144 145 146 147 148 149 150 151 152 153
    def is_xpu(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_xpu()

154 155 156
    def split_files(self, files):
        """
        split files before distributed training,
157 158 159 160
        example 1: files is [a, b, c ,d, e]  and trainer_num = 2, then trainer
                   0 gets [a, b, c] and trainer 1 gets [d, e].
        example 2: files is [a, b], and trainer_num = 3, then trainer 0 gets
                   [a], trainer 1 gets [b],  trainer 2 gets []
161 162 163 164 165 166 167

        Args:
            files(list): file list need to be read.

        Returns:
            list: files belongs to this worker.
        """
168 169 170
        if not isinstance(files, list):
            raise TypeError("files should be a list of file need to be read.")

T
tangwei12 已提交
171
        trainer_id = self.worker_index()
T
tangwei12 已提交
172 173 174
        trainers = self.worker_num()

        remainder = len(files) % trainers
175
        blocksize = len(files) // trainers
T
tangwei12 已提交
176 177 178 179 180 181 182 183 184 185 186 187

        blocks = [blocksize] * trainers
        for i in range(remainder):
            blocks[i] += 1

        trainer_files = [[]] * trainers
        begin = 0
        for i in range(trainers):
            trainer_files[i] = files[begin:begin + blocks[i]]
            begin += blocks[i]

        return trainer_files[trainer_id]
188

189
    def init(self, role_maker=None):
190 191 192 193 194 195 196 197 198 199 200
        """
        should be called only once in user's python scripts,
        init() will initialize RoleMaker which is used for identifying
            current node's role, e.g. worker, server, etc.

        Args:
            role_maker(RoleMakerBase): subclass of RoleMakerBase.

        Returns:
            None
        """
201
        self._executor = Executor(fluid.CPUPlace())
202 203

        if role_maker and not isinstance(role_maker, RoleMakerBase):
204 205 206 207
            from paddle.fluid.incubate.fleet.base.role_maker import RoleMakerBase as RoleMakerBaseIncubate
            if role_maker and not isinstance(role_maker, RoleMakerBaseIncubate):
                raise TypeError(
                    "role_maker must be an instance of RoleMakerBase")
208

209
        self._role_maker = role_maker
210
        self._role_maker.generate_role()
T
tangwei12 已提交
211
        self._is_initialized = True
212

X
xujiaqi01 已提交
213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228
    def all_reduce_worker(self, input, output):
        """
        all reduce between workers, only support array of one dim.

        Args:
            input(list|numpy.array): array of one dim
            output(list|numpy.array): array of one dim
        """
        self._role_maker.all_reduce_worker(input, output)

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

229
    @abc.abstractmethod
T
tangwei12 已提交
230
    def init_worker(self):
231 232 233
        pass

    @abc.abstractmethod
234
    def init_server(self, model_dir=None, **kwargs):
235 236 237
        pass

    @abc.abstractmethod
238
    def run_server(self):
239 240 241 242 243 244 245 246 247 248 249 250
        pass

    @abc.abstractmethod
    def stop_worker(self):
        pass

    @abc.abstractmethod
    def distributed_optimizer(self, optimizer, strategy=None):
        pass

    @abc.abstractmethod
    def save_inference_model(self,
251
                             executor,
252 253 254 255 256 257 258 259
                             dirname,
                             feeded_var_names,
                             target_vars,
                             main_program=None,
                             export_for_deployment=True):
        pass

    @abc.abstractmethod
260
    def save_persistables(self, executor, dirname, main_program=None):
261 262 263 264 265 266 267 268 269 270 271 272 273 274 275
        pass


class DistributedOptimizer(object):
    """
    DistributedOptimizer is a wrapper for paddle.fluid.optimizer
    A user should pass a paddle.fluid.optimizer to DistributedOptimizer
    minimize() function is implemented.
    DistributedOptimizer is the starting point for a user who wants to
    run distributed training. The optimized information will be stored in
    Fleet() instance who holds the global information about current distributed
    training.

    Args:
        optimizer(Optimizer): subclass of Optimizer.
T
tangwei12 已提交
276
        strategy(any): the user define config for Optimizer.
277 278 279 280 281 282 283 284

    Returns:
        None

    """
    __metaclass__ = abc.ABCMeta

    def __init__(self, optimizer, strategy=None):
G
gongweibao 已提交
285
        if not isinstance(optimizer, SGD.__bases__) \
M
MRXLT 已提交
286 287
                and not isinstance(optimizer, OptimizerWithMixedPrecision) \
                and not isinstance(optimizer, SGD_v2.__base__):
288
            raise TypeError("optimizer must be an instance of Optimizer")
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 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346

        self._optimizer = optimizer
        self._strategy = strategy

    @abc.abstractmethod
    def backward(self,
                 loss,
                 startup_program=None,
                 parameter_list=None,
                 no_grad_set=None,
                 callbacks=None):
        """
        First part of `minimize`, do auto-diff to append backward ops for
        the current program.

        Args:
            loss (Variable): loss variable to run optimizations.
            startup_program (Program): startup_program for initializing parameters
                in `parameter_list`.
            parameter_list (list): list of Variables to update.
            no_grad_set (set|None): set of Variables should be ignored.
            callbacks (list|None): list of callables to run when appending backward
                operator for one parameter.

        Return:
            list: list of (param, grad) pair, grad is the output of backward.

        Examples:
            See examples in `apply_gradients`.
        """
        pass

    @abc.abstractmethod
    def apply_gradients(self, params_grads):
        """
        Second part of `minimize`, appending optimization operators for
        given `params_grads` pairs.

        Args:
            params_grads (list): list of (param, grad) pair to do optimization.

        Returns:
            list: A list of operators appended to the current program.

        Examples:
            .. code-block:: python

                loss = network()
                optimizer = fluid.optimizer.SGD(learning_rate=0.1)
                params_grads = optimizer.backward(loss)
                # you may append operations for params_grads here
                # ...
                optimizer.apply_gradients(params_grads)
        """
        pass

    @abc.abstractmethod
    def minimize(self,
T
tangwei12 已提交
347 348 349
                 losses,
                 scopes=None,
                 startup_programs=None,
350 351 352 353 354 355 356 357 358
                 parameter_list=None,
                 no_grad_set=None):
        """
        Add operations to minimize `loss` by updating `parameter_list`.

        This method combines interface `backward()` and
        `apply_gradients()` into one.

        Args:
T
tangwei12 已提交
359 360 361
            losses (Variable|Variable List): loss variable to run optimizations.
            scopes (Scope| Scope List): scope instance.
            startup_programs (Program|Program List): startup_program for initializing parameters
362 363 364 365 366 367 368 369 370
                in `parameter_list`.
            parameter_list (list): list of Variables to update.
            no_grad_set (set|None): set of Variables should be ignored.

        Returns:
            tuple: (optimize_ops, params_grads) which are, list of operators appended;
            and list of (param, grad) Variables pair for optimization.
        """
        pass