U
    dE>                     @   s   d Z ddlZddlmZ ddlZddlZddlZddl	Z	ddl
mZmZmZmZ ddlmZ ddlmZmZmZmZmZmZ edZeej dZd	d
 ZdddZG dd deZG dd deZ G dd deZ!e  a"dd Z#dS )a  
This module provides a python-land multithreaded data input mechanism
for Caffe2 nets.

Basic usage is as follows:
   coordinator = data_workers.init_data_input_workers(
      net,
      ["data", "label"],
      my_fetch_fun,
      batch_size=32,
      input_source_name="train",
      dont_rebatch=False
   )
   ...
   coordinator.start()

First argument is the Caffe2 net (or model helper), and second argument
is list of input blobs that are to be fed.

Argument 'input_source_name' is used to distinguish different sources of data,
such as train or test data. This is to ensure the data does not get mixed up,
although two nets would share blobs.

To do the actual data loading, one defines a "fetcher function"
that has call signature
   my_fetch_fun(worker_id, batch_size)

Optionally, one can define a "init function" that is called once before
threads start, and has call signature:
   my_init_fun(data_coordinator, global_coordinator)

If dont_rebatch is set to True, the data input is not batched into equal sized
chunks but data directly provided by fetchers is used.

'batch_columns' can be used to specify which dimension is the batch dimension,
for each of the inputs. Default is 0 for all iputs.

'timeout' is the timeout in seconds after which if no data is available, the
net will fail (default 600s = 10 mins).

This function returns a list of numpy arrays corresponding to the different
input blobs. In the example above, it would return two arrays, one for the
data blob and another for the labels. These arrays can have arbitrary number
of elements (i.e they do not need to match the batch size). The batch size
is provided for the function as a hint only.

For example, fetcher function could download images from a remote service or
load random images from a directory on a file system.

For a dummy example, see the data_workers_test unit test.

Note that for data_parallel_models, init_data_input_workers will be called
for each GPU. Note that the 'coordinator' returned by the function is same
each time.
    N)chain)	workspacecorescopeutils)
caffe2_pb2)MetricsStateWorkerCoordinatorGlobalWorkerCoordinatorWorker
run_workerZdata_workers<   c                 C   s   t td| S Nr   )listrange)Znum_workers r   >/tmp/pip-unpacked-wheel-ua33x9lu/caffe2/python/data_workers.pyget_worker_idsR   s    r      train   FX  c                    s   t  }|d krtjtjd}t|t| ||t  |t	|||	|
|d dd t
|D }t|||  fdd|D }|tjtd|t   gd |_t tS )N)Zdevice_type)timeoutc                 S   s   g | ]}t  qS r   )global_coordinatorZget_new_worker_id).0ir   r   r   
<listcomp>y   s   z+init_data_input_workers.<locals>.<listcomp>c                    s4   g | ],}t jtd |t| gdqS )zdata_workers fetcher id {}targetnameargs)	threadingThreadr   format
DataWorker)r   	worker_idbatch_feeder
batch_sizecoordinator	fetch_funmetricsr   r   r      s   
 zEnqueuer {} {}r   )r   ZCurrentDeviceScoper   ZDeviceOptionZCPUr   BatchFeederZCurrentNameScoper   	get_queuer   r
   appendr"   r#   enqueuerr$   Z_workersadd)netinput_blob_namesr+   r)   Znum_worker_threadsinput_source_namemax_buffered_batchesZinit_funZexternal_loggersdont_rebatchbatch_columnsr   device_optionZ
worker_idsworkersr   r'   r   init_data_input_workersV   sJ    
   

r:   c                   @   s   e Zd Zd ddZdd Zdd Zdd	 Zd
d Zdd Zdd Z	dd Z
dd Zdd Zdd Zdd Zdd Zd!ddZdS )"r-   r   c                 C   s   d| _ || _|| _|| _g | _|| _|| _|| _|| _d| _	| 
| | | d| _d| _t | _|	| _|   || _|
d krdd |D }
|
| _d S )Nr      c                 S   s   g | ]}d qS )r   r   )r   _r   r   r   r      s     z(BatchFeeder.__init__.<locals>.<listcomp>)_counter_input_blob_names_batch_size_internal_queue_queues_device_option
_namescope_timeout_input_source_name_c2_queue_capacity_create_caffe2_queues_create_caffe2_ops_inputs_prev_secondstime_last_warning_dont_rebatch_init_scratch_metrics_batch_columns)selfr2   r3   r)   r8   	namescoper4   queuer,   r6   r7   r   r   r   r   __init__   s*    


