data_feeder.py 23.3 KB
Newer Older
1
#   Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved.
D
dzhwinter 已提交
2
#
D
dzhwinter 已提交
3 4 5
# 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
D
dzhwinter 已提交
6
#
D
dzhwinter 已提交
7
#     http://www.apache.org/licenses/LICENSE-2.0
D
dzhwinter 已提交
8
#
D
dzhwinter 已提交
9 10 11 12 13 14
# 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
from . import core
16
import numpy as np
C
chengduoZH 已提交
17
import os
Y
yuyang18 已提交
18
import multiprocessing
19
import warnings
Y
Yu Yang 已提交
20

21 22 23 24 25 26 27
from .framework import (
    Variable,
    default_main_program,
    _current_expected_place,
    _non_static_mode,
    _in_eager_without_dygraph_check,
)
C
chengduo 已提交
28
from .framework import _cpu_num, _cuda_ids
29

Y
Yu Yang 已提交
30 31
__all__ = ['DataFeeder']

L
Leo Chen 已提交
32 33 34
_PADDLE_DTYPE_2_NUMPY_DTYPE = {
    core.VarDesc.VarType.BOOL: 'bool',
    core.VarDesc.VarType.FP16: 'float16',
35
    core.VarDesc.VarType.BF16: 'uint16',
L
Leo Chen 已提交
36 37 38 39 40 41 42 43 44 45 46
    core.VarDesc.VarType.FP32: 'float32',
    core.VarDesc.VarType.FP64: 'float64',
    core.VarDesc.VarType.INT8: 'int8',
    core.VarDesc.VarType.INT16: 'int16',
    core.VarDesc.VarType.INT32: 'int32',
    core.VarDesc.VarType.INT64: 'int64',
    core.VarDesc.VarType.UINT8: 'uint8',
    core.VarDesc.VarType.COMPLEX64: 'complex64',
    core.VarDesc.VarType.COMPLEX128: 'complex128',
}

Y
Yu Yang 已提交
47

S
sneaxiy 已提交
48
def convert_dtype(dtype):
P
pkpk 已提交
49
    if isinstance(dtype, core.VarDesc.VarType):
L
Leo Chen 已提交
50 51
        if dtype in _PADDLE_DTYPE_2_NUMPY_DTYPE:
            return _PADDLE_DTYPE_2_NUMPY_DTYPE[dtype]
52
    elif isinstance(dtype, type):
53
        # This branch is for NumPy scalar types
54
        if dtype in [
55 56 57 58 59 60 61 62 63 64 65 66
            bool,
            np.float16,
            np.uint16,
            np.float32,
            np.float64,
            np.int8,
            np.int16,
            np.int32,
            np.int64,
            np.uint8,
            np.complex64,
            np.complex128,
67 68
        ]:
            return dtype.__name__
P
pkpk 已提交
69
    else:
70
        # This branch is for np.dtype and str
P
pkpk 已提交
71
        if dtype in [
72 73 74 75 76 77 78 79 80 81 82 83
            'bool',
            'float16',
            'uint16',
            'float32',
            'float64',
            'int8',
            'int16',
            'int32',
            'int64',
            'uint8',
            'complex64',
            'complex128',
P
pkpk 已提交
84
        ]:
85 86 87
            # NOTE(SigureMo): Since the np.dtype object is not an instance of
            # type, so it will not be handled by the previous branch. We need
            # to convert it to str here.
P
pkpk 已提交
88
            return str(dtype)
89 90 91
        # NOTE(zhangbo): Now numpy does not support bfloat, and paddle use uint16 to represent bfloat16, and there binaries are consistent.
        if dtype in ['bfloat16']:
            return 'uint16'
P
pkpk 已提交
92

93
    raise TypeError(
94
        "dtype must be any of [bool, float16, uint16, float32, float64, int8, int16, "
95 96
        "int32, int64, uint8, complex64, complex128, bfloat16], but received %s"
        % dtype
97
    )
S
sneaxiy 已提交
98 99


100 101 102
def check_variable_and_dtype(
    input, input_name, expected_dtype, op_name, extra_message=''
):
103
    check_type(input, input_name, Variable, op_name, extra_message)
104 105 106 107
    check_dtype(input.dtype, input_name, expected_dtype, op_name, extra_message)


