zj ^ddlmZddlZddlZddlZddlZddlZddlZddlm Z m Z m Z m Z m Z mZmZddlZddlmZmZmZddlmZmZddlmZmZmZdd lmZmZdd lm Z m!Z!m"Z"e rTddl#Z#dd l$m%Z%m&Z&ddl'mZ(dd lm)Z)dd l*m+Z+ddl,m-Z-ddl.m/Z/ddl0m1Z1edZ2edZ3Gdde Z4dZ5da6da7da8ddZ9dZ:dZ;GddZ<GddZ=dS) ) annotationsN) TYPE_CHECKINGAnyAnyStrCallableProtocolTypeVaroverload)_current_expected_place_get_paddle_place_get_paddle_place_list)corein_dynamic_mode) BatchSamplerIterableDatasetSubset)DistributedBatchSampler_InfiniteIterableSampler)_DataLoaderIterMultiProcess_DataLoaderIterSingleProcess _DatasetKind)MappingSequence)Tensor) PlaceLike)_Place)_DataLoaderIterBase)Dataset_K_VceZdZeddZeddZedd ZeddZeddZdS) _CollateFnbatch5Sequence[npt.NDArray[Any]] | Sequence[numbers.Number]returnnpt.NDArray[Any]cdSNselfr%s `/lsinfo/ai/hellotax_ai/data_center/backend/venv/lib/python3.11/site-packages/paddle/io/reader.py__call__z_CollateFn.__call__Cs  #sSequence[Tensor]rcdSr*r+r,s r.r/z_CollateFn.__call__H?Bsr0Sequence[AnyStr]rcdSr*r+r,s r.r/z_CollateFn.__call__Kr3r0Sequence[Mapping[_K, _V]]Mapping[_K, _V]cdSr*r+r,s r.r/z_CollateFn.__call__Ns "cr0Sequence[Sequence[_V]] Sequence[_V]cdSr*r+r,s r.r/z_CollateFn.__call__SsKN3r0N)r%r&r'r()r%r1r'r)r%r4r'r)r%r6r'r7)r%r9r'r:)__name__ __module__ __qualname__r r/r+r0r.r$r$Bs  # # #  # BBB B BBB B  " " "  " NNN NNNr0r$<Fc|a|adSr*) USE_AUTOTUNE TUNING_STEPS) use_autotune tuning_stepss r.set_autotune_configrFasLLLLr0ct|dkrtSt|dkrt|dtsJ|dadS)Nrr)lenUSE_PINNED_MEMORY isinstancebool)argss r.use_pinned_memoryrMhsM 4yyA~~  4yyA~~*T!Wd";";~~; Gr0ct|ttfs|g}g}|D][}t|tjs*tj}|||}||\|Sr*)rJlisttuplerPlace set_placeappend)placesretptmps r._convert_placesrXqs ftUm , , C !TZ(( *,,C MM!   A 1 Jr0c8eZdZdZdZdZdZdZdZdZ dS) AuToTunecJ||_tjdz |_dS)Nr )loadermultiprocessing cpu_countmax_num_worker)r-r\s r.__init__zAuToTune.__init__s% -799A=r0ctr|s |jjS|}| |jjSt j}t jdt jdt|jjzd}td}t jdt|j zd}||j kr||_| |}|dz|kr|}|}n&| ||||j }||krnI|}t jdt|zdzt|z|d z }||j kt j d t|zt jd tt j|z zd z|S) Nz(========= DataLoader Auto Tune =========zUser config for DataLoader: rinfz"Tuning Range for num_workers: 0 ~ g?z num_workers: avg_cost: r z'auto_tune dataLoader best_num_workers: z AutoTuning Cost for DataLoader: z seconds)rB need_autotuner\ num_workersget_autotune_loadertimeloggingdebugstrfloatr_evaluate_reader_costis_bestinfo)r-auto_tune_loaderauto_tune_startbest_num_workersmin_costreavg_cost update_nums r.r/zAuToTune.__call__s +d&8&8&:&: +;* * 3355  #;* *)++ @AAA *S1H-I-I I   << 03t7J3K3K K    D///+6  (001ABBH$))##.  !\\$$'  !111'1$ Mk""# h--     1 K/D///0  5'>54r0ct|tzt|}t|t t |}|S)N)indices)minrCrHrrOrange)r-dataset batch_size num_samples sub_datasets r.get_sub_datasetzAuToTune.get_sub_datasetsD*|3S\\BB Wd53E3E.F.FGGG r0ctj|j}|jjj}t |jjt jjr|jjj}| ||}t j|||jjj |jjj |jjj |jjj |_nt |jjt jjrc|jjjj}| ||}t j|||jjj |_nd}|S)Nrr num_replicasrankshuffle drop_last)rrr)copyr\ batch_samplerrrJpaddleiorrrnranks local_rankrrrsampler data_source)r-r\rrrs r.rfzAuToTune.get_autotune_loaders<4;''[.9  K %vy'H   k/7G..w CCK#)9#D#D#%![6=[.9 19+3= $E$$F  1693I J J k/7CG..w CCK#)9#9#9#%+3=$:$$F F r0cg}d}tj}t|D]A\}}|tj|z tj}Bt|dkr0t |ddt|ddz }n/t |ddt|ddz }|S)Nrr )rg enumeraterSrHsum)r-readercostsrsstartidatas r.rlzAuToTune.evaluate_reader_costs  ((  GAt LLu, - - -IKKEE u::>>59~~E!""I6HH59~~E!""I6Hr0c2d}|dz}d}||kr|dkr||j_||}tjdt |zdzt |z|dz }||dz|zkr|S|dz }|dz}||kr|dk|S)Nrrzfor back num_workers: rcgffffff?g?)r\rerlrhrirj) r-r best_workers best_timenum_work_boundarystepreboundaryrgs r.rmzAuToTune.is_bests"Q& ---$((&1DK #,,V44D M(k""# d))    AIDi$&111""q  H---$((r0N) r<r=r>r`r/rdrrfrlrmr+r0r.rZrZs~>>>8 8 8 t  8   r0rZceZdZUdZded<ded<ded<ded<ded <d ed <d ed <ded<ded<ded<ded<ded<ded<ded<ded<ded< d-d.d(Zd/d)Zd0d+Zd0d,ZdS)1 DataLoaderal" 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. reader_buffer_size (int, optional): This option takes effect only when use_buffer_reader is set to True. It specifies the number of batches the buffer reader prefetches in advance. Note that Increasing this value will result in a linear increase in CPU or GPU memory usage. Default 2. 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` rK return_list_CollateFn | None collate_fnuse_buffer_readerintreader_buffer_sizeprefetch_factorCallable[[int], None] | Noneworker_init_fn Dataset[Any]rSequence[Tensor] | None feed_listz list[_Place]rTrer dataset_kinduse_shared_memorytimeoutz.BatchSampler | _InfiniteIterableSampler | Nonerrauto_collate_batchNTrFrr &PlaceLike | Sequence[PlaceLike] | NoneBatchSampler | Nonerrpersistent_workersr'Nonecp||_| |_| |_| |_| |_||_||_|sts | Jd||_|t}t|ttfrt|}nt|}t||_| dks Jd| dkr6t"jdkst"jdkrt'jdd} | |_| dks Jd||_|r | dkrd|_|dks Jd ||_t|t0r7t2j|_|rt9d ||t9d nt2j|_|#|d kr|s|r Jd ||_d|_nk|d|_d|_nZ|dks Jd||_t|t0rtA|||_ntC|||||_tDj#j$j%&r[tDj#j'j()}|[tDj#*}tDj#+ttYd|dg}d|j-vrt9d|.dtDj#/}|0d}tc|jj|z |_t|jt@rd}d}n|jj2}|jj3}ti||j|||||_||_3|jdu|_5d|_6tr#todn to|_6||_8d|_9tu|;|_dS)Nz.feed_list should be set when return_list=Falserz*num_workers should be a non-negative valuervrwzDataLoader 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)rrrrdp) dim_namesz;Auto-DP mode requires the mesh to include a 'dp' dimension.rT),..F ftUm , , /+F33FF&v..F%f-- a!M ?? LH $ $ (?(? MO   K&"""$P"""!2  +!1!1%*D "!|||E||| g / / 1 , 1D   UGUU( F) !- 0D   $??7?9??)#?D"/D "DOO  !%D "DOO>>>-">>)DO'?33 %=Z&&""&2#)#' &&&"   + 9 I I K K %+099;;D|".==?? )55q),,--$64>)) Q99f(1133G!--d33M!$"4"?-"OPPDO$,.FGG 9! ,4 .8 !8?*# """D #"&"4D"@    )++39J9L9L O$6 #D>>2244r0c|jtjkrtd|jrt |jSt |jS)Nz'length of IterableDataset not supported)rrrrrrHrrrzs r.__len__zDataLoader.__len__bsN   1 1 1FGG G& )4-...4<(((r0rc|jdkrt|S|jr<|jt ||_n|j|jSt |S)Nr)rerrrr_resetrzs r.__iter__zDataLoader.__iter__ksl  q /55 5  % 5~%! !.t44 4r0c*|Sr*)rrzs r.r/zDataLoader.__call__ws}}r0)NNTNrFFNrTr r TrNF)$rrrrrTrrrKrrrrrrKrrKrrrerrrKrrrrrrKrrrrrrKr'r)r'r)r'r) r<r=r>__doc____annotations__r`rrr/r+r0r.rr slll\!!!!0000&&&&LLLAAAAOOO .29= -1(,"&"# "&7;#(%W5W5W5W5W5r)))) 5 5 5 5r0r)r@)> __future__rrrhr]rxrgrtypingrrrrrr r rbase.frameworkr r r frameworkrr dataloaderrrrdataloader.batch_samplerrrdataloader.dataloader_iterrrrnumberscollections.abcrr numpy.typingnptrpaddle._typingrpaddle._typing.device_liker$paddle.io.dataloader.dataloader_iterrpaddle.io.dataloader.datasetr r!r"r$QUEUE_GET_TIMEOUTrIrBrCrFrMrXrZrr+r0r.rs#"""""    .-------========== !ONNN11111111((((((111111HHHHHH444444 B BOOOOOXOOO.      $$$   FFFFFFFFRoooooooooor0