zBatchFeeder.__init__c                 C   s   d| _ t | _d S r   )rI   rK   rJ   rQ   r   r   r   start   s    zBatchFeeder.startc              
   C   s<   z&| jD ]}ttd|gg  qW 5 | j ddd X d S )Nr   T)forceZCloseBlobsQueue)_log_inputs_per_intervalrA   r   RunOperatorOncer   CreateOperator)rQ   qr   r   r   stop   s    

zBatchFeeder.stopc                 C   s$   t | j  t | j  d S N)r   Z
ResetBlobs_scratch_blobvalues_scratch_statusrU   r   r   r   cleanup   s    zBatchFeeder.cleanupc              	   C   s|   t   }t   }| rxz| jjdddW S  tjk
rt   t   | dkrltdt   |  t   }Y qY qX qd S )NT      ?blockr   g      $@z2** Data input is slow: (still) no data in {} secs.)	rK   	is_activer@   getQueueEmptylogwarningr$   )rQ   data_input_coordinator
start_timeZlast_warningr   r   r   _get   s    
zBatchFeeder._getc                 C   s   |d krt d dS t|t| jks0td|D ]}t|tjs4tdq4| jsd}|dd  D ]8}|j	| j
|  |d j	| j
d  kstd|d7 }qdt|dkrt d dS d	S )
NzFetcher function returned NoneFz"Expecting data blob for each inputz*Fetcher function must return a numpy array   r   z5Each returned input must have equal number of samplesz!Worker provided zero length inputT)ri   rj   lenr>   AssertionError
isinstancenpZndarrayrM   shaperP   )rQ   chunkdjr   r   r   _validate_chunk   s.    


zBatchFeeder._validate_chunkc                 C   s   |  |sd S | rz| j }|dk r^t | j tkr^tdd	|| j
  t | _|  jd7  _| jj|ddd | |d jd  W d S  tjk
r   td	 Y qY qX qd S )
Nr   z&Warning, data loading lagging behind: zqueue size={}, name={}rn   Trb   rc   r   z Queue full: stalling fetchers...)rw   re   r@   qsizerK   rL   LOG_INT_SECSri   rj   r$   rE   r=   putrX   rs   rg   Fulldebug)rQ   rt   rk   rx   r   r   r   rz      s"    



zBatchFeeder.putc                 C   sL   |  |}|d krd S | rHt| j| j|D ]\}}}| ||| q.d S r]   )rm   re   zipr>   rA   _enqueue)rQ   rk   databr[   cr   r   r   _enqueue_batch_direct   s    
z!BatchFeeder._enqueue_batch_directc                 C   s  | j r| | dS dd | jD }| jd }|d jd dksT|d j| | jk r| r| |}|dkrpq.t|D ]F\}}|| jd dkr|	 ||< qxt
j|| || j| d||< qxq.t }z|d jd dkr|d j| | jkrg }g }	t|D ]<\}}
t
j|
| jg| j| d\}}|| |	| q|	}z| jj|dd W n tjk
rt   Y nX |d j| | jkst| rt| j| j|D ]\}
}}| |
|| qW 5 | jdt |  X dS )	z
        This pulls data from the python-side queue and collects them
        into batch-sized pieces, unless dont_rebatch is set to true.
        Nc                 S   s   g | ]}t g qS r   )rr   array)r   ru   r   r   r   r     s     z.BatchFeeder._enqueue_batch.<locals>.<listcomp>r   )ZaxisZenqueue_timeF)rd   )rM   r   r>   rP   rs   r?   re   rm   	enumeratecopyrr   r/   rK   rO   
put_metricsplitr@   rz   rg   r{   rp   r}   rA   r~   )rQ   rk   Z	cur_batchZfirst_batch_colrt   rv   Z
chunk_elemrl   leftoverZtrimmed_batchr   r   lr[   r   r   r   _enqueue_batch  sj    


    


  zBatchFeeder._enqueue_batchc                 C   s   i | _ i | _| jD ]<}| j| d | j }t|| j |< t|d | j|< qt| j  | j D ]$}t	j