def check_type(input, input_name, expected_type, op_name, extra_message=''):
108 109 110 111 112 113 114
    # NOTE [ Why skip dynamic graph check ]:
    # 1. If the input type / dtype of a layer is wrong, it will be reported
    # directly on that line. User can easily print the relevant information
    # on which line. It is easier to debug, so there is no need to check
    # in dynamic graph mode.
    # 2. Performance considerations. Because these checks are executed at
    # each step in dynamic graph mode, it will bring a heavy performance burden.
J
Jiabin Yang 已提交
115
    if _non_static_mode():
116
        return
117 118

    # NOTE: `in_declarative_mode` is used to determined whether this op is called under
H
hjyp 已提交
119
    # @to_static in transformation from dygrah to static layer. We add VarBase in
120
    # expected_type to skip checking because varBase may be created and used in unusual way.
121
    from .dygraph.base import in_declarative_mode
122

123 124 125
    # Need a better design to be fix this.
    if in_declarative_mode():
        if not isinstance(expected_type, tuple):
126 127
            expected_type = (expected_type,)
        expected_type += (core.VarBase,)
J
Jiabin Yang 已提交
128
        if _in_eager_without_dygraph_check():
129
            expected_type += (core.eager.Tensor,)
130 131 132 133
    elif isinstance(input, core.VarBase):
        raise TypeError(
            "Please use `with fluid.dygraph.guard()` as context or `fluid.enable_dygraph()` to switch to imperative mode firstly. "
            "Because received '{}' in {} is a imperative Variable.".format(
134 135 136
                input_name, op_name
            )
        )
137
    elif hasattr(core, "eager"):
138
        if isinstance(input, core.eager.Tensor):
139 140 141
            raise TypeError(
                "Please use `with fluid.dygraph.guard()` as context or `fluid.enable_dygraph()` to switch to imperative mode firstly. "
                "Because received '{}' in {} is a imperative Variable.".format(
142 143 144
                    input_name, op_name
                )
            )
145 146
    if not isinstance(input, expected_type):
        raise TypeError(
147 148 149
            "The type of '%s' in %s must be %s, but received %s. %s"
            % (input_name, op_name, expected_type, type(input), extra_message)
        )
150 151


152 153 154
def check_dtype(
    input_dtype, input_name, expected_dtype, op_name, extra_message=''
):
155
    # See NOTE [ Why skip dynamic graph check ]
J
Jiabin Yang 已提交
156
    if _non_static_mode():
157
        return
158 159
    if convert_dtype(input_dtype) in ['float16']:
        warnings.warn(
160 161 162
            "The data type of '%s' in %s only support float16 in GPU now. %s"
            % (input_name, op_name, extra_message)
        )
163
    if convert_dtype(input_dtype) in ['uint16'] and op_name not in [
164 165 166
        'reshape',
        'lookup_table',
        'scale',
167 168 169
    ]:
        warnings.warn(
            "The data type of '%s' in %s only support bfloat16 in OneDNN now. %s"
170 171
            % (input_name, op_name, extra_message)
        )
172 173
    if convert_dtype(input_dtype) not in expected_dtype:
        raise TypeError(
174 175 176
            "The data type of '%s' in %s must be %s, but received %s. %s"
            % (
                input_name,
177
                op_name,
178 179 180 181 182 183 184 185 186 187 188 189 190 191
                expected_dtype,
                convert_dtype(input_dtype),
                extra_message,
            )
        )


def check_shape(
    shape,
    op_name,
    expected_shape_type=(list, tuple, Variable),
    expected_element_type=(int, Variable),
    expected_tensor_dtype=('int32', 'int64'),
):
192
    # See NOTE [ Why skip dynamic graph check ]
J
Jiabin Yang 已提交
193
    if _non_static_mode():
194 195 196 197 198 199 200
        return
    check_type(shape, 'shape', expected_shape_type, op_name)
    if expected_element_type is not None and not isinstance(shape, Variable):
        for item in shape:
            check_type(item, 'element of shape', expected_element_type, op_name)
            if expected_tensor_dtype is not None and isinstance(item, Variable):
                check_dtype(
201 202 203
                    item.dtype,
                    'element of shape',
                    expected_tensor_dtype,
204
                    op_name,
205 206 207 208
                    'If element of shape is Tensor, its data type should be {}'.format(
                        ', '.join(expected_tensor_dtype)
                    ),
                )
209 210 211 212
    if expected_tensor_dtype is not None and isinstance(shape, Variable):
        check_dtype(shape.dtype, 'shape', expected_tensor_dtype, op_name)


