U
    dl                     @   sn   d dl Z d dlmZmZ d dlZd dlZd dlmZm	Z	m
Z
 d dlmZ dgZG dd dZG dd dZdS )	    N)AnyList)
DataLoaderIterDataPipecommunication)IterableWrapperDataLoader2c                   @   s.   e Zd ZdddZdd Zdd Zd	d
 ZdS )_ThreadingDataLoader2r   Nc           
      C   s   g | _ g | _|| _t|D ]h}tj|\}}}}tjj	j
||| |  | j |||f tjtj||}	| j|	 qd S N)threads	datapipes
collate_fnranger   Z	eventloopZSpawnThreadForDataPipelinetorchutilsdatagraph_settingsZapply_shardingstartappenditerZQueueWrapperprotocolZIterDataPipeQueueProtocolClient)
selfdatapipenum_workersr   Z	worker_idthread	req_queue	res_queueZthread_localdatapipeZlocal_datapipe r   L/tmp/pip-unpacked-wheel-ua33x9lu/torch/utils/data/dataloader_experimental.py__init__   s    z_ThreadingDataLoader2.__init__c              	   c   s   d}d}g }t |t | jk r| jD ]X}||kr$z| }|V  W q$ tk
r`   || Y q$ tjjk
rz   d}Y q$X q$|rt	d qd S )NFTgMbP?)
lenr   Znonblocking_nextStopIterationr   r   r   ZNotAvailabletimesleep)r   Znot_availableforeverZexclude_datapipesZdpvaluer   r   r   __iter__   s    

z_ThreadingDataLoader2.__iter__c                 C   s   |    d S r
   )_cleanup_all_threads)r   r   r   r   __del__0   s    z_ThreadingDataLoader2.__del__c                 C   s*   dd }| j D ]\}}}|||| qd S )Nc                 S   s$   | tj  | }|   d S r
   )putr   messagesZTerminateRequestgetjoin)r   r   r   _r   r   r   clean_me4   s    z<_ThreadingDataLoader2._cleanup_all_threads.<locals>.clean_me)r   )r   r.   r   r   r   r   r   r   r'   3   s    z*_ThreadingDataLoader2._cleanup_all_threads)r   N)__name__
__module____qualname__r   r&   r(   r'   r   r   r   r   r	      s   
r	   c                
   @   s"   e Zd Zd
ddddddd	ZdS )r      Nr   F   mp)prefetch_factorpersistent_workersbatch_outside_workerparallelism_modec                C   sx  t |tr@d }|d k	r td|d k	r0td|}tjjjj||d}|rX|rXtd|s|d k	r|j||	d}|d krtjjj	j
j}t|d |d d |||d|
|||d}nT|dkr|d k	r|s||}|rtd	|d k	rtd
t|||d}n
td||s
|S |d kr"tjjj	j
j}t|j||	d|}|S n4|dkrRtdt|||||||||	|
|||dS d S )Nz/batch_sampler is not yet supported by DataPipesz)sampler is not yet supported by DataPipes)shufflez:pin_memory is not yet compatible with batch_outside_worker)	drop_lastF)
batch_sizer9   samplerbatch_samplerr   r   
pin_memoryr:   timeoutworker_init_fnr5   r6   r   z;pin_memory is not yet supported by DataPipes with Threadingz?worker_init_fn is not yet supported by DataPipes with Threading)r   r   zUnsupported parallelism modez9thread parallelism mode is not supported for old DataSets)
isinstancer   	Exceptionr   r   r   r   Zapply_shuffle_settingsbatch_utilsZcollateZdefault_collater   mapr	   r   )clsZdatasetr;   r9   r<   r=   r   r   r>   r:   r?   r@   r5   r6   r7   r8   Zdata_loaderr   r   r   r   __new__=   s    


 
zDataLoader2.__new__)
r2   NNNr   NFFr   N)r/   r0   r1   rG   r   r   r   r   r   <   s              )r"   typingr   r   Z'torch.utils.data.backward_compatibilityr   Ztorch.utils.data.graph_settingsZtorch.utils.datar   r   r   Ztorch.utils.data.datapipes.iterr   __all__r	   r   r   r   r   r   <module>   s   ,