parallel_executor.py 15.7 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14
#   Copyright (c) 2018 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.

15 16 17 18
from __future__ import print_function
from . import core
from . import framework
from . import executor
19
from . import compiler
Y
Yu Yang 已提交
20
import sys
21

X
Xin Pan 已提交
22
__all__ = ['ParallelExecutor']
Y
yuyang18 已提交
23 24

ExecutionStrategy = core.ParallelExecutor.ExecutionStrategy
Y
yuyang18 已提交
25
BuildStrategy = core.ParallelExecutor.BuildStrategy
26 27 28


class ParallelExecutor(object):
C
chengduoZH 已提交
29
    """
C
chengduo 已提交
30 31 32 33 34 35 36 37 38
    ParallelExecutor is designed for data parallelism, which focuses on distributing
    the data across different nodes and every node operates on the data in parallel.
    If you use ParallelExecutor to run the current program on GPU, the node means GPU
    device, and ParallelExecutor will get the available GPU device automatically on
    the current machine. If you use ParallelExecutor to run the current program on CPU,
    the node means the CPU device, and you can specify the CPU device number by adding
    'CPU_NUM' environment variable, for example 'CPU_NUM=4', if the environment variable
    is not found, ParallelExecutor will call `multiprocessing.cpu_count` to get the number
    of CPUs in the system.
C
chengduoZH 已提交
39

40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86
    Examples:
        .. code-block:: python

          import paddle.fluid as fluid
          import numpy
          import os

          use_cuda = True
          place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()

          # NOTE: If you use CPU to run the program, you need
          # to specify the CPU_NUM, otherwise, fluid will use
          # all the number of the logic core as the CPU_NUM,
          # in that case, the batch size of the input should be
          # greater than CPU_NUM, if not, the process will be
          # failed by an exception.
          if not use_cuda:
              os.environ['CPU_NUM'] = str(2)

          exe = fluid.Executor(place)

          train_program = fluid.Program()
          startup_program = fluid.Program()
          with fluid.program_guard(train_program, startup_program):
              data = fluid.layers.data(name='X', shape=[1], dtype='float32')
              hidden = fluid.layers.fc(input=data, size=10)
              loss = fluid.layers.mean(hidden)
              test_program = fluid.default_main_program().clone(for_test=True)
              fluid.optimizer.SGD(learning_rate=0.01).minimize(loss)

          startup_program.random_seed=1
          exe.run(startup_program)

          train_exe = fluid.ParallelExecutor(use_cuda=use_cuda,
                                             main_program=train_program,
                                             loss_name=loss.name)
          test_exe = fluid.ParallelExecutor(use_cuda=use_cuda,
                                            main_program=test_program,
                                            share_vars_from=train_exe)

          x = numpy.random.random(size=(10, 1)).astype('float32')
          loss_data, = train_exe.run(feed={"X": x},
                                     fetch_list=[loss.name])

          loss_data, = test_exe.run(feed={"X": x},
                                    fetch_list=[loss.name])

C
chengduoZH 已提交
87 88 89 90 91
    Args:
        use_cuda (bool): Whether to use CUDA or not.
        loss_name (str): The loss name must set in training. Default None.
        main_program (Program): The program that need to run, if not provided,
            then default_main_program will be used. Default None.
C
chengduo 已提交
92
        share_vars_from(ParallelExecutor): If provide, it will share variables
C
chengduoZH 已提交
93
            from the specified ParallelExecutor. Default None.
C
chengduo 已提交
94 95 96 97 98 99 100 101 102
        exec_strategy(ExecutionStrategy): exec_strategy is used to control how to run
            the program in ParallelExecutor, for example how many threads are used to
            execute the program, how many iterations to clean up the temp variables
            which is generated during execution. For more information, please refer
            to fluid.ExecutionStrategy. Default None.
        build_strategy(BuildStrategy): build_strategy is used to control how to
            build the SSA Graph in ParallelExecutor by setting the property,
            for example reduce_strategy, gradient_scale_strategy. For more information,
            please refer to fluid.BuildStrategy. Default None.
C
chengduoZH 已提交
103 104 105
        num_trainers(int): If greater than 1, NCCL will be initialized with
            multiple rank of nodes, each node should have same number of GPUs.
            Distributed training will be enabled then. Default 1.
W
Wu Yi 已提交
106
        trainer_id(int): Must use together with num_trainers. trainer_id is the
C
chengduoZH 已提交
107
            "rank" of current node starts from 0. Default 0.
W
Wu Yi 已提交
108
        scope(Scope): scope to run with, default use fluid.global_scope().
C
chengduoZH 已提交
109 110 111 112 113 114 115 116 117

    Returns:
        ParallelExecutor: The initialized ParallelExecutor object.

    Raises:
        TypeError: If share_vars_from is provided, but not ParallelExecutor object.

    """

