o * i>\@sddlmZddlZddlZddlZddlZddlZddlZddlm Z m Z m Z m Z m Z mZmZddlZddlmZmZmZddlmZmZddlmZmZmZdd lmZmZdd lm Z m!Z!m"Z"e rddl#Z#dd l$m%Z%m&Z&ddl'mZ(dd lm)Z)dd l*m+Z+ddl,m-Z-ddl.m/Z/ddl0m1Z1edZ2edZ3Gddde Z4dZ5da6da7da8d"ddZ9ddZ:ddZ;GdddZrrDrange)r+dataset batch_sizeZ num_samples sub_datasetr)r)r,get_sub_datasetszAuToTune.get_sub_datasetcCst|j}|jjj}t|jjtjjr:|jjj}| ||}tjj|||jjj |jjj |jjj |jjj d|_|St|jjtjjr^|jjjj}| ||}tjj|||jjj d|_|Sd}|S)NrhriZ num_replicasZrankshuffle drop_last)rhrirn)copyrM batch_samplerrir@paddleiorrhrkZnranksZ local_rankrmrnrZsamplerZ data_source)r+rMrirhrjr)r)r,rVs6       zAuToTune.get_autotune_loadercCsg}d}t}t|D]\}}|t|t}q t|dkr6t|ddt|dd}|St|ddt|dd}|S)Nrr )rW enumeraterFr>sum)r+readerZcostsr_startidatar)r)r,r\s    zAuToTune.evaluate_reader_costc Csd}|d}d}||krF|dkrF||j_||}tdt|dt||d7}||d|kr6|S|d7}|d9}||krF|dks|S)Nrrzfor back num_workers: rSgffffff?g?)rMrUr\rXrYrZ) r+ruZ best_workersZ best_timeZnum_work_boundarysteprUboundaryrWr)r)r,r]s, zAuToTune.is_bestN) r6r7r8rQr-rTrkrVr\r]r)r)r)r,rLs: rLc@seZdZUdZded<ded<ded<ded<d ed <d ed <d ed<ded<ded<ded<ded<ded<ded<ded<ded<               d0d1d'd(Zd2d)d*Zd3d,d-Zd3d.d/ZdS)4 DataLoadera! DataLoader provides an iterator which iterates given dataset once by the batch_sampler. DataLoader supports single-process and multi-process data loading, multi-process workers will be used to load data asynchronously if :attr:`num_workers` is set as a positive number. DataLoader supports map-style dataset and iterable-style dataset. For map-style dataset(can get a sample from dataset with a given index), please see :code:`paddle.io.Dataset`. For iterable-style dataset(get samples from dataset iteratively, like a Python iterator), please see :code:`paddle.io.IterableDataset`. For :code:`batch_sampler` please see :code:`paddle.io.BatchSampler` Notes: GPU tensor operation is not supported in subprocess currently, please don't use GPU tensor operations in pipeline which will be performed in subprocess, such as dataset transforms, collate_fn, etc. Numpy array and CPU tensor operation is supported. **Disable automatic batching** In certain cases such as some NLP tasks, instead of automatic batching, handling batching manually in dataset is needed by users. For these cases, automatic batching is disabled if both :attr:`batch_size` and :attr:`batch_sampler` is set as None, each data got from :attr:`dataset` should be batched data and will be processed with function define by :attr:`collate_fn` or :attr:`default_collate_fn`. Notes: When automatic batching is disabled, :attr:`default_collate_fn` will do nothing to data from dataset. Args: dataset(Dataset): the dataset to load data from, should be an instance of subclass of :code:`paddle.io.Dataset` or :code:`paddle.io.IterableDataset`. feed_list (list(Tensor)|tuple(Tensor)|None, optional): feed Tensor list. The Tensors should be created by :code:`paddle.static.data()`. :attr:`feed_list` must be set if :attr:`return_list` is False. Default None. places(list(Place)|tuple(Place)|list(str)|None, optional): a list of Place, to put data onto, :attr:`places` can be None, if :attr:`places` is None, default place(CPUPlace or CUDAPlace(0)) will be used. Default None. If ``places`` is list of string, the string in the list can be ``cpu``, ``gpu:x`` and ``gpu_pinned``, where ``x`` is the index of the GPUs. return_list (bool, optional): whether the return value on each device is presented as a list. If :attr:`return_list=False`, the return value on each device would be a dict of str -> Tensor, where the key of the dict is the name of each fed Tensors. If :attr:`return_list=True`, the return value on each device would be a list(Tensor). :attr:`return_list` can only be True in dynamic graph mode. Default True. batch_sampler(BatchSampler|None, optional): an instance of `paddle.io.BatchSampler` to generate batch indices to draw samples from :attr:`dataset` and combine a batch. Default None. batch_size(int|None, optional): sample number in a mini-batch, a substitution parameter for :attr:`batch_sampler`, if :attr:`batch_sampler` is not set, a default `paddle.io.BatchSampler` will be used and initialize by :attr:`batch_size`, :attr:`shuffle` and :attr:`drop_last`. Default 1. shuffle(bool, optional): whether to shuffle indices order before generate batch indices, a substitution parameter for :attr:`batch_sampler` see :attr:`batch_size`. Default False. drop_last(bool, optional): whether drop the last incomplete batch dataset size is not divisible by the batch size, a substitution parameter for :attr:`batch_sampler`, see :attr:`batch_size`. Default False collate_fn(Callable|None, optional): function to generate mini-batch data by merging the sample list, None for only stack each fields of sample in axis 0(same as :attr::`np.stack(..., axis=0)`). Default None num_workers(int, optional): the number of subprocess to load data, 0 for no subprocess used and loading data in main process. Default 0 use_buffer_reader (bool, optional): whether to use buffered reader. If use_buffer_reader=True, the DataLoader would prefetch batch data asynchronously, so it would speed up data feeding and occupies a little more CPU or GPU memory, i.e., the memory of one batch input data. Default True. prefetch_factor (int, optional): Number of batch data the DataLoader would prefetch if use_buffer_reader=True. Default 2. use_shared_memory (bool, optional): whether to use shared memory to speed up putting data into inter-process queue, set :attr:`use_shared_memory` as True only when the shared memory space on your machine(e.g. space of '/dev/shm' on Linux operating system) is large enough. Shared memory will only be enabled in multi-process mode(num_workers > 0). Default True. timeout(int, optional): the timeout value for getting data form output queue of subprocesses. Default 0. worker_init_fn(Callable|None, optional): init function which will be called with worker id on each subprocess starting if not set as None. Default None. persistent_workers(bool, optional): whether to keep the workers in the DataLoader. Default False. Returns: DataLoader: an iterable object for data iterating, each element of the generated data is a Tensor. Examples: .. code-block:: python >>> # doctest: +SOLO('can not use multiprocessing testing `paddle.io.DataLoader`') >>> import numpy as np >>> import paddle >>> import paddle.nn as nn >>> import paddle.nn.functional as F >>> from paddle.io import Dataset, BatchSampler, DataLoader >>> BATCH_NUM = 20 >>> BATCH_SIZE = 16 >>> EPOCH_NUM = 4 >>> IMAGE_SIZE = 784 >>> CLASS_NUM = 10 >>> # define a random dataset >>> class RandomDataset(Dataset): # type: ignore[type-arg] ... def __init__(self, num_samples): ... self.num_samples = num_samples ... ... def __getitem__(self, idx): ... image = np.random.random([IMAGE_SIZE]).astype('float32') ... label = np.random.randint(0, CLASS_NUM - 1, (1, )).astype('int64') ... return image, label ... ... def __len__(self): ... return self.num_samples ... >>> dataset = RandomDataset(BATCH_NUM * BATCH_SIZE) >>> class SimpleNet(nn.Layer): ... def __init__(self): ... super().__init__() ... self.fc = nn.Linear(IMAGE_SIZE, CLASS_NUM) ... ... def forward(self, image, label=None): ... return self.fc(image) ... >>> simple_net = SimpleNet() >>> opt = paddle.optimizer.SGD(learning_rate=1e-3, ... parameters=simple_net.parameters()) ... >>> loader = DataLoader(dataset, ... batch_size=BATCH_SIZE, ... shuffle=True, ... drop_last=True, ... num_workers=2) ... >>> for e in range(EPOCH_NUM): ... for i, (image, label) in enumerate(loader()): ... out = simple_net(image) ... loss = F.cross_entropy(out, label) ... avg_loss = paddle.mean(loss) ... avg_loss.backward() ... opt.minimize(avg_loss) ... simple_net.clear_gradients() ... print("Epoch {} batch {}: loss = {}".format(e, i, np.mean(loss.numpy()))) Notes: For reading iterable dataset with multiprocess Dataloader, please see :code:`paddle.io.IterableDataset` rA return_list_CollateFn | None collate_fnuse_buffer_readerintprefetch_factorCallable[[int], None] | Noneworker_init_fn Dataset[Any]rhSequence[Tensor] | None feed_listz list[_Place]rGrUr dataset_kinduse_shared_memorytimeoutz.BatchSampler | _InfiniteIterableSampler | Nonerprnauto_collate_batchNTrFrr &PlaceLike | Sequence[PlaceLike] | NoneBatchSampler | Nonerirmpersistent_workersr%NonecCs||_| |_| |_| |_||_||_|sts|dusJd||_|dur)t}t |t t fr5t |}nt |}t||_| dksFJd| dkr[tjdksTtjdkr[tdd} | |_| dksfJd| |_| rr| dkrrd|_|dkszJd ||_t |trtj|_|rtd ||durtd ntj|_|dur|d kr|s|rJd ||_d|_n+|durd|_d|_n |dksJd||_t |trt|||_n t ||||d|_t!j"j#j$%rFt!j"j&j'(}|durt!j")}t!j"j*t t+d|dgd}d|j,vrtd|-dt!j".}|/d}t0|jj||_t |jtr2d}d}n|jj1}|jj2}t3||j||||d|_||_2|jdu|_4d|_5trbt6dur^dnt6|_5||_7d|_8t9|:|_dS)Nz.feed_list should be set when return_list=Falserz*num_workers should be a non-negative valuer`razDataLoader with multi-process mode is not supported on MacOs and Windows currently. Please use single-process mode with num_workers = 0 insteadz*prefetch_factor should be a positive valueFz&timeout should be a non-negative valuez5IterableDataset not support shuffle, but got shuffle=z0IterableDataset expect unspecified batch_samplerrzJbatch_size/shuffle/drop_last should not be set when batch_sampler is givenzMbatch_size should be None or a positive value when batch_sampler is not given)rhrirmrnZdp) dim_namesz;Auto-DP mode requires the mesh to include a 'dp' dimension.rlT);r}rrrrrhrrr r@rDrEr r rKrGrbrcwarningswarnrUrrrrITERr ValueErrorZMAPrprirrrq distributedZ auto_parallelZ auto_dp_utilsZin_auto_dp_modeZfleetautoZget_meshZget_world_sizeZ ProcessMeshrgrZget_rank_by_dim_and_process_idZget_rankZ get_dim_sizerrmrnrrZ pin_memoryrC_persistent_workers _iteratorrLr-)r+rhrrGr}rprirmrnrrUrrrrrrZmeshZ word_sizeZdp_rankZ dp_world_sizer)r)r,rQs               zDataLoader.__init__cCs.|jtjkr td|jrt|jSt|jS)Nz'length of IterableDataset not supported)rrrrrr>rprhrdr)r)r,__len__Zs   zDataLoader.__len__rcCsJ|jdkr t|S|jr!|jdurt||_|jS|j|jSt|S)Nr)rUrrrr_resetrdr)r)r,__iter__cs    zDataLoader.__iter__cCs|Sr()rrdr)r)r,r-oszDataLoader.__call__)NNTNrFFNrTr TrNF)"rhrrrrGrr}rArprrirrmrArnrArr~rUrrrArrrrArrrrrrAr%r)r%r)r%r) r6r7r8__doc____annotations__rQrrr-r)r)r)r,r| sL *    r|)r:)> __future__rrorXrNrbrWrtypingrrrrrrr rqZbase.frameworkr r r Z frameworkrrZ dataloaderrrrZdataloader.batch_samplerrrZdataloader.dataloader_iterrrrnumberscollections.abcrrZ numpy.typingZnptrZpaddle._typingrZpaddle._typing.device_likerZ$paddle.io.dataloader.dataloader_iterrZpaddle.io.dataloader.datasetrr r!r"ZQUEUE_GET_TIMEOUTr?r;r<r=rCrKrLr|r)r)r)r,sH $