data_feeder.py 10.4 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 16
from __future__ import print_function

17
from . import core
Y
Yu Yang 已提交
18
import numpy
C
chengduoZH 已提交
19
import os
20 21
import six
from six.moves import zip, range, xrange
Y
yuyang18 已提交
22
import multiprocessing
Y
Yu Yang 已提交
23

24
from .framework import Variable, default_main_program
Y
Yu Yang 已提交
25 26 27 28 29 30 31 32 33

__all__ = ['DataFeeder']


class DataToLoDTensorConverter(object):
    def __init__(self, place, lod_level, shape, dtype):
        self.place = place
        self.lod_level = lod_level
        self.shape = shape
34 35 36 37 38 39 40
        negtive_count = 0
        for s in self.shape:
            if s < 0:
                negtive_count += 1
            if negtive_count > 1:
                self.shape = None
                break
41
        if dtype == core.VarDesc.VarType.FP32:
Y
Yu Yang 已提交
42
            self.dtype = 'float32'
43
        elif dtype == core.VarDesc.VarType.INT64:
Y
Yu Yang 已提交
44
            self.dtype = 'int64'
45
        elif dtype == core.VarDesc.VarType.FP64:
Y
Yu Yang 已提交
46
            self.dtype = 'float64'
47
        elif dtype == core.VarDesc.VarType.INT32:
Y
Yu Yang 已提交
48
            self.dtype = 'int32'
F
fengjiayi 已提交
49 50
        elif dtype == core.VarDesc.VarType.UINT8:
            self.dtype = 'uint8'
Y
Yu Yang 已提交
51 52
        else:
            raise ValueError("dtype must be any of [int32, float32, int64, "
F
fengjiayi 已提交
53
                             "float64, uint8]")
Y
Yu Yang 已提交
54 55 56 57

        self.data = []
        self.lod = []

58
        for i in six.moves.range(lod_level):
59
            self.lod.append([])
Y
Yu Yang 已提交
60 61 62 63 64 65 66 67

    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:
68
            lod[0].append(len(data))
Y
Yu Yang 已提交
69
            for each_data in data:
K
Kexin Zhao 已提交
70
                self._feed_impl_(each_data, lod[1:], lod_level - 1)
Y
Yu Yang 已提交
71 72

    def done(self):
73 74 75
        arr = numpy.array(self.data, dtype=self.dtype)
        if self.shape:
            arr = arr.reshape(self.shape)
Y
Yu Yang 已提交
76 77 78
        t = core.LoDTensor()
        t.set(arr, self.place)
        if self.lod_level > 0:
79
            t.set_recursive_sequence_lengths(self.lod)
Y
Yu Yang 已提交
80 81 82 83
        return t


class DataFeeder(object):
C
chengduoZH 已提交
84
    """
C
chengduoZH 已提交
85 86
    DataFeeder converts the data that returned by a reader into a data
    structure that can feed into Executor and ParallelExecutor. The reader
C
chengduoZH 已提交
87
    usually returns a list of mini-batch data entries. Each data entry in
C
chengduoZH 已提交
88 89
    the list is one sample. Each sample is a list or a tuple with one
    feature or multiple features.
C
chengduoZH 已提交
90 91 92 93 94 95

    The simple usage shows below:

    ..  code-block:: python

        place = fluid.CPUPlace()
C
chengduoZH 已提交
96
        img = fluid.layers.data(name='image', shape=[1, 28, 28])
C
chengduoZH 已提交
97
        label = fluid.layers.data(name='label', shape=[1], dtype='int64')
C
chengduoZH 已提交
98 99
        feeder = fluid.DataFeeder([img, label], fluid.CPUPlace())
        result = feeder.feed([([0] * 784, [9]), ([1] * 784, [1])])
C
chengduoZH 已提交
100 101 102 103 104 105 106 107 108 109 110 111 112 113 114


    If you want to feed data into GPU side separately in advance when you
    use multi-GPU to train a model, you can use `decorate_reader` function.

    ..  code-block:: python

        place=fluid.CUDAPlace(0)
        feeder = fluid.DataFeeder(place=place, feed_list=[data, label])
        reader = feeder.decorate_reader(
            paddle.batch(flowers.train(), batch_size=16))

    Args:
        feed_list(list): The Variables or Variables'name that will
            feed into model.
C
chengduoZH 已提交
115 116 117 118
        place(Place): place indicates feed data into CPU or GPU, if you want to
            feed data into GPU, please using `fluid.CUDAPlace(i)` (`i` represents
            the GPU id), or if you want to feed data into CPU, please using
            `fluid.CPUPlace()`.
C
chengduoZH 已提交
119 120 121 122
        program(Program): The Program that will feed data into, if program
            is None, it will use default_main_program(). Default None.

    Raises:
C
chengduoZH 已提交
123
        ValueError: If some Variable is not in this Program.
C
chengduoZH 已提交
124 125 126 127 128 129 130 131

    Examples:
        .. code-block:: python

            # ...
            place = fluid.CPUPlace()
            feed_list = [
                main_program.global_block().var(var_name) for var_name in feed_vars_name
C
chengduoZH 已提交
132
            ] # feed_vars_name is a list of variables' name.
C
chengduoZH 已提交
133 134 135 136 137 138
            feeder = fluid.DataFeeder(feed_list, place)
            for data in reader():
                outs = exe.run(program=main_program,
                               feed=feeder.feed(data))
    """