X
Xin Pan 已提交
118 119
    def __init__(self,
                 use_cuda,
120 121
                 loss_name=None,
                 main_program=None,
Y
Yu Yang 已提交
122
                 share_vars_from=None,
Y
yuyang18 已提交
123
                 exec_strategy=None,
Y
yuyang18 已提交
124
                 build_strategy=None,
T
typhoonzero 已提交
125
                 num_trainers=1,
126
                 trainer_id=0,
X
Xin Pan 已提交
127
                 scope=None):
Y
yuyang18 已提交
128 129
        if build_strategy is None:
            build_strategy = BuildStrategy()
C
chengduo 已提交
130 131 132 133 134 135 136 137 138 139 140 141 142 143

        # TODO(paddle-dev): trainer_id and num_trainers should be removed from parameter list.
        if num_trainers != 1 and build_strategy.num_trainers != num_trainers:
            sys.stderr.write(
                'The value of build_strategy.num_trainers[%d] is overwritten '
                'by the passed num_trainers[%d].\n' %
                (build_strategy.num_trainers, num_trainers))
            build_strategy.num_trainers = num_trainers
        if trainer_id != 0 and build_strategy.trainer_id != trainer_id:
            sys.stderr.write(
                'The value of build_strategy.trainer_id[%d] is overwritten '
                'by the passed trainer_id[%d].\n' %
                (build_strategy.trainer_id, trainer_id))
            build_strategy.trainer_id = trainer_id
144

S
sneaxiy 已提交
145 146
        self._places = framework.cuda_places(
        ) if use_cuda else framework.cpu_places()
147
        self._scope = scope if scope is not None else executor.global_scope()
X
Xin Pan 已提交
148

149
        if main_program is not None and main_program._enable_dgc:
C
chengduo 已提交
150
            assert build_strategy.num_trainers > 1, "dgc is not useful when num_trainers <= 1"
G
gongweibao 已提交
151 152 153
            assert build_strategy.reduce_strategy == BuildStrategy.ReduceStrategy.AllReduce, "dgc \
                only used for allreduce"

C
chengduo 已提交
154
            assert build_strategy.num_trainers * len(
155
                self._places) > 1, "dgc is not useful for single card training"
G
gongweibao 已提交
156
            assert use_cuda, "dgc only used under cuda"
157

158 159
        main_program = main_program if main_program is not None \
            else framework.default_main_program()
160

161
        self._compiled_program = compiler.CompiledProgram(main_program)
C
chengduo 已提交
162 163 164 165
        if share_vars_from:
            assert isinstance(
                share_vars_from, ParallelExecutor
            ), "The share_vars_from should be ParallelExecutor."
166

167 168 169 170
        self._compiled_program.with_data_parallel(
            loss_name=loss_name,
            build_strategy=build_strategy,
            exec_strategy=exec_strategy,
C
chengduo 已提交
171 172
            share_vars_from=share_vars_from._compiled_program
            if share_vars_from else None)
G
gongweibao 已提交
173

174
        self._place = core.CUDAPlace(0) if use_cuda else core.CPUPlace()
C
chengduo 已提交
175
        self._exe = executor.Executor(self._place)
176

177
    def run(self, fetch_list, feed=None, feed_dict=None, return_numpy=True):
