3
\ * @ s d ddgZ ddlZddlZddlZddlZddlZddlZddlZddlm Z m
Z
ddl Z ddl m
Z
ddl mZ ejjZdd lmZmZmZmZmZ G d
d d eZe ZG d d deZG d d deZ dS )
Queue SimpleQueue
JoinableQueue N)EmptyFull )
connection)context)debuginfoFinalizeregister_after_fork
is_exitingc @ s e Zd Zd(ddZdd Zdd Zdd Zd)d d
Zd*ddZdd Z dd Z
dd Z dd Z dd Z
dd Zd d Zd d Zd d! Zed"d# Zed$d% Zed&d' Zd S )+r r c C s |dkrddl m} || _tjdd\| _| _|j | _t j
| _ t j
dkrTd | _n
|j | _|j|| _d| _| j t j
dkrt| tj d S )Nr r )
SEM_VALUE_MAXF)duplexwin32)Z synchronizer _maxsizer Pipe_reader_writerLock_rlockosgetpid_opidsysplatform_wlockZBoundedSemaphore_sem
_ignore_epipe _after_forkr
r )selfmaxsizectx r$ ./usr/lib64/python3.6/multiprocessing/queues.py__init__$ s
zQueue.__init__c C s. t j| | j| j| j| j| j| j| j| j fS )N)
r assert_spawningr r r r r r r r )r! r$ r$ r% __getstate__9 s
zQueue.__getstate__c C s0 |\| _ | _| _| _| _| _| _| _| j d S )N) r r r r r r r r r )r! stater$ r$ r% __setstate__> s $zQueue.__setstate__c C sb t d tjtj | _tj | _d | _d | _ d| _
d| _ d | _ | j
j| _| jj| _| jj| _d S )NzQueue._after_fork()F)r
threading Conditionr _notempty collectionsdeque_buffer_thread _jointhread_joincancelled_closed_closer
send_bytes _send_bytesr
recv_bytes _recv_bytespoll_poll)r! r$ r$ r% r C s
zQueue._after_forkTNc C sP | j j||st| j. | jd kr,| j | jj| | jj W d Q R X d S )N) r acquirer r- r1
_start_threadr0 appendnotify)r! objblocktimeoutr$ r$ r% putP s
z Queue.putc
C s |r2|d kr2| j | j }W d Q R X | jj nr|rBtj | }| j j||sTtzB|rv|tj }| j|stn | j st| j }| jj W d | j j X t j
|S )N) r r9 r releasetimeZ monotonicr< r r; _ForkingPicklerloads)r! rA rB resZdeadliner$ r$ r% get[ s&
z Queue.getc C s | j | jjj S )N)r r _semlockZ
_get_value)r! r$ r$ r% qsizes s z Queue.qsizec C s
| j S )N)r; )r! r$ r$ r% emptyw s z Queue.emptyc C s | j jj S )N)r rJ _is_zero)r! r$ r$ r% fullz s z
Queue.fullc C s
| j dS )NF)rI )r! r$ r$ r%
get_nowait} s zQueue.get_nowaitc C s | j |dS )NF)rC )r! r@ r$ r$ r%
put_nowait s zQueue.put_nowaitc C s2 d| _ z| jj W d | j}|r,d | _| X d S )NT)r4 r closer5 )r! rQ r$ r$ r% rQ s z Queue.closec C s t d | jr| j d S )NzQueue.join_thread())r
r2 )r! r$ r$ r% join_thread s zQueue.join_threadc C s6 t d d| _y| jj W n tk
r0 Y nX d S )NzQueue.cancel_join_thread()T)r
r3 r2 ZcancelAttributeError)r! r$ r$ r% cancel_join_thread s zQueue.cancel_join_threadc C s t d | jj tjtj| j| j| j| j | j
j | j fdd| _
d| j
_t d | j
j t d | jst| j
tjtj| j
gd
d| _t| tj| j| jgd d| _d S ) NzQueue._start_thread()ZQueueFeederThread)targetargsnameTzdoing self._thread.start()z ... done self._thread.start() )Z exitpriority
)r
r0 clearr+ ZThreadr _feedr- r7 r r rQ r r1 Zdaemonstartr3 r _finalize_joinweakrefrefr2 _finalize_closer5 )r! r$ r$ r% r= s(
zQueue._start_threadc C s4 t d | }|d k r(|j t d nt d d S )Nzjoining queue threadz... queue thread joinedz ... queue thread already dead)r
join)ZtwrZthreadr$ r$ r% r^ s
zQueue._finalize_joinc C s. t d | | jt |j W d Q R X d S )Nz telling queue thread to quit)r
r> _sentinelr? )buffernotemptyr$ r$ r% ra s
zQueue._finalize_closec C sP t d |j}|j}|j}| j} t}
tjdkr<|j} |j} nd } xy| z| sX| W d | X ybx\| }
|
|
krt d | d S tj |
}
| d kr||
qh| z ||
W d | X qhW W n t
k
r Y nX W qD t k
rF } zJ|rt |ddt
jkrd S t r&td| d S dd l}|j W Y d d }~X qDX qDW d S )Nz$starting thread to feed data to piper z%feeder thread got sentinel -- exitingerrnor zerror in queue thread: %s)r
r< rD waitpopleftrc r r rF dumps
IndexError Exceptiongetattrrf ZEPIPEr r traceback print_exc)rd re r6 Z writelockrQ Z ignore_epipeZnacquireZnreleaseZnwaitZbpopleftsentinelZwacquireZwreleaser@ erm r$ r$ r% r\ sR
z Queue._feed)r )TN)TN)__name__
__module__ __qualname__r& r( r* r rC rI rK rL rN rO rP rQ rR rT r= staticmethodr^ ra r\ r$ r$ r$ r% r " s$
c @ s@ e Zd ZdddZdd Zdd Zdd
d Zd d
Zdd Zd S )r r c C s* t j| ||d |jd| _|j | _d S )N)r# r )r r& Z Semaphore_unfinished_tasksr, _cond)r! r" r# r$ r$ r% r& s zJoinableQueue.__init__c C s t j| | j| jf S )N)r r( rv ru )r! r$ r$ r% r( s zJoinableQueue.__getstate__c C s, t j| |d d |dd \| _| _d S )N rx )r r* rv ru )r! r) r$ r$ r% r* s zJoinableQueue.__setstate__TNc C sl | j j||st| jJ | j8 | jd kr4| j | jj| | j j
| jj W d Q R X W d Q R X d S )N) r r< r r- rv r1 r= r0 r> ru rD r? )r! r@ rA rB r$ r$ r% rC s
zJoinableQueue.putc C s@ | j 0 | jjds td| jjj r2| j j W d Q R X d S )NFz!task_done() called too many times)rv ru r<
ValueErrorrJ rM Z
notify_all)r! r$ r$ r% task_done' s
zJoinableQueue.task_donec C s, | j | jjj s | j j W d Q R X d S )N)rv ru rJ rM rg )r! r$ r$ r% rb . s zJoinableQueue.join)r )TN) rq rr rs r& r( r* rC rz rb r$ r$ r$ r% r
s
c @ s< e Zd Zdd Zdd Zdd Zdd Zd d
Zd d Zd
S )r c C sH t jdd\| _| _|j | _| jj| _tj dkr:d | _
n
|j | _
d S )NF)r r ) r r r r r r r: r; r r r )r! r# r$ r$ r% r& 9 s
zSimpleQueue.__init__c C s
| j S )N)r; )r! r$ r$ r% rL B s zSimpleQueue.emptyc C s t j| | j| j| j| jfS )N)r r' r r r r )r! r$ r$ r% r( E s
zSimpleQueue.__getstate__c C s" |\| _ | _| _| _| j j| _d S )N)r r r r r: r; )r! r) r$ r$ r% r* I s zSimpleQueue.__setstate__c C s& | j | jj }W d Q R X tj|S )N)r r r8 rF rG )r! rH r$ r$ r% rI M s zSimpleQueue.getc C sD t j|}| jd kr"| jj| n | j | jj| W d Q R X d S )N)rF ri r r r6 )r! r@ r$ r$ r% rC S s
zSimpleQueue.putN) rq rr rs r& rL r( r* rI rC r$ r$ r$ r% r 7 s ) __all__r r r+ r. rE r_ rf Zqueuer r Z_multiprocessing r r Z reductionZForkingPicklerrF utilr
r r r
r objectr rc r r r$ r$ r$ r%
s$
b
*