a 7Sic+@sUdZddlZddlZddlZddlZddlZddlZddlZddlZddl m Z ddl m Z m Z mZmZmZmZmZmZmZddlZddlZddlmZddlmZddlZddlmZddlm Z ddl!m"Z"m#Z#m$Z$m%Z%m&Z&m'Z'm(Z(m)Z)dd l*m+Z+m,Z,dd l!m-Z-gd Z.ed d dZ/edZ0e e1gdfZ2e ee0ge fZ3e-j4j5Z5e3e6d<e-j4j7Z7e-j8j9Z9e:e;ZGddde%Z?ddZ@ddZAGdddee/ZBGddde=ZCGdddeCZDGdd d eCZEdS)!aDefinition of the DataLoader and associated iterators that subclass _BaseDataLoaderIter To support these two classes, in `./_utils` we define many utility methods and functions to be run in multiprocessing. E.g., the data loading worker loop is in `./_utils/worker.py`. N) timedelta) AnyCallableIterableTypeVarGenericSequenceListOptionalUnion)ExceptionWrapper)string_classes) IterDataPipe MapDataPipeIterableDatasetSamplerSequentialSampler RandomSampler BatchSamplerDataset)!_IterDataPipeSerializationWrapper _MapDataPipeSerializationWrapper)_utils) DataLoaderget_worker_infodefault_collatedefault_convertT_coT) covariantTrc@s eZdZdZdZeddZdS) _DatasetKindrrcCs2|tjkrtj||||Stj||||SdSN)r!Maprfetch_MapDatasetFetcher_IterableDatasetFetcher)kinddatasetauto_collation collate_fn drop_lastr,W/var/www/html/django/DPS/env/lib/python3.9/site-packages/torch/utils/data/dataloader.pycreate_fetcherMs z_DatasetKind.create_fetcherN)__name__ __module__ __qualname__r#r staticmethodr.r,r,r,r-r!Isr!cs(eZdZdZfddZddZZS)_InfiniteConstantSamplerzAnalogous to ``itertools.repeat(None, None)``. Used as sampler for :class:`~torch.utils.data.IterableDataset`. Args: data_source (Dataset): dataset to sample from cstt|ddSr")superr3__init__self __class__r,r-r5]sz!_InfiniteConstantSampler.__init__ccs dVqdSr"r,r6r,r,r-__iter__`sz!_InfiniteConstantSampler.__iter__)r/r0r1__doc__r5r: __classcell__r,r,r8r-r3Us r3cCs(tr tr ttfSdSdS)N)rr)dist is_availableis_initializedget_world_sizeget_rankr,r,r,r-_get_distributed_settingsesrBcCsX|}tjj}|j}|j}||9}|||}tjjj||||durT||dSr")torchutilsdatar num_workersr(graph_settingsapply_sharding)worker_init_fn world_sizeZrank_id worker_idZglobal_worker_idinfoZ total_workersdatapiper,r,r-_sharding_worker_init_fnls  rNcsZeZdZUdZeeed<eeed<eed<e ed<e ed<e ed<e e e fed<eed <eed <ed ed <d Zd+dd ddeeeeee e e e dfe e ee edfeeee e e eeee edddZd dddZeddZejddZfddZd ddd Zed!d"Zed#d$Zedd%d&Zd'd(Zd)d*ZZS),ra Data loader. Combines a dataset and a sampler, and provides an iterable over the given dataset. The :class:`~torch.utils.data.DataLoader` supports both map-style and iterable-style datasets with single- or multi-process loading, customizing loading order and optional automatic batching (collation) and memory pinning. See :py:mod:`torch.utils.data` documentation page for more details. Args: dataset (Dataset): dataset from which to load the data. batch_size (int, optional): how many samples per batch to load (default: ``1``). shuffle (bool, optional): set to ``True`` to have the data reshuffled at every epoch (default: ``False``). sampler (Sampler or Iterable, optional): defines the strategy to draw samples from the dataset. Can be any ``Iterable`` with ``__len__`` implemented. If specified, :attr:`shuffle` must not be specified. batch_sampler (Sampler or Iterable, optional): like :attr:`sampler`, but returns a batch of indices at a time. Mutually exclusive with :attr:`batch_size`, :attr:`shuffle`, :attr:`sampler`, and :attr:`drop_last`. num_workers (int, optional): how many subprocesses to use for data loading. ``0`` means that the data will be loaded in the main process. (default: ``0``) collate_fn (callable, optional): merges a list of samples to form a mini-batch of Tensor(s). Used when using batched loading from a map-style dataset. pin_memory (bool, optional): If ``True``, the data loader will copy Tensors into device/CUDA pinned memory before returning them. If your data elements are a custom type, or your :attr:`collate_fn` returns a batch that is a custom type, see the example below. drop_last (bool, optional): set to ``True`` to drop the last incomplete batch, if the dataset size is not divisible by the batch size. If ``False`` and the size of dataset is not divisible by the batch size, then the last batch will be smaller. (default: ``False``) timeout (numeric, optional): if positive, the timeout value for collecting a batch from workers. Should always be non-negative. (default: ``0``) worker_init_fn (callable, optional): If not ``None``, this will be called on each worker subprocess with the worker id (an int in ``[0, num_workers - 1]``) as input, after seeding and before data loading. (default: ``None``) generator (torch.Generator, optional): If not ``None``, this RNG will be used by RandomSampler to generate random indexes and multiprocessing to generate `base_seed` for workers. (default: ``None``) prefetch_factor (int, optional, keyword-only arg): Number of batches loaded in advance by each worker. ``2`` means there will be a total of 2 * num_workers batches prefetched across all workers. (default: ``2``) persistent_workers (bool, optional): If ``True``, the data loader will not shutdown the worker processes after a dataset has been consumed once. This allows to maintain the workers `Dataset` instances alive. (default: ``False``) pin_memory_device (str, optional): the data loader will copy Tensors into device pinned memory before returning them if pin_memory is set to true. .. warning:: If the ``spawn`` start method is used, :attr:`worker_init_fn` cannot be an unpicklable object, e.g., a lambda function. See :ref:`multiprocessing-best-practices` on more details related to multiprocessing in PyTorch. .. warning:: ``len(dataloader)`` heuristic is based on the length of the sampler used. When :attr:`dataset` is an :class:`~torch.utils.data.IterableDataset`, it instead returns an estimate based on ``len(dataset) / batch_size``, with proper rounding depending on :attr:`drop_last`, regardless of multi-process loading configurations. This represents the best guess PyTorch can make because PyTorch trusts user :attr:`dataset` code in correctly handling multi-process loading to avoid duplicate data. However, if sharding results in multiple workers having incomplete last batches, this estimate can still be inaccurate, because (1) an otherwise complete batch can be broken into multiple ones and (2) more than one batch worth of samples can be dropped when :attr:`drop_last` is set. Unfortunately, PyTorch can not detect such cases in general. See `Dataset Types`_ for more details on these two types of datasets and how :class:`~torch.utils.data.IterableDataset` interacts with `Multi-process data loading`_. .. warning:: See :ref:`reproducibility`, and :ref:`dataloader-workers-random-seed`, and :ref:`data-loading-randomness` notes for random seed related questions. r( batch_sizerF pin_memoryr+timeoutsamplerpin_memory_deviceprefetch_factor_BaseDataLoaderIter _iteratorFrNr)rTpersistent_workersrS)r(rOshufflerR batch_samplerrFr*rPr+rQrIrTrYrScCs$tjd|dkrtd| dkr,td|dkrD|dkrDtd|dksPJ|rd|dkrdtd||_||_||_||_||_| |_ | |_ | |_ t |jt rt|j|_t\}}|dkrtt|j |||_ ntjjj|j||nZt |jtrFt|j|_t\}}|dkr0tt|j |||_ ntjjj|j||t |trtj|_t |t r|durtjjjj||d}n|d vrtd ||durtd |n|durtd |nt|}tj |_|dur|rtd |dur6|dks$|s$|dus$| r,tdd}d} n|durN| rNtd|dur|jtjkrnt!}n|rt"|| d}nt#|}|dur|durt$||| }||_%| |_&||_'||_(| |_)|dur|j*rt+j,j-}nt+j,j.}||_/||_0d|_1d|_2d|_3|4t5ddddS)Nzpython.data_loaderrzXnum_workers option should be non-negative; use num_workers=0 to disable multiprocessing.z%timeout option should be non-negativerWzpprefetch_factor option could only be specified in multiprocessing.let num_workers > 0 to enable multiprocessing.z/persistent_workers option needs num_workers > 0)rZ>FNzXDataLoader with IterableDataset: expected unspecified shuffle option, but got shuffle={}zXDataLoader with IterableDataset: expected unspecified sampler option, but got sampler={}zdDataLoader with IterableDataset: expected unspecified batch_sampler option, but got batch_sampler={}z1sampler option is mutually exclusive with shufflerz[batch_sampler option is mutually exclusive with batch_size, shuffle, sampler, and drop_lastFzVbatch_size=None option disables auto-batching and is mutually exclusive with drop_last generatorTZ DataloaderenabledTrue)6rC_C_log_api_usage_once ValueErrorr(rFrTrPrSrQrImultiprocessing_context isinstancerrrB functoolspartialrNrDrErGrHrrrr!r _dataset_kindZapply_shuffle_settingsformatboolr#r3rrrrOr+rRr[r]_auto_collationrcollaterrr*rY_DataLoader__initialized_IterableDataset_len_calledrVcheck_worker_number_rationality set_vital)r7r(rOrZrRr[rFr*rPr+rQrIrcr]rTrYrSwsrankr,r,r-r5s                        zDataLoader.__init__returncCs&|jdkrt|S|t|SdSNr)rF_SingleProcessDataLoaderIterrn_MultiProcessingDataLoaderIterr6r,r,r- _get_iterators zDataLoader._get_iteratorcCs|jSr")$_DataLoader__multiprocessing_contextr6r,r,r-rcsz"DataLoader.multiprocessing_contextcCs~|durt|jdkrdt|trFt}||vr 0), but got num_workers={}) rFrdr multiprocessingget_all_start_methodsrbrh get_contextpython_multiprocessingcontext BaseContext TypeErrorrx)r7rcZvalid_start_methodsr,r,r-rcs&   cs8|jr"|dvr"td||jjtt|||dS)N)rOr[rRr+r(rYz6{} attribute should not be set after {} is initialized)rlrbrhr9r/r4r __setattr__)r7attrvalr8r,r-rs zDataLoader.__setattr__cCsD|jr8|jdkr8|jdur&||_n |j||jS|SdSrt)rYrFrVrw_resetr6r,r,r-r:s    zDataLoader.__iter__cCs |jduSr")r[r6r,r,r-rjszDataLoader._auto_collationcCs|jr |jS|jSdSr")rjr[rRr6r,r,r-_index_samplerszDataLoader._index_samplercCsd|jtjkrVt|j}|_|jdurRddlm}|j rD||j}n|||j}|St|j SdS)Nr)ceil) rgr!rlenr(rmrOmathrr+r)r7lengthrr,r,r-__len__s    zDataLoader.__len__cCsdd}|jr|jdkrdSd}d}ttdrXzttd}d}WntyVYn0|durtt}|durt|}|durt|||j|dS|j|krt|||j|dS)NcSs0|durd||rdndnd}d||}|S)Nz|Our suggested max number of worker in current system is {}{}, which is smaller than what this DataLoader is going to create.rXz% (`cpuset` is not taken into account)zUDataLoader is not able to compute a suggested max number of worker in current system.zThis DataLoader will create {} worker processes in total. {} Please be aware that excessive worker creation might get DataLoader running slow or even freeze, lower the worker number to avoid potential slowness/freeze if necessary.)rh)Znum_worker_suggestZnum_worker_createdcpuset_checkedZsuggested_max_worker_msgwarn_msgr,r,r-_create_warning_msgs  zGDataLoader.check_worker_number_rationality.._create_warning_msgrFsched_getaffinityT) rFhasattrosrr Exception cpu_countwarningswarn)r7rZmax_num_worker_suggestrrr,r,r-rns8   z*DataLoader.check_worker_number_rationalitycCst|jtrtjdtjdj|jd}t rt rt }t }t j}|dkr2t|}|tj|td|d|tjd}t}||krttj|tjd}tt|dttjdkrtd |d |d tjd q|tjd |tj| }|dksJnd }t}t|dkrttj|tj}tt|dttjdkr>td|d|d tjd q>td|d||tjd}|dkrttj|tjd}qt |}|SdSdS)Nr,dtyper\rz Shared seed (z) sent to store on rank 0r)secondszTimed out receiving the signal from the distribtued store on Rank 0 that all other Ranks have received the shared seed. (world_size=z , received=z , timeout=)rXzGTimed out receiving the shared seed from the distribtued store on Rank z. (world_size=z) received from store on rank )!rdr(rrCemptyint64random_r]itemr=r>r?rAr@distributed_c10d_get_default_storestrsetrDATAPIPE_SHARED_SEEDloggerrLaddDATAPIPE_SHARED_SEED_COUNTERtimesleep#DATAPIPE_SHARED_SEED_CHECK_INTERVALr$DATAPIPE_SHARED_SEED_DEFAULT_TIMEOUT RuntimeErrorrgetint)r7 _shared_seedrqrpstoreZ_shared_seed_strZ_shared_seed_recv_cntstartr,r,r-_get_shared_seed8sf           zDataLoader._get_shared_seed) rNNNrNFFrNNN) r/r0r1r;rr__annotations__r rrifloatr rrrrlr _collate_fn_t_worker_init_fn_tr5rwpropertyrcsetterrr:rjrrrnrr<r,r,r8r-rzs` Q    *     Mrc@sleZdZeddddZddddZdd d Zd d Zd dZe dddZ e Z e dddZ ddZdS)rUN)loaderrscCs*|j|_||_t|jtrHt}||jtj j j |j||_|j |_ |j|_|j|_|j|_|j|_|j|_|j|_t|jdkr|jotj|_d|_n$|jsd}t||j|_|j|_|j |_!|j"|_#t$|j|_%tj&dtj'dj(|j)d*|_+|j,|_-d|_.d/|j0j1|_2dS)Nrzpin memory device is set and pin_memory flag is not used then device pinned memory won't be usedplease set pin_memory to true, if you need to use the device pin memoryr,rr\z!enumerate(DataLoader)#{}.__next__)3r(_datasetrrrdrrC Generator manual_seedrDrErGapply_shuffle_seedrgrmrjr+ _drop_lastrrF _num_workersrT_prefetch_factorrrSrPcudar> _pin_memory_pin_memory_devicerrrQ_timeoutr* _collate_fniter _sampler_iterrrrr]r _base_seedrY_persistent_workers _num_yieldedrhr9r/ _profile_name)r7r shared_rngrr,r,r-r5ms8      z_BaseDataLoaderIter.__init__rrcCs|Sr"r,r6r,r,r-r:sz_BaseDataLoaderIter.__iter__FcCs^t|j|_d|_|j|_||_t|jt rZt }| |jt j jj|j||_dSrt)rrrrrmrrrdrrrCrrrDrErGr)r7r first_iterrr,r,r-rs    z_BaseDataLoaderIter._resetcCs t|jSr")nextrr6r,r,r- _next_indexsz_BaseDataLoaderIter._next_indexcCstdSr")NotImplementedErrorr6r,r,r- _next_datasz_BaseDataLoaderIter._next_datacCstjj|j|jdur$||}|jd7_|j t j kr|j dur|j|j krd |j|j |j}|jdkr|d7}t||WdS1s0YdS)NrzwLength of IterableDataset {} was reported to be {} (when accessing len(dataloader)), but {} samples have been fetched. rzFor multiprocessing data-loading, this could be caused by not properly configuring the IterableDataset replica at each worker. Please see https://pytorch.org/docs/stable/data.html#torch.utils.data.IterableDataset for examples.)rCautogradprofilerrecord_functionrrrrrrgr!rrmrhrrrr)r7rErr,r,r-__next__s$     z_BaseDataLoaderIter.__next__cCs t|jSr")rrr6r,r,r-rsz_BaseDataLoaderIter.__len__cCstd|jjdS)Nz{} cannot be pickled)rr9r/r6r,r,r- __getstate__sz _BaseDataLoaderIter.__getstate__)F)r/r0r1rr5r:rrrrrrrrrr,r,r,r-rUls$ rUcs$eZdZfddZddZZS)rucsNtt|||jdksJ|jdks,Jt|j|j|j |j |j |_ dSrt) r4rur5rrr!r.rgrrjrr_dataset_fetcher)r7rr8r,r-r5s z%_SingleProcessDataLoaderIter.__init__cCs.|}|j|}|jr*tj||j}|Sr")rrr$rrrPr)r7indexrEr,r,r-rs  z'_SingleProcessDataLoaderIter._next_data)r/r0r1r5rr<r,r,r8r-rus rucseZdZdZfddZdfdd ZejfddZd d Z d d Z d dZ ddZ dddZ ddZeddZddZZS)rvzHIterates once over the DataLoader's dataset, as specified by the samplercstt|||jdksJ|jdks,J|jdurXz:_MultiProcessingDataLoaderIter.__init__..)r)8r4rvr5rrrcryrIZ_worker_init_fnQueue_worker_result_queue_worker_pids_set _shutdownEvent_workers_done_event _index_queues_workersrangecancel_join_threadProcessrworker _worker_looprgrrjrrrrrdaemonrappendr threading_pin_memory_thread_done_eventqueue _data_queueThreadrP_pin_memory_looprCrcurrent_devicer_pin_memory_threadatexitregister_clean_up_workersignal_handling_set_worker_pidsidtuple_set_SIGCHLD_handlerr)r7rrci index_queuerZpin_memory_threadrr8r,r-r5 sf          " z'_MultiProcessingDataLoaderIter.__init__Fcst||d|_d|_i|_d|_ddt|jD|_t t|j|_ |st|jD]}|j | tj|jq\|j}|dkr|\}}t|tjjr|dusJ|d8}qt|j|jD] }|qdS)NrcSsg|]}dqS)Tr,rrr,r,r- mrz9_MultiProcessingDataLoaderIter._reset..r)r4r _send_idx _rcvd_idx _task_info_tasks_outstandingrr_workers_status itertoolscycle_worker_queue_idx_cyclerputrr_ResumeIterationr _get_datardr_try_put_index)r7rridxZresume_iteration_cntZ return_idxZ return_data_r8r,r-r]s$   z%_MultiProcessingDataLoaderIter._resetc s>z|jj|d}d|fWSty8}zg}t|jD].\}}|j|r:|s:||||q:t |dkrd dd|D}t d ||t |tjrWYd}~dSddlddl}zd } fd d t| D} Wn>ty }z$|j|jkr t d dWYd}~n d}~00WYd}~n d}~00dS) NrQTrz, css|]}t|jVqdSr")rrrr,r,r-rrz?_MultiProcessingDataLoaderIter._try_get_data..z1DataLoader worker (pid(s) {}) exited unexpectedly)FN csg|] }qSr,)NamedTemporaryFilertempfiler,r-rrz@_MultiProcessingDataLoaderIter._try_get_data..aToo many open files. Communication with the workers is no longer possible. Please increase the limit using `ulimit -n` in the shell or change the sharing strategy by calling `torch.multiprocessing.set_sharing_strategy('file_system')` at the beginning of your code)rrr enumeraterris_aliver_mark_worker_as_unavailablerjoinrrhrdrEmptyrerrnorOSErrorEMFILE) r7rQrEeZfailed_workersrKrZpids_strrZfds_limit_marginfsr,rr- _try_get_data~s6      z,_MultiProcessingDataLoaderIter._try_get_datacCs~|jdkr4||j\}}|r"|Std|jnF|jrd|jrZ|\}}|r:|Sq:tdn|\}}|rd|SqddS)Nrz%DataLoader timed out after {} secondsz%Pin memory thread exited unexpectedly)rrrrhrrr)r7successrEr,r,r-r s     z(_MultiProcessingDataLoaderIter._get_datacCsD|j|jkrR|j|j}|d}t|dksd|j|r8qd|j|j=|jd7_q|js`|tt|j|jdkr|j|jd}| |S|j s|j dksJ| \}}|j d8_ |j tjkr t|tjjr |jrd|j|j<n ||j|q||jkr,|j||f7<q|j|=| |SqdS)NrrWrF)rrrrrr_shutdown_workers StopIterationpop _process_datarrr rgr!rrdrr_IterableDatasetStopIterationrKrr )r7rLrKrEr r,r,r-r1s6       z)_MultiProcessingDataLoaderIter._next_datacCs|j|j|jksJz |}Wnty6YdS0t|jD]}t|j}|j|rBqdqBdS|j | |j |f|f|j |j <|jd7_|j d7_ dSNr) rrrrr rrrrrrrr)r7rr Zworker_queue_idxr,r,r-r bs    z-_MultiProcessingDataLoaderIter._try_put_indexcCs,|jd7_|t|tr(||Sr$)rr rdr reraise)r7rEr,r,r-r"vs  z,_MultiProcessingDataLoaderIter._process_datacCsL|j|s|jr|sJ|j|}|dd|j|<|j|ksHJdS)NF)rrrrris_set)r7rKshutdownqr,r,r-r}s   z:_MultiProcessingDataLoaderIter._mark_worker_as_unavailablec CsZtj}|dus|durdS|jsVd|_zt|drh|j|jd|j |j |j |j t t|jD]"}|js|j|r|j|ddq|jD]}|j tjdq|jD]}| | qW|jrtjt|d|_|jD]}|r|qn@|jr4tjt|d|_|jD]}|r:|q:0dS)NTr)NN)r'rF)rpython_exit_statusrrrrrrrrrcloserrrrrrrMP_STATUS_CHECK_INTERVALrrr_remove_worker_pidsrr terminate)r7r)rKrr(r,r,r-rsB              z0_MultiProcessingDataLoaderIter._shutdown_workerscCs:z"|jtjdW|r6|n|r4|0dS)Nr)rrr+rr-)rr,r,r-rs  z/_MultiProcessingDataLoaderIter._clean_up_workercCs |dSr")rr6r,r,r-__del__sz&_MultiProcessingDataLoaderIter.__del__)F)F)r/r0r1r;r5rrr+rr rr r"rrr2rr.r<r,r,r8r-rvs 7 P!!1 F rv)Fr;rerloggingrrrrrdatetimertypingrrrrrrr r r ryr|rCZtorch.distributed distributedr=torch.multiprocessingtorch.utils.data.graph_settingsZ torch._utilsr Z torch._sixr rXrrrrrrrr#torch.utils.data.datapipes.datapiperrr__all__rr rrrrkrrrrr getLoggerr/robjectr!r3rBrNrrUrurvr,r,r,r-sP ,    (     uZ