a d:@s.ddlmZmZddlmZddlmZmZmZm Z m Z m Z m Z m Z ddlZddlmZmZddlmZddlmZddlmZmZdd lmZmZmZmZmZmZdd l m!Z!dd l"m#Z#dd d dZ$GdddeZ%eedddZ&Gddde%Z'Gddde'Z(Gddde Z)Gddde%Z*dS))ABCabstractmethod)deepcopy)AnyCallableIterableIteratorListOptionalSizedTupleN)apply_to_collectionapply_to_collections) DataLoader)has_len)CombinedLoader CycleIterator)_add_capture_metadata_collate_patch_dataloader_get_iterators"_teardown_dataloader_get_iterators IteratorStateMergedIteratorStatepatch_dataloader_iterator)MisconfigurationException)_fault_tolerant_trainingreturncCsdSNrrrm/var/www/html/stable-diffusion-webui/venv/lib/python3.9/site-packages/pytorch_lightning/utilities/fetching.py_profile_nothing%sr c@sPeZdZdZeedddZeddddZeddd Zeedd d d Z ddd dZ d0e ddddZ e eddddZee dddZee ddddZddddZeeeddddZeedd d!Zeedd"d#Zeeedd$d%Zddd&d'Zddd(d)Zedd*d+Zddd,d-Zddd.d/Z dS)1AbstractDataFetcheraThis base class should be used to implement a fault tolerant ``DataFetcher``. It is required to override the ``fetching_function`` with fetching logic. Example:: class SimpleDataFetcher(AbstractDataFetcher): def fetching_function(self): while True: try: return next(self.dataloader_iter), False except StopIteration: return None, True rcCsdS)z&Override with your own fetching logic.Nrselfrrrfetching_function9sz%AbstractDataFetcher.fetching_functionNcCsdS)z*Override with your own pre-fetching logic.Nrr"rrr prefetching=szAbstractDataFetcher.prefetchingcCsdS)z=Hook to override to handle the logic before fetching a batch.Nrr"rrron_fetch_startAsz"AbstractDataFetcher.on_fetch_startbatch start_outputrcCsdSz>Hook to extend which handles the logic after fetching a batch.Nrr#r(r)rrr on_fetch_endDsz AbstractDataFetcher.on_fetch_endcCsdS)zDHook to override to indicate the `DataFetcher` to wait for an event.Nrr"rrrwaitGszAbstractDataFetcher.waitrprefetch_batchesrcCs>|dkrtd||_d|_d|_d|_d|_t|_t|_dS)Nrz(`prefetch_batches` should at least be 0.F) rr/ _dataloaderdataloader_iterfetcheddoner _start_profiler_stop_profilerr#r/rrr__init__JszAbstractDataFetcher.__init__) dataloaderkwargsrcKs"||||_t|dSr)rr0r_attach_data_fetcher)r#r8r9rrrsetupUs zAbstractDataFetcher.setupcCs$|jdurtd|jjd|jS)N`z6` should have been `setup` with a dataloader iterable.)r0r __class____name__r"rrrr8[s  zAbstractDataFetcher.dataloader)r8rcCs2t|ttfsdSt|tr"|j}t|ttdSr) isinstancerrloadersr r)r8rrrrcs  z1AbstractDataFetcher._add_capture_metadata_collatecs0ttddfdd }tjjttf|dS)N)loaderiteratorrcs<t|tr|j}|j}t|tr8tr8|_t||dSr)r?rrAZ _loader_iterrr_lightning_fetcherr)rArBr"rr_apply_patch_fnns  z9AbstractDataFetcher._apply_patch.._apply_patch_fn)rrrr@ loader_iters)r#rDrr"r _apply_patchms z AbstractDataFetcher._apply_patch)r1dataloader_iter_statesrcCst|ddduri|_t|dddur.t|_|D].}|j}||jvrPg|j|<|j||q2|j|jkr|D]>}t|jrt |j|_ |j}|j| d}|j ||qrdS)N cache_statesstater) getattrrHrrInameappendr2r/lenrprevious_statepopupdate)r#r1rGZ iter_stateZ iter_namerIrrr_store_dataloader_iter_statezs      z0AbstractDataFetcher._store_dataloader_iter_statecCst|jtr|jjS|jSr)r?r8rr@r"rrrr@s zAbstractDataFetcher.loaderscCs,|jdurtdt|jtr&|jjS|jS)NzCThe `dataloader_iter` isn't available outside the __iter__ context.)r1rr?r8rrEr"rrrrEs   z AbstractDataFetcher.loader_iterscCsttddd}t|jt|S)NrBrcSs|jSr)rI)rBrrr collect_statesz0AbstractDataFetcher.state..collect_state)rrr rE)r#rSrrrrIszAbstractDataFetcher.statecs*tddfdd }tjttf|dS)N)rArcs*t|tr|j}t|tr&tr&|_dSr)r?rrArrrC)rAr"rr_attach_data_fetcher_fns zIAbstractDataFetcher._attach_data_fetcher.._attach_data_fetcher_fn)rr r@r)r#rTrr"rr:sz(AbstractDataFetcher._attach_data_fetchercCs(|t|j|_|||Sr)resetiterr8r1rFr%r"rrr__iter__s  zAbstractDataFetcher.__iter__cCs|Sr)r$r"rrr__next__szAbstractDataFetcher.__next__cCsd|_d|_dS)NrF)r2r3r"rrrrUszAbstractDataFetcher.resetcCsF|t|jtr|jt|jtr6t|jd|_tdSr)rUr?r0rrZ$_shutdown_workers_and_reset_iteratorr1rr"rrrteardowns    zAbstractDataFetcher.teardown)r)!r> __module__ __qualname____doc__rrr$r%r&r,r-intr7rr;propertyr8 staticmethodrrFrr rrQr@rErrIr:rWrXrUrYrrrrr!)s:    r!r(rcCs|Srr)r(rrr_no_op_batch_to_devicesracseZdZdZdeeddfdd Zdeee e ge fddfd d Z e d d d Z e e ddddZ dd ddZe d ddZeddddZe e dddZdd fdd ZZS) DataFetcheraoThis class is used to control batch fetching flow. Args: prefetch_batches: Number of batches to pre-fetch. Pre-fetching at least 1 batch is necessary to properly track whether a batch is the last one (available with :attr:`self.done`) under any training setup. store_on_device: Whether to store the pre-fetched batches on device. TN)r/store_on_devicercs*tj|d||_t|_g|_d|_dS)N)r/F)superr7rdrabatch_to_devicebatches_has_len)r#r/rdr=rrr7s zDataFetcher.__init__)r8rfrcs(t|t||_|dur$||_dSr)rer;rrhrf)r#r8rfrirrr;s  zDataFetcher.setuprcCs |dSr)r4r"rrrr&szDataFetcher.on_fetch_startr'cCs||j|dSr*)r5rgrLr+rrrr,szDataFetcher.on_fetch_endc CsT|j}|dusJt|jD]2}z||WqtyLd|_YqPYq0qdS)NT)r1ranger/_fetch_next_batch StopIterationr3)r#rB_rrrr%s  zDataFetcher.prefetchingc Cs|jdusJ|jrP|jd}z||jWqtyL|j |_Yq0nX|jsz||j|jd}Wqty}zd|_|WYd}~qd}~00nt|||S)NrT)r1rgrOrkrlr3r-move_to_device)r#r(errrr$s"   zDataFetcher.fetching_functionrRc Cs|}z t|}Wn0tyD}z||WYd}~n d}~00|jd7_|js|jr|j}t|t stJ|jt |k|_ | ||dS)Nrc) r&nextrlr5r2r/rhr8r?r rMr3r,)r#rBr)r(ror8rrrrks  zDataFetcher._fetch_next_batchr`cCs|jr||}|Sr)rdrfr#r(rrrrn$s zDataFetcher.move_to_devicecstg|_dSr)rerUrgr"rirrrU)s zDataFetcher.reset)rcT)N)r>rZr[r\r]boolr7rr rrr;r&r,r%r$rrkrnrU __classcell__rrrirrbs   rbcspeZdZdZeeddfdd Zeedfdd Zd d d d Zeej j dd ddZ dd ddZ Z S)InterBatchParallelDataFetcheraThis class implements inter-batch parallelism, which aims at hiding the latency of host-to-device copy of input batches behind computationally intensive operations. code-block:: Without parallelization: batch 0: [HtoD][forward][backward] batch 1: [HtoD][forward][backward] batch 2: [HtoD][forward][backward] With parallelization, the latency of HtoD copy can be hidden: batch 0: [HtoD][forward][backward] batch 1: [HtoD] [forward][backward] batch 2: [HtoD] [forward][backward] N)argsr9rcs(tj|i|tj|_g|_dSr)rer7torchcudaZStream cuda_streamevents)r#rur9rirrr7Bs z&InterBatchParallelDataFetcher.__init__r`cs>tj|jt|WdS1s00YdSr)rvrwstreamrxrernrqrirrrnGsz,InterBatchParallelDataFetcher.move_to_deviceztorch.cuda.EventrcCstj}||Sr)rvrwEventr4r#eventrrrr&Ks z,InterBatchParallelDataFetcher.on_fetch_start)r(r}rcCs,||j|||j|dSr)r5rgrLrecordry)r#r(r}rrrr,Qs z*InterBatchParallelDataFetcher.on_fetch_endcCs|jd}|dS)Nr)ryrOr-r|rrrr-Ws z"InterBatchParallelDataFetcher.wait)r>rZr[r\rr7rnr&rvrwr{r,r-rsrrrirrt.s rtc@s0eZdZdZeeddddZedddZdS) StepFuncDataLoaderIterztThis class is a wrapper to keep track of dataloader iterator fetching event while left entirely to user control.N)rB data_fetcherrcCs||_||_dSr)rBr)r#rBrrrrr7bszStepFuncDataLoaderIter.__init__rc Csjz4|jt|j}|j|jjd7_|WStyd}zd|j_|WYd}~n d}~00dS)NrcT)rr4rprBr5r2rlr3)r#datarorrrrXfs   zStepFuncDataLoaderIter.__next__) r>rZr[r\rr!r7rrXrrrrr]srcsNeZdZdZd eddfdd Zdddd Zeeefdd d Z Z S) DataLoaderIterDataFetcheraThis class is used to return directly the `dataloader_iter` to the ``LightningModule`` training_step for users to implement their own pre-fetching logic. This feature can be activated as follows: Example:: Class MyModel(LightningModule): def __init__(self): self.automatic_optimization = False def training_step(self, dataloader_iter: Iterator, batch_idx: int) -> None: # it is the user responsibility to fetch and move the batch to the right device. batch = next(dataloader_iter) batch = batch.to(self.device) ... rNr.cstd|_dS)NF)rer7rdr6rirrr7s z"DataLoaderIterDataFetcher.__init__rcCs&|j}|dusJtt|||_dSr)r1rVrrB)r#rBrrrr%s z%DataLoaderIterDataFetcher.prefetchingcCs|js|j|jfStdSr)r3r2rBrlr"rrrr$s z+DataLoaderIterDataFetcher.fetching_function)r) r>rZr[r\r]r7r%r rr$rsrrrirrrsr)+abcrrcopyrtypingrrrrr r r r rvZ#lightning_utilities.core.apply_funcr rZtorch.utils.data.dataloaderrZlightning_fabric.utilities.datarZ$pytorch_lightning.trainer.supportersrrZ(pytorch_lightning.utilities.auto_restartrrrrrrZ&pytorch_lightning.utilities.exceptionsrZ#pytorch_lightning.utilities.importsrr r!rarbrtrrrrrrs$ (      b/