o ! i"@sddlmZddlmZddlmZmZmZddlm Z ddl Z ddl m Z m Z GdddZddddZGddde ZGdddeZGdddeZGddde ZdS)) annotations) SpawnContext)EmptyFullQueue)ThreadN) DataLoaderDatasetc@s4eZdZdZdddd Zd d Zd d ZddZdS) ThreadBuffera Iterates over values from self.src in a separate thread but yielding them in the current thread. This allows values to be queued up asynchronously. The internal thread will continue running so long as the source has values or until the stop() method is called. One issue raised by using a thread in this way is that during the lifetime of the thread the source object is being iterated over, so if the thread hasn't finished another attempt to iterate over it will raise an exception or yield unexpected results. To ensure the thread releases the iteration and proper cleanup is done the stop() method must be called which will join with the thread. Args: src: Source data iterable buffer_size: Number of items to buffer from the source timeout: Time to wait for an item from the buffer, or to wait while the buffer is full when adding items {Gz? buffer_sizeinttimeoutfloatcCs.||_||_||_t|j|_d|_d|_dSNF)srcr rrbuffer gen_thread is_running)selfrr rrZ/home/dell461/cl/sdc2/last_ska_mid/HISourceFinder-master-l/src/monai/data/thread_buffer.py__init__(s   zThreadBuffer.__init__c CsN|jD]!}|jr!z |jj||jdWn tyYnwn|jsdSqdS)Nr)rrrputrr)rZsrc_valrrrenqueue_values0s  zThreadBuffer.enqueue_valuescCs$d|_|jdur |jd|_dSr)rrjoinrrrrstop<s   zThreadBuffer.stopccsd|_t|jdd|_|jzK|jrH|js|jsOz |jj|j dVWn t y3Ynw|jrV|js|jrW| dSW| dSW| dSW| dS| w)NT)targetdaemonr) rrrrstartis_aliveremptygetrrrrrrr__iter__Ds&     zThreadBuffer.__iter__N)r r )r rrr)__name__ __module__ __qualname____doc__rrrr&rrrrr s   r r r r rrrrepeatsccs2t|||d}|D] }t|D]}|Vqq dS)a Create a ThreadBuffer object using the `src`, `buffer_size`, and `timeout` parameters given for the constructor arguments of the same names, and yield each generated object `repeats` number of times successively. Args: src: Source data iterable buffer_size: Number of items to buffer from the source timeout: Time to wait for an item from the buffer, or to wait while the buffer is full when adding items repeats: Number of repeat generations to perform which is asynchronous from the generation of the next value Returns: Generator yield (repeated) values from `src` asynchronously )rr rN)r range)rr rr+rbatch_rrrbuffer_iteratorSs r/cs,eZdZdZeddZfddZZS)_ProcessThreadzHShim class to make a thread look like a process to the DataLoader class.cCst|SN)idrrrrpidksz_ProcessThread.pidcs0ztWdtjjjj_dSdtjjjj_wr1)superruntorchutilsdata_utilsworker _worker_infor __class__rrr5os "z_ProcessThread.run)r'r(r)r*propertyr3r5 __classcell__rrr<rr0hs  r0c@s eZdZdZddZddZdS) _ProcessQueuezTShim class to make a thread queue look like a process queue to the DataLoader class.cCdSr1rrrrrcloseyz_ProcessQueue.closecCrAr1rrrrrcancel_join_thread|rCz _ProcessQueue.cancel_join_threadN)r'r(r)r*rBrDrrrrr@vs r@c@seZdZdZeZeZdS)_ProcessThreadContextZ processthreadN)r'r(r)_namer0Processr@rrrrrrEsrEcs8eZdZdZ    ddfdd ZfddZZS)ThreadDataLoadera Subclass of `DataLoader` using a `ThreadBuffer` object to implement `__iter__` method asynchronously. This will iterate over data from the loader as expected however the data is generated on a separate thread. Use this class where a `DataLoader` instance is required and not just an iterable object. The default behaviour with `repeats` set to 1 is to yield each batch as it is generated, however with a higher value the generated batch is yielded that many times while underlying dataset asynchronously generates the next. Typically not all relevant information is learned from a batch in a single iteration so training multiple times on the same batch will still produce good training with minimal short-term overfitting while allowing a slow batch generation process more time to produce a result. This duplication is done by simply yielding the same object many times and not by regenerating the data. Another typical usage is to accelerate light-weight preprocessing (usually cached all the deterministic transforms and no IO operations), because it leverages the separate thread to execute preprocessing to avoid unnecessary IPC between multiple workers of DataLoader. And as CUDA may not work well with the multi-processing of DataLoader, `ThreadDataLoader` can be useful for GPU transforms. For more details: https://github.com/Project-MONAI/tutorials/blob/master/acceleration/fast_model_training_guide.md. The `use_thread_workers` will cause workers to be created as threads rather than processes although everything else in terms of how the class works is unchanged. This allows multiple workers to be used in Windows for example, or in any other situation where thread semantics is desired. Please note that some MONAI components like several datasets and random transforms are not thread-safe and can't work as expected with `thread workers`, need to check all the preprocessing components carefully before enabling `use_thread_workers`. See: * Fischetti et al. "Faster SGD training by minibatch persistency." ArXiv (2018) https://arxiv.org/abs/1806.07353 * Dami et al., "Faster Neural Network Training with Data Echoing" ArXiv (2020) https://arxiv.org/abs/1907.05550 * Ramezani et al. "GCN meets GPU: Decoupling "When to Sample" from "How to Sample"." NeurIPS (2020). https://proceedings.neurips.cc/paper/2020/file/d714d2c5a796d5814c565d78dd16188d-Paper.pdf Args: dataset: input dataset. buffer_size: number of items to buffer from the data source. buffer_timeout: time to wait for an item from the buffer, or to wait while the buffer is full when adding items. repeats: number of times to yield the same batch. use_thread_workers: if True and num_workers > 0 the workers are created as threads instead of processes kwargs: other arguments for `DataLoader` except for `dataset`. r r Fdatasetr r rbuffer_timeoutrr+use_thread_workersboolc sP|r|dddkrt|d<d|d<tj|fi|||_||_||_dS)N num_workersrmultiprocessing_contextFpersistent_workers)r%rEr4rr rJr+)rrIr rJr+rKkwargsr<rrrs  zThreadDataLoader.__init__c#s&tt|j|j|jEdHdSr1)r/r4r&r rJr+rr<rrr&s$zThreadDataLoader.__iter__)r r r F) rIr r rrJrr+rrKrL)r'r(r)r*rr&r?rrr<rrHs+rH)r r r )r rrrr+r) __future__rZmultiprocessing.contextrqueuerrr threadingrr6 monai.datarr r r/r0r@rErHrrrrs   <