|tg tj| jd qfd S )NZ	_scratch_Z_statusr8   )r^   r`   r>   rC   rE   r   ZBlobReferencer   r_   r   FeedBlobrr   r   ZastypeZfloat32rB   )rQ   	blob_nameZscratch_namer   r   r   r   rN   >  s*    
 zBatchFeeder._init_scratchc                 C   sT   t j| j| || jd tjd|| j| g| j| | j| g| jd}t | dS )zM
        Enqueue the correctly sized batch arrays to Caffe2's queue.
        r   ZSafeEnqueueBlobsN)r   r   r^   rB   r   rZ   r`   rY   )rQ   r   rS   Zdata_arropr   r   r   r~   U  s    zBatchFeeder._enqueuec                 C   sF   dd }| j D ]2}|d d | j }||d| jd}| j| qdS )z/
        Creates queues on caffe2 side
        c              	   S   s&   t tjdg | gd|d t| S )NZCreateBlobsQueuern   	num_blobscapacity)r   rY   r   rZ   ZScopedBlobReference)
queue_namer   r   r   r   r   create_queuek  s     z7BatchFeeder._create_caffe2_queues.<locals>.create_queueZ_c2queuer<   rn   r   N)r>   rE   rF   rA   r/   )rQ   r2   r   r   qnamer[   r   r   r   rG   g  s    	
  z!BatchFeeder._create_caffe2_queuesc                 C   s2   t | j| jD ]\}}|j||t| jd qdS )z4
        Creates dequeue-ops on caffe2 side
        )Ztimeout_secsN)r}   rA   r>   ZDequeueBlobsfloatrD   )rQ   r2   r[   r   r   r   r   rH   {  s    zBatchFeeder._create_caffe2_opsFc                 C   s   |  j |7  _ t }|| j }|tks,|rt| j | }| j }td	| j
| j| td	| | jd|d | jd|d | jd|d | j  | j  d| _ || _d S )Nz{}/{}: {} inputs/secz-- queue: {} batchesinputs_per_secFZ
queue_sizeZtime_elapsedr   )rI   rK   rJ   ry   intr@   rx   ri   infor$   rE   rC   rO   r   Zlog_metricsZreset_metrics)rQ   inputsrW   Zcurrent_secondsZdelta_secondsr   rx   r   r   r   rX     s6    

    

z$BatchFeeder._log_inputs_per_intervalN)r   )F)__name__
__module____qualname__rT   rV   r\   ra   rm   rw   rz   r   r   rN   r~   rG   rH   rX   r   r   r   r   r-      s    
	;r-   c                   @   s$   e Zd Zdd Zdd Zdd ZdS )GlobalCoordinatorc                 C   s   t |  i | _d S r]   )r   rT   rA   rU   r   r   r   rT     s    
zGlobalCoordinator.__init__c                 C   s4   t |tst|| jkr*tj|d| j|< | j| S )N)maxsize)rq   r   rp   rA   rg   )rQ   r   r5   r   r   r   r.     s    
zGlobalCoordinator.get_queuec                 C   sL   t d|| | jD ].}|j|kr|jj|kr||j_|j| qd S )Nz$Reset data input {}, batch size {}: )	ri   r   r$   Z_coordinatorsZ_worker_name_staterC   r?   rH   )rQ   rR   r    r2   r)   r   r   r   r   reset_data_input  s
    
z"GlobalCoordinator.reset_data_inputN)r   r   r   rT   r.   r   r   r   r   r   r     s   r   c                   @   s$   e Zd Zdd Zdd Zdd ZdS )r%   c                 C   s$   t j| ||||d || _|| _d S )N)
worker_funr,   )r   rT   r?   _batch_feeder)rQ   r*   r&   r   r,   r)   r(   r   r   r   rT     s
    	zDataWorker.__init__c                 C   s$   |  | j| j}| j|| j d S r]   )Z_worker_funZ
_worker_idr?   r   rz   Z_coordinator)rQ   Z
input_datar   r   r   run  s    zDataWorker.runc                 C   s   | j dt | j  d S )NZfetcher_time)rO   r   rK   Z_start_timerU   r   r   r   finish  s     zDataWorker.finishN)r   r   r   rT   r   r   r   r   r   r   r%     s   r%   c                 C   s   |   r||  q d S r]   )re   r   )r*   r(   r   r   r   r0     s    r0   )r   r   r   NNFNr   )$__doc__rS   rg   	itertoolsr   loggingr"   Znumpyrr   rK   Zcaffe2.pythonr   r   r   r   Zcaffe2.protor   Zcaffe2.python.parallel_workersr   r	   r
   r   r   r   	getLoggerri   setLevelINFOry   r   r:   r-   r   r%   r   r0   r   r   r   r   <module>	   s:   8 
	        
@  