213
class DataToLoDTensorConverter:
Y
Yu Yang 已提交
214 215 216 217
    def __init__(self, place, lod_level, shape, dtype):
        self.place = place
        self.lod_level = lod_level
        self.shape = shape
218 219 220 221 222 223 224
        negtive_count = 0
        for s in self.shape:
            if s < 0:
                negtive_count += 1
            if negtive_count > 1:
                self.shape = None
                break
S
sneaxiy 已提交
225 226
        self.dtype = convert_dtype(dtype)
        self._reset()
Y
Yu Yang 已提交
227

S
sneaxiy 已提交
228
    def _reset(self):
Y
Yu Yang 已提交
229
        self.data = []
230
        self.lod = [[] for _ in range(self.lod_level)]
Y
Yu Yang 已提交
231 232 233 234 235 236 237 238

    def feed(self, data):
        self._feed_impl_(data, self.lod, self.lod_level)

    def _feed_impl_(self, data, lod, lod_level):
        if lod_level == 0:
            self.data.append(data)
        else:
239
            lod[0].append(len(data))
Y
Yu Yang 已提交
240
            for each_data in data:
K
Kexin Zhao 已提交
241
                self._feed_impl_(each_data, lod[1:], lod_level - 1)
Y
Yu Yang 已提交
242

S
sneaxiy 已提交
243
    def _check_shape(self, shape):
S
sneaxiy 已提交
244 245 246
        for s1, s2 in zip(self.shape, shape):
            if s1 != s2 and s1 >= 0 and s2 >= 0:
                raise ValueError(
247 248 249 250
                    "Shape not match. What is defined in data layer is {}, but receive {}".format(
                        self.shape, shape
                    )
                )
S
sneaxiy 已提交
251

Y
Yu Yang 已提交
252
    def done(self):
253
        arr = np.array(self.data, dtype=self.dtype)
S
sneaxiy 已提交
254 255
        if self.shape:
            if len(arr.shape) != len(self.shape):
S
sneaxiy 已提交
256 257 258 259
                try:
                    arr = arr.reshape(self.shape)
                except ValueError:
                    raise ValueError(
260 261 262 263
                        "Reshape error. What is defined in data layer is {}, but receive {}".format(
                            self.shape, arr.shape
                        )
                    )
Y
Yu Yang 已提交
264 265 266
        t = core.LoDTensor()
        t.set(arr, self.place)
        if self.lod_level > 0:
267
            t.set_recursive_sequence_lengths(self.lod)
S
sneaxiy 已提交
268
        self._reset()
Y
Yu Yang 已提交
269 270 271
        return t


272
class BatchedTensorProvider:
S
sneaxiy 已提交
273 274 275 276 277 278 279 280 281 282
    def __init__(self, feed_list, place, batch_size, generator, drop_last):
        self.place = place
        self.batch_size = batch_size
        self.generator = generator
        self.converters = []
        self.drop_last = drop_last

        for var in feed_list:
            assert var.lod_level == 0, "lod_level must be 0"
            self.converters.append(
283 284 285 286 287 288 289
                DataToLoDTensorConverter(
                    place=self.place,
                    lod_level=0,
                    shape=var.shape,
                    dtype=var.dtype,
                )
            )
S
sneaxiy 已提交
290 291 292 293 294 295 296

    def _done(self):
        return [c.done() for c in self.converters]

    def __call__(self):
        idx = 0
        for each_sample in self.generator():
297
            for each_slot, each_converter in zip(each_sample, self.converters):
S
sneaxiy 已提交
298 299 300 301 302 303 304 305 306 307 308 309 310
                each_converter.data.append(each_slot)

            idx += 1
            if idx == self.batch_size:
                idx = 0
                yield self._done()

        if not self.drop_last and idx > 0:
            yield self._done()
        else:
            [c._reset() for c in self.converters]