X
Xin Pan 已提交
178
        """
Y
Yu Yang 已提交
179 180 181 182 183 184 185
        Run a parallel executor with fetch_list.

        The feed parameter can be a dict or a list. If feed is a dict, the
        feed data will be split into multiple devices. If feed is a list, we
        assume the data has been splitted into multiple devices, the each
        element in the list will be copied to each device directly.

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 227 228 229 230 231 232 233 234 235 236 237
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              import numpy
              import os

              use_cuda = True
              place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()

              # NOTE: If you use CPU to run the program, you need
              # to specify the CPU_NUM, otherwise, fluid will use
              # all the number of the logic core as the CPU_NUM,
              # in that case, the batch size of the input should be
              # greater than CPU_NUM, if not, the process will be
              # failed by an exception.
              if not use_cuda:
                  os.environ['CPU_NUM'] = str(2)

              exe = fluid.Executor(place)

              train_program = fluid.Program()
              startup_program = fluid.Program()
              with fluid.program_guard(train_program, startup_program):
                  data = fluid.layers.data(name='X', shape=[1], dtype='float32')
                  hidden = fluid.layers.fc(input=data, size=10)
                  loss = fluid.layers.mean(hidden)
                  fluid.optimizer.SGD(learning_rate=0.01).minimize(loss)

              startup_program.random_seed=1
              exe.run(startup_program)

              train_exe = fluid.ParallelExecutor(use_cuda=use_cuda,
                                                 main_program=train_program,
                                                 loss_name=loss.name)

              # If the feed is a dict:
              # the image will be splitted into devices. If there is two devices
              # each device will process an image with shape (5, 1)
              x = numpy.random.random(size=(10, 1)).astype('float32')
              loss_data, = train_exe.run(feed={"X": x},
                                         fetch_list=[loss.name])

              # If the feed is a list:
              # each device will process each element in the list.
              # the 1st device will process an image with shape (10, 1)
              # the 2nd device will process an image with shape (9, 1)
              #
              # you can use exe.device_count to get the device number.
              x2 = numpy.random.random(size=(9, 1)).astype('float32')
              loss_data, = train_exe.run(feed=[{"X": x}, {"X": x2}],
                                         fetch_list=[loss.name])
Y
Yu Yang 已提交
238

Y
Yu Yang 已提交
239 240
        Args:
            fetch_list(list): The fetched variable names
Y
Yu Yang 已提交
241 242 243
            feed(list|dict|None): The feed variables. If the feed is a dict,
                tensors in that dict will be splitted into each devices. If
                the feed is a list, each element of the list will be copied
C
chengduoZH 已提交
244
                to each device. Default None.
Y
Yu Yang 已提交
245
            feed_dict: Alias for feed parameter, for backward compatibility.
C
chengduoZH 已提交
246
                This parameter has been deprecated. Default None.
C
chengduo 已提交
247
            return_numpy(bool): Whether converts the fetched tensor to numpy.
248
                Default: True.
C
chengduoZH 已提交
249 250 251

        Returns:
            List: The fetched result list.
Y
Yu Yang 已提交
252

C
chengduoZH 已提交
253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268
        Raises:
            ValueError: If the feed is a list, but its length is not equal the
                length of active places, or its element's is not dict.

        NOTES:
            1. If the feed's type is dict, the number of data that feeds to
               ParallelExecutor must be bigger than active places. Otherwise,
               it will throw exception from C++ side. Special attention should be
               paid to check whether the last batch of the dataset is bigger
               than active places.
            2. If active places are more than one, the fetch results for each
               variable is a list, and each element of this list is the variable of
               respective active place.

        Examples:
            .. code-block:: python
Y
Yu Yang 已提交
269

C
chengduoZH 已提交
270 271 272 273 274
                pe = fluid.ParallelExecutor(use_cuda=use_cuda,
                                            loss_name=avg_cost.name,
                                            main_program=fluid.default_main_program())
                loss = pe.run(feed=feeder.feed(cur_batch),
                              fetch_list=[avg_cost.name]))
X
Xin Pan 已提交
275
        """
C
chengduo 已提交
276 277 278 279 280
        return self._exe.run(program=self._compiled_program,
                             scope=self._scope,
                             feed=feed,
                             fetch_list=fetch_list,
                             return_numpy=return_numpy)
T
typhoonzero 已提交
281

Y
Yu Yang 已提交
282 283
    @property
    def device_count(self):
284
        return len(self._places)
285 286 287 288 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

    def drop_local_exe_scopes(self):
        """
        Drop the local execution scope immediately.

        During the execution of the Program, the generate intermediate
        results are placed in local execution scope, in some model the
        creation and deletion of those intermediate results are time-consuming.
        To resolve that problem, ParallelExecutor provides an option in
        ExecutionStrategy, i.g. num_iteration_per_drop_scope, this option
        indicates how many iterations to run before dropping the local execution
        scope. But in some situation, each iteration generates different
        intermediate results, it will lead to the result that the memory which
        is needed by local execution scope gradually increase. And if you want
        to run another program at this time, there may be insufficient storage,
        At this point you should drop the local execution scope of other Programs.

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              import numpy
              import os

              use_cuda = True
              # NOTE: If you use CPU to run the program, you need
              # to specify the CPU_NUM, otherwise, fluid will use
              # all the number of the logic core as the CPU_NUM,
              # in that case, the batch size of the input should be
              # greater than CPU_NUM, if not, the process will be
              # failed by an exception.
              if not use_cuda:
                  os.environ['CPU_NUM'] = str(2)

              train_program = fluid.Program()
              startup_program = fluid.Program()
              with fluid.program_guard(train_program, startup_program):
                  data = fluid.layers.data(name='X', shape=[1], dtype='float32')
                  hidden = fluid.layers.fc(input=data, size=10)
                  loss = fluid.layers.mean(hidden)

              place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
327
              exe = fluid.Executor(place)
328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350
              exe.run(startup_program)

              parallel_exe = fluid.ParallelExecutor(use_cuda=use_cuda,
                                                 main_program=train_program,
                                                 loss_name=loss.name)

              x = numpy.random.random(size=(10, 1)).astype('float32')
              loss_data, = parallel_exe.run(feed={"X": x},
                                         fetch_list=[loss.name])

              parallel_exe.drop_local_exe_scopes()
        """
        assert isinstance(
            self._compiled_program._executor,
            core.ParallelExecutor), "The Executor should be ParallelExecutor."
        self._compiled_program._executor.drop_local_exe_scopes()

    # This API is used to check whether DropLocalExeScopes can work.
    def _need_create_local_exe_scopes(self):
        assert isinstance(
            self._compiled_program._executor,
            core.ParallelExecutor), "The Executor should be ParallelExecutor."
        return self._compiled_program._executor._need_create_local_exe_scopes()