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 s\ | j s t| jj||s t| j. | jd kr8| j | jj | | jj
W d Q R X d S )N) r4 AssertionErrorr 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! rB rC 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 rK _is_zero)r! r$ r$ r% fullz s z
Queue.fullc C s
| j dS )NF)rJ )r! r$ r$ r%
get_nowait} s zQueue.get_nowaitc C s | j |dS )NF)rD )r! rA 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! rR r$ r$ r% rR s z Queue.closec C s$ t d | jst| jr | j d S )NzQueue.join_thread())r
r4 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 rR 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% rb 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= rE waitpopleftrd r r rG dumps
IndexError Exceptiongetattrrg ZEPIPEr r traceback print_exc)re rf r6 Z writelockrR Z ignore_epipeZnacquireZnreleaseZnwaitZbpopleftsentinelZwacquireZwreleaserA ern r$ r$ r% r] sR
z Queue._feed)r )TN)TN)__name__
__module__ __qualname__r& r( r* r rD rJ rL rM rO rP rQ rR rS rU r> staticmethodr_ rb 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( rw rv )r! r$ r$ r% r( s zJoinableQueue.__getstate__c C s, t j| |d d |dd \| _| _d S )N ry )r r* rw rv )r! r) r$ r$ r% r* s zJoinableQueue.__setstate__TNc C sx | j s t| jj||s t| jJ | j8 | jd kr@| j | j j
| | j j | jj
W d Q R X W d Q R X d S )N)r4 r< r r= r r- rw r1 r> r0 r? rv rE r@ )r! rA rB rC r$ r$ r% rD 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)rw rv r=
ValueErrorrK rN 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)rw rv rK rN rh )r! r$ r$ r% rc . s zJoinableQueue.join)r )TN) rr rs rt r& r( r* rD r{ rc 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% rM 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 rG rH )r! rI r$ r$ r% rJ M s zSimpleQueue.getc C sD t j|}| jd kr"| jj| n | j | jj| W d Q R X d S )N)rG rj r r r6 )r! rA r$ r$ r% rD S s
zSimpleQueue.putN) rr rs rt r& rM r( r* rJ rD r$ r$ r$ r% r 7 s ) __all__r r r+ r. rF r` rg Zqueuer r Z_multiprocessing r r Z reductionZForkingPicklerrG utilr
r r r
r objectr rd r r r$ r$ r$ r%
s$
b
*