311
class DataFeeder:
C
chengduoZH 已提交
312
    """
313
    :api_attr: Static Graph
314

C
chengduoZH 已提交
315
    DataFeeder converts the data that returned by a reader into a data
316 317
    structure that can feed into Executor. The reader is usually a
    python generator that returns a list of mini-batch data entries.
318 319 320 321

    Parameters:
        feed_list (list): Variables or names of Variables that need
            to feed.
322 323 324 325
        place (:ref:`api_fluid_CPUPlace` | :ref:`api_fluid_CUDAPlace` ):
            place indicates the device (CPU | GPU) the data will be fed into, if
            you want to feed data into GPU, please using :code:`fluid.CUDAPlace(i)`
            (:code:`i` represents the GPU id), or if you want to feed data into CPU,
326
            please using :code:`fluid.CPUPlace()`.
327 328
        program (:ref:`api_fluid_Program` , optional): The Program that will
            feed data into, if program is None, it will use default_main_program().
329
            Default None.
C
chengduoZH 已提交
330 331

    Raises:
332
        :code:`ValueError` - If some Variables are not in this Program.
C
chengduoZH 已提交
333

334
    Example:
335 336 337 338 339
        ..  code-block:: python

            import numpy as np
            import paddle
            import paddle.fluid as fluid
340

C
chengduoZH 已提交
341
            place = fluid.CPUPlace()
342
            def reader():
343 344
                for _ in range(4):
                    yield np.random.random([4]).astype('float32'), np.random.random([3]).astype('float32'),
345

346 347
            main_program = fluid.Program()
            startup_program = fluid.Program()
348

349
            with fluid.program_guard(main_program, startup_program):
350 351
                data_1 = fluid.data(name='data_1', shape=[None, 2, 2], dtype='float32')
                data_2 = fluid.data(name='data_2', shape=[None, 1, 3], dtype='float32')
352 353 354
                out = fluid.layers.fc(input=[data_1, data_2], size=2)
                # ...
            feeder = fluid.DataFeeder([data_1, data_2], place)
355

356 357
            exe = fluid.Executor(place)
            exe.run(startup_program)
358

359
            feed_data = feeder.feed(reader())
360

361 362 363
            # print feed_data to view feed results
            # print(feed_data['data_1'])
            # print(feed_data['data_2'])
364

365 366 367
            outs = exe.run(program=main_program,
                            feed=feed_data,
                            fetch_list=[out])
368
            print(outs)
369

C
chengduoZH 已提交
370 371
    """

F
fengjiayi 已提交
372
    def __init__(self, feed_list, place, program=None):
Y
Yu Yang 已提交
373 374 375 376
        self.feed_dtypes = []
        self.feed_names = []
        self.feed_shapes = []
        self.feed_lod_level = []
F
fengjiayi 已提交
377 378
        if program is None:
            program = default_main_program()
Y
Yu Yang 已提交
379
        for each_var in feed_list:
380
            if isinstance(each_var, str):
F
fengjiayi 已提交
381
                each_var = program.block(0).var(each_var)
Y
Yu Yang 已提交
382 383 384 385 386
            if not isinstance(each_var, Variable):
                raise TypeError("Feed list should contain a list of variable")
            self.feed_dtypes.append(each_var.dtype)
            self.feed_names.append(each_var.name)
            self.feed_lod_level.append(each_var.lod_level)
S
sneaxiy 已提交
387
            self.feed_shapes.append(each_var.shape)
Y
Yu Yang 已提交
388 389 390 391

        self.place = place

    def feed(self, iterable):
C
chengduoZH 已提交
392
        """
393
        According to :code:`feed_list` of :code:`DataFeeder` and :code:`iterable` , converts
394
        the input into a data structure that can feed into Executor.
C
chengduoZH 已提交
395

396 397
        Parameters:
            iterable (generator): user defined python generator to read the raw input data
C
chengduoZH 已提交
398

399
        Returns:
400
            :code:`dict`: a :code:`dict` that contains (variable name - converted tensor) pairs
401

402
        Example:
403 404
            ..  code-block:: python

405 406 407 408 409 410
                # In this example, reader - generator will return a list of ndarray of 3 elements
                # feed API will convert each ndarray input into a tensor
                # the return result is a dict with keys: data_1, data_2, data_3
                # result['data_1']  a LoD-Tensor with shape of  [5, 2, 1, 3]. 5 is batch size, and [2, 1, 3] is the real shape of data_1.
                # result['data_2'], result['data_3'] are similar.
                import numpy as np
411
                import paddle.fluid as fluid
412

413
                def reader(limit=5):
414 415
                    for i in range(1, limit + 1):
                        yield np.ones([6]).astype('float32') * i , np.ones([1]).astype('int64') * i, np.random.random([9]).astype('float32')
416

417 418 419
                data_1 = fluid.data(name='data_1', shape=[None, 2, 1, 3])
                data_2 = fluid.data(name='data_2', shape=[None, 1], dtype='int64')
                data_3 = fluid.data(name='data_3', shape=[None, 3, 3], dtype='float32')
420
                feeder = fluid.DataFeeder(['data_1','data_2', 'data_3'], fluid.CPUPlace())
421 422


423 424 425
                result = feeder.feed(reader())
                print(result['data_1'])
                print(result['data_2'])
426
                print(result['data_3'])
427

C
chengduoZH 已提交
428
        """