F
fengjiayi 已提交
139
    def __init__(self, feed_list, place, program=None):
Y
Yu Yang 已提交
140 141 142 143
        self.feed_dtypes = []
        self.feed_names = []
        self.feed_shapes = []
        self.feed_lod_level = []
F
fengjiayi 已提交
144 145
        if program is None:
            program = default_main_program()
Y
Yu Yang 已提交
146
        for each_var in feed_list:
147
            if isinstance(each_var, six.string_types):
F
fengjiayi 已提交
148
                each_var = program.block(0).var(each_var)
Y
Yu Yang 已提交
149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167
            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)
            shape = each_var.shape
            batch_size_dim = -1
            for i, s in enumerate(shape):
                if s < 0:
                    batch_size_dim = i
                    break
            if batch_size_dim == -1:
                raise ValueError("Variable {0} must has a batch size dimension",
                                 each_var.name)
            self.feed_lod_level.append(each_var.lod_level)
            self.feed_shapes.append(shape)

        self.place = place

    def feed(self, iterable):
C
chengduoZH 已提交
168
        """
C
chengduoZH 已提交
169 170
        According to feed_list and iterable, converters the input into
        a data structure that can feed into Executor and ParallelExecutor.
C
chengduoZH 已提交
171 172 173 174 175 176 177

        Args:
            iterable(list|tuple): the input data.

        Returns:
            dict: the result of conversion.
        """
Y
Yu Yang 已提交
178
        converter = []
179
        for lod_level, shape, dtype in six.moves.zip(
Y
Yu Yang 已提交
180 181 182 183 184 185 186 187 188
                self.feed_lod_level, self.feed_shapes, self.feed_dtypes):
            converter.append(
                DataToLoDTensorConverter(
                    place=self.place,
                    lod_level=lod_level,
                    shape=shape,
                    dtype=dtype))

        for each_sample in iterable:
189 190 191
            assert len(each_sample) == len(converter), (
                "The number of fields in data (%s) does not match " +
                "len(feed_list) (%s)") % (len(each_sample), len(converter))
192 193
            for each_converter, each_slot in six.moves.zip(converter,
                                                           each_sample):
Y
Yu Yang 已提交
194 195
                each_converter.feed(each_slot)
        ret_dict = {}
196 197
        for each_name, each_converter in six.moves.zip(self.feed_names,
                                                       converter):
Y
Yu Yang 已提交
198 199
            ret_dict[each_name] = each_converter.done()
        return ret_dict
Y
yuyang18 已提交
200 201

    def feed_parallel(self, iterable, num_places=None):
C
chengduoZH 已提交
202 203
        """
        Takes multiple mini-batches. Each mini-batch will be feed on each
C
chengduoZH 已提交
204
        device in advance.
C
chengduoZH 已提交
205 206 207

        Args:
            iterable(list|tuple): the input data.
C
chengduoZH 已提交
208
            num_places(int): the number of devices. Default None.
C
chengduoZH 已提交
209 210 211 212 213 214 215

        Returns:
            dict: the result of conversion.

        Notes:
            The number of devices and number of mini-batches must be same.
        """
Y
yuyang18 已提交
216 217 218
        if isinstance(self.place, core.CUDAPlace):
            places = [
                core.CUDAPlace(i)
219 220
                for i in six.moves.xrange(
                    self._get_number_of_places_(num_places))
Y
yuyang18 已提交
221 222 223 224
            ]
        else:
            places = [
                core.CPUPlace()
225 226
                for _ in six.moves.xrange(
                    self._get_number_of_places_(num_places))
Y
yuyang18 已提交
227 228 229 230 231 232 233 234 235
            ]

        if len(iterable) != len(places):
            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.")

        place = self.place
236
        for p, batch in six.moves.zip(places, iterable):
Y
yuyang18 已提交
237 238 239 240 241 242 243 244 245 246
            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):
            return core.get_cuda_device_count()
        else:
C
chengduoZH 已提交
247 248 249
            cpu_num = int(
                os.environ.get('CPU_NUM', multiprocessing.cpu_count()))
            return cpu_num
Y
yuyang18 已提交
250 251 252 253 254 255

    def decorate_reader(self,
                        reader,
                        multi_devices,
                        num_places=None,
                        drop_last=True):
C
chengduoZH 已提交
256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273
        """
        Converter the input data into a data that returned by reader into
        multiple mini-batches. Each mini-batch will be feed on each device.

        Args:
            reader(fun): the input data.
            multi_devices(bool): the number of places. Default None.
            num_places(int): the number of places. Default None.
            drop_last(bool): the number of places. Default None.

        Returns:
            dict: the result of conversion.

        Raises:
            ValueError: If drop_last is False and the data batch which cannot
            fit for devices.
        """

Y
yuyang18 已提交
274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292
        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 "
                        "not implemented")

        return __reader_creator__