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

import core
import multiprocessing
import framework
import executor
J
JiayiFeng 已提交
19
import warnings
Y
Yu Yang 已提交
20
import sys
C
chengduoZH 已提交
21
import os
22

Y
yuyang18 已提交
23
__all__ = ['ParallelExecutor', 'ExecutionStrategy', 'BuildStrategy']
Y
yuyang18 已提交
24 25

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


class ParallelExecutor(object):
C
chengduoZH 已提交
30 31 32 33 34 35 36 37 38 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
    """
    ParallelExecutor can run program in parallel.

    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.
        share_vars_from(ParallelExecutor): If provied, it will share variables
            from the specified ParallelExecutor. Default None.
        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.
        trainer_id(int: Must use together with num_trainers. trainer_id is the
            "rank" of current node starts from 0. Default 0.

    Returns:
        ParallelExecutor: The initialized ParallelExecutor object.

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

    Examples:
        .. code-block:: python

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

          train_loss, = train_exe.run([loss.name], feed=feed_dict)
          test_loss, = test_exe.run([loss.name], feed=feed_dict)
    """

X
Xin Pan 已提交
64 65
    def __init__(self,
                 use_cuda,
66 67
                 loss_name=None,
                 main_program=None,
Y
Yu Yang 已提交
68
                 share_vars_from=None,
Y
yuyang18 已提交
69
                 exec_strategy=None,
Y
yuyang18 已提交
70
                 build_strategy=None,
T
typhoonzero 已提交
71
                 num_trainers=1,
72
                 trainer_id=0,
Y
yuyang18 已提交
73 74 75 76 77 78 79 80 81
                 **kwargs):
        if len(kwargs) != 0:
            err_msg = ""
            for key in kwargs:
                if key in dir(ExecutionStrategy):
                    err_msg += \
                        "Setting {0} by constructor is deprecated. Use " \
                        "strategy=ExecutionStrategy(); strategy.{0}=xxx; " \
                        "pe=ParallelExecutor(exec_strategy=strategy) " \
Y
yuyang18 已提交
82 83 84 85 86 87 88 89 90 91
                        "instead.\n ".format(key)
                elif key in dir(BuildStrategy):
                    err_msg += \
                        "Setting {0} by constructor is deprecated. Use " \
                        "strategy=BuildStrategy(); See help(" \
                        "paddle.fluid.ParallelExecutor.BuildStrategy) \n".format(
                            key)
                else:
                    err_msg += "Setting {0} by constructor is deprecated. Use strategy.\n".format(
                        key)
Y
yuyang18 已提交
92
            raise ValueError(err_msg)
93

X
Xin Pan 已提交
94 95
        self._places = []
        self._act_places = []
96 97 98
        if use_cuda:
            for i in xrange(core.get_cuda_device_count()):
                p = core.Place()
X
Xin Pan 已提交
99 100 101
                self._act_places.append(core.CUDAPlace(i))
                p.set_place(self._act_places[-1])
                self._places.append(p)
102
        else:
C
chengduoZH 已提交
103 104 105
            cpu_num = int(
                os.environ.get('CPU_NUM', multiprocessing.cpu_count()))
            for i in xrange(cpu_num):
106
                p = core.Place()
L
Luo Tao 已提交
107
                self._act_places.append(core.CPUPlace())
X
Xin Pan 已提交
108 109 110
                p.set_place(self._act_places[-1])
                self._places.append(p)
        assert self._places, "no place for execution"
111

Y
yuyang18 已提交
112 113
        if exec_strategy is None:
            exec_strategy = ExecutionStrategy()
114
        exec_strategy.use_cuda = use_cuda
Y
yuyang18 已提交
115 116

        if exec_strategy.num_threads == 0:
X
Xin Pan 已提交
117 118 119
            if use_cuda:
                # Experiments on se-resnext shows that too many threads hurt
                # performance. Worth tunning for other models in the future.
C
chengduoZH 已提交
120
                exec_strategy.num_threads = len(self._places) * 4
X
Xin Pan 已提交
121
            else:
C
chengduoZH 已提交
122 123 124
                cpu_num = int(
                    os.environ.get('CPU_NUM', multiprocessing.cpu_count()))
                exec_strategy.num_threads = cpu_num
125

Y
yuyang18 已提交
126 127 128
        if build_strategy is None:
            build_strategy = BuildStrategy()

129 130
        main = main_program
        main = main if main else framework.default_main_program()
131
        scope = executor.global_scope()
132 133 134 135 136
        # FIXME(Yancey1989): it's a temporary approach to determinate the distribute
        # train program, call self.bcast_param() at the end of each mini-batch.
        self.is_dist = True if "recv" in [
            op.type for op in main.global_block().ops
        ] else False
137

138 139 140
        if share_vars_from and not isinstance(share_vars_from,
                                              ParallelExecutor):
            raise TypeError("share_vars_from must be ParallelExecutor.")
C
chengduoZH 已提交
141

142 143 144
        local_scopes = share_vars_from.executor.local_scopes(
        ) if share_vars_from else []