Y
Yu Yang 已提交
429
        converter = []
430 431 432
        for lod_level, shape, dtype in zip(
            self.feed_lod_level, self.feed_shapes, self.feed_dtypes
        ):
Y
Yu Yang 已提交
433
            converter.append(
434 435 436 437 438 439 440
                DataToLoDTensorConverter(
                    place=self.place,
                    lod_level=lod_level,
                    shape=shape,
                    dtype=dtype,
                )
            )
Y
Yu Yang 已提交
441 442

        for each_sample in iterable:
443
            assert len(each_sample) == len(converter), (
444 445 446
                "The number of fields in data (%d) does not match "
                + "len(feed_list) (%d)"
            ) % (len(each_sample), len(converter))
447
            for each_converter, each_slot in zip(converter, each_sample):
Y
Yu Yang 已提交
448 449
                each_converter.feed(each_slot)
        ret_dict = {}
450
        for each_name, each_converter in zip(self.feed_names, converter):
Y
Yu Yang 已提交
451 452
            ret_dict[each_name] = each_converter.done()
        return ret_dict
Y
yuyang18 已提交
453 454

    def feed_parallel(self, iterable, num_places=None):
C
chengduoZH 已提交
455
        """
456
        Similar with feed function, feed_parallel is used with multiple devices (CPU|GPU).
457 458
        Here :code:`iterable` is a list of python generators. The data return by each
        generator in the list will be fed into a separate device.
C
chengduoZH 已提交
459

460
        Parameters:
461
            iterable (list|tuple): list of user-defined python generators. The element
462
                number should match the :code:`num_places`.
463
            num_places (int, optional): the number of devices. If not provided (None),
464
                all available devices on the machine will be used. Default None.
C
chengduoZH 已提交
465

466 467
        Returns:
            :code:`generator`: a :code:`generator` that generate dict which contains (variable name - converted tensor) pairs,
468
            the total number of dicts will be generated matches with the :code:`num_places`
C
chengduoZH 已提交
469

470
        .. note::
471
            The number of devices - :code:`num_places` should equal to the generator (element of :code:`iterable` ) number
472

473
        Example:
474 475
            ..  code-block:: python

476
                import numpy as np
477
                import paddle.fluid as fluid
478

479 480 481 482 483
                def generate_reader(batch_size, base=0, factor=1):
                    def _reader():
                        for i in range(batch_size):
                            yield np.ones([4]) * factor + base, np.ones([4]) * factor + base + 5
                    return _reader()
484 485 486 487

                x = fluid.data(name='x', shape=[None, 2, 2])
                y = fluid.data(name='y', shape=[None, 2, 2], dtype='float32')

H
HongyuJia 已提交
488
                z = paddle.add(x, y)
489

490
                feeder = fluid.DataFeeder(['x','y'], fluid.CPUPlace())
491
                place_num = 2
492 493 494 495 496
                places = [fluid.CPUPlace() for x in range(place_num)]
                data = []
                exe = fluid.Executor(fluid.CPUPlace())
                exe.run(fluid.default_startup_program())
                program = fluid.CompiledProgram(fluid.default_main_program()).with_data_parallel(places=places)
497

T
tianshuo78520a 已提交
498
                # print sample feed_parallel r result
499 500 501
                # for item in list(feeder.feed_parallel([generate_reader(5, 0, 1), generate_reader(3, 10, 2)], 2)):
                #     print(item['x'])
                #     print(item['y'])
502

503 504 505
                reader_list = [generate_reader(5, 0, 1), generate_reader(3, 10, 2)]
                res = exe.run(program=program, feed=list(feeder.feed_parallel(reader_list, 2)), fetch_list=[z])
                print(res)
506

C
chengduoZH 已提交
507
        """
Y
yuyang18 已提交
508 509
        if isinstance(self.place, core.CUDAPlace):
            places = [
510 511
                core.CUDAPlace(i)
                for i in range(self._get_number_of_places_(num_places))
Y
yuyang18 已提交
512 513 514
            ]
        else:
            places = [
515 516
                core.CPUPlace()
                for _ in range(self._get_number_of_places_(num_places))
Y
yuyang18 已提交
517 518 519
            ]

        if len(iterable) != len(places):
520 521 522 523 524 525
            raise ValueError(
                "feed_parallel takes multiple mini-batches. Each "
                "mini-batch will be feed on each device. The "
                "number of devices and number of mini-batches "
                "must be same."
            )
Y
yuyang18 已提交
526 527

        place = self.place
528
        for p, batch in zip(places, iterable):
Y
yuyang18 已提交
529 530 531 532 533 534 535 536
            self.place = p
            yield self.feed(batch)
        self.place = place

    def _get_number_of_places_(self, num_places):
        if num_places is not None:
            return int(num_places)
        elif isinstance(self.place, core.CUDAPlace):
C
chengduo 已提交
537
            return len(_cuda_ids())
Y
yuyang18 已提交
538
        else:
C
chengduo 已提交
539
            return _cpu_num()
Y
yuyang18 已提交
540

541 542 543
    def decorate_reader(
        self, reader, multi_devices, num_places=None, drop_last=True
    ):
C
chengduoZH 已提交
544
        """
545 546 547 548 549
        Decorate the reader (generator) to fit multiple devices. The reader generate
        multiple mini-batches. Each mini-batch will be fed into a single device.

        Parameters:
            reader(generator): a user defined python generator used to get :code:`mini-batch` of data.
550 551
                A :code:`mini-batch` can be regarded as a python generator that returns batches of input
                entities, just like the below :code:`_mini_batch` in the code example.
552 553 554 555
            multi_devices(bool): indicate whether to use multiple devices or not.
            num_places(int, optional): if :code:`multi_devices` is True, you can specify the number
                of devices(CPU|GPU) to use, if multi_devices is None, the function will use all the
                devices of the current machine. Default None.
556
            drop_last(bool, optional): whether to drop the last round of data if it is not enough to
557 558
                feed all devices. Default True.

559
        Returns:
560
            :code:`generator`: a new :code:`generator` which return converted dicts that can be fed into Executor
561

C
chengduoZH 已提交
562
        Raises:
563
            :code:`ValueError`: If drop_last is False and the data cannot fit devices perfectly.
564

565
        Example:
566 567
            ..  code-block:: python

568
                import numpy as np
569 570
                import paddle
                import paddle.fluid as fluid
571
                import paddle.fluid.compiler as compiler
572

573 574 575 576
                def reader():
                    def _mini_batch(batch_size):
                        for i in range(batch_size):
                            yield np.random.random([16]).astype('float32'), np.random.randint(10, size=[1])
577

578 579
                    for _ in range(10):
                        yield _mini_batch(np.random.randint(1, 10))
580

581 582
                place_num = 3
                places = [fluid.CPUPlace() for _ in range(place_num)]
583

584
                # a simple network sample
585 586
                data = fluid.data(name='data', shape=[None, 4, 4], dtype='float32')
                label = fluid.data(name='label', shape=[None, 1], dtype='int64')
587
                hidden = fluid.layers.fc(input=data, size=10)
588

589 590
                feeder = fluid.DataFeeder(place=places[0], feed_list=[data, label])
                reader = feeder.decorate_reader(reader, multi_devices=True, num_places=3, drop_last=True)
591

592
                exe = fluid.Executor(places[0])
593
                exe.run(fluid.default_startup_program())
594
                compiled_prog = compiler.CompiledProgram(
595
                         fluid.default_main_program()).with_data_parallel(places=places)
596

597
                for i,data in enumerate(reader()):
598 599
                    # print data if you like
                    # print(i, data)
600
                    ret = exe.run(compiled_prog, feed=data, fetch_list=[hidden])
601 602
                    print(ret)

C
chengduoZH 已提交
603 604
        """

Y
yuyang18 已提交
605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620
        def __reader_creator__():
            if not multi_devices:
                for item in reader():
                    yield self.feed(item)
            else:
                num = self._get_number_of_places_(num_places)
                item = []
                for batch in reader():
                    item.append(batch)
                    if len(item) == num:
                        yield list(self.feed_parallel(item, num))
                        item = []
                if not drop_last and len(item) != 0:
                    raise ValueError(
                        "The data batch which cannot fit for devices will be "
                        "dropped is not implementation. Other strategies are "
621 622
                        "not implemented"
                    )
Y
yuyang18 已提交
623 624

        return __reader_creator__