T
typhoonzero 已提交
145
        self.persistable_vars = [
146
            v.name
147 148
            for v in filter(
                lambda var: var.persistable and var.type != core.VarDesc.VarType.RAW,
T
typhoonzero 已提交
149
                main.list_vars())
150 151
        ]

152
        self.executor = core.ParallelExecutor(
X
Xin Pan 已提交
153
            self._places,
154
            set([
W
Wu Yi 已提交
155
                p.name for p in main.global_block()._iter_parameters()
156 157
                if not p.stop_gradient
            ]),
158 159 160
            set(self.persistable_vars), main.desc, loss_name
            if loss_name else '', scope, local_scopes, exec_strategy,
            build_strategy, num_trainers, trainer_id)
161 162
        self.scope = scope

163
    def run(self, fetch_list, feed=None, feed_dict=None, return_numpy=True):
X
Xin Pan 已提交
164
        """
Y
Yu Yang 已提交
165 166 167 168 169 170 171 172
        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.

        For example, if the feed is a dict:
C
chengduoZH 已提交
173

Y
Yu Yang 已提交
174 175 176 177 178 179
        >>> exe = ParallelExecutor()
        >>> # the image will be splitted into devices. If there is two devices
        >>> # each device will process an image with shape (24, 1, 28, 28)
        >>> exe.run(feed={'image': numpy.random.random(size=(48, 1, 28, 28))})

        For example, if the feed is a list:
C
chengduoZH 已提交
180

Y
Yu Yang 已提交
181 182 183 184 185 186 187 188 189 190
        >>> exe = ParallelExecutor()
        >>> # each device will process each element in the list.
        >>> # the 1st device will process an image with shape (48, 1, 28, 28)
        >>> # the 2nd device will process an image with shape (32, 1, 28, 28)
        >>> #
        >>> # you can use exe.device_count to get the device number.
        >>> exe.run(feed=[{"image": numpy.random.random(size=(48, 1, 28, 28))},
        >>>               {"image": numpy.random.random(size=(32, 1, 28, 28))},
        >>>              ])

Y
Yu Yang 已提交
191 192
        Args:
            fetch_list(list): The fetched variable names
Y
Yu Yang 已提交
193 194 195
            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 已提交
196
                to each device. Default None.
Y
Yu Yang 已提交
197
            feed_dict: Alias for feed parameter, for backward compatibility.
C
chengduoZH 已提交
198
                This parameter has been deprecated. Default None.
C
chengduo 已提交
199
            return_numpy(bool): Whether converts the fetched tensor to numpy.
200
                Default: True.
C
chengduoZH 已提交
201 202 203

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

C
chengduoZH 已提交
205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220
        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 已提交
221

C
chengduoZH 已提交
222 223 224 225 226
                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 已提交
227
        """
228
        if feed is None and feed_dict is not None:
J
JiayiFeng 已提交
229
            feed = feed_dict
Y
Yu Yang 已提交
230
            print >> sys.stderr, "`feed_dict` is deprecated. Please use `feed=`"
Y
Yu Yang 已提交
231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266

        if isinstance(feed, dict):
            feed_tensor_dict = dict()
            for feed_name in feed:
                feed_tensor = feed[feed_name]
                if not isinstance(feed_tensor, core.LoDTensor):
                    feed_tensor = core.LoDTensor()
                    # always set to CPU place, since the tensor need to be splitted
                    # it is fast in CPU
                    feed_tensor.set(feed[feed_name], core.CPUPlace())
                feed_tensor_dict[feed_name] = feed_tensor

            self.executor.feed_and_split_tensor_into_local_scopes(
                feed_tensor_dict)
        elif isinstance(feed, list) or isinstance(feed, tuple):
            if len(feed) != len(self._act_places):
                raise ValueError(
                    "Feed a list of tensor, the list should be the same size as places"
                )

            res = list()

            for i, each in enumerate(feed):
                if not isinstance(each, dict):
                    raise TypeError(
                        "Each element of feed list should be a dict")
                res_dict = dict()
                for feed_name in each:
                    tensor = each[feed_name]
                    if not isinstance(tensor, core.LoDTensor):
                        tmp = core.LoDTensor()
                        tmp.set(tensor, self._act_places[i])
                        tensor = tmp
                    res_dict[feed_name] = tensor
                res.append(res_dict)
            self.executor.feed_tensors_into_local_scopes(res)
X
Xin Pan 已提交
267

268
        fetch_var_name = '@FETCHED_VAR_NAME@'
Y
Yu Yang 已提交
269
        self.executor.run(fetch_list, fetch_var_name)
270
        arr = self.scope.find_var(fetch_var_name).get_lod_tensor_array()
271 272 273 274

        if self.is_dist:
            self.bcast_params()

C
chengduo 已提交
275 276 277
        if return_numpy:
            return executor.as_numpy(arr)

278
        return [arr[i] for i in range(len(arr))]
T
typhoonzero 已提交
279 280

    def bcast_params(self):
C
chengduoZH 已提交
281 282 283 284
        """
        Broadcast the parameters to other devices. It is used during
        distributed training.
        """
T
typhoonzero 已提交
285
        self.executor.bcast_params(set(self.persistable_vars))
Y
Yu Yang 已提交
286 287 288 289

    @property
    def device_count(self):
        return len(self._act_places)