U
e5d- @ 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' Zed(d) Zd S )-r r c C s |dkrddl m} || _tjdd\| _| _| | _t
| _ t j
dkrTd | _n
| | _||| _d| _| t j
dkrt| tj d S )Nr r )
SEM_VALUE_MAXFZduplexwin32)Z synchronizer _maxsizer Pipe_reader_writerLock_rlockosgetpid_opidsysplatform_wlockZBoundedSemaphore_sem
_ignore_epipe _after_forkr
r selfmaxsizectx r% ./usr/lib64/python3.8/multiprocessing/queues.py__init__$ s
zQueue.__init__c C s. t | | 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 |\| _ | _| _| _| _| _| _| _| d S r( ) r r r r r r r r r r" stater% r% r& __setstate__> s zQueue.__setstate__c C sb t d tt | _t | _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_pollr* r% r% r& r C s
zQueue._after_forkTNc C sf | j rtd| d| j||s(t| j. | jd krB| | j | | j
W 5 Q R X d S NzQueue z
is closed) r8
ValueErrorr acquirer r1 r5
_start_threadr4 appendnotifyr" objblocktimeoutr% r% r& putP s
z Queue.putc C s | j rtd| d|rH|d krH| j | }W 5 Q R X | j nr|rXt | }| j||sjt zB|r|t }|
|st n |
st | }| j W 5 | j X t |S r@ )
r8 rA r r= r releasetime monotonicrB r r? _ForkingPicklerloads)r" rH rI resZdeadliner% r% r& get\ s*
z Queue.getc C s | j | jj S r( )r r _semlockZ
_get_valuer* r% r% r& qsizev s z Queue.qsizec C s
| S r( r? r* r% r% r& emptyz s z Queue.emptyc C s | j j S r( )r rR _is_zeror* r% r% r& full} s z
Queue.fullc C s
| dS NF)rQ r* r% r% r&
get_nowait s zQueue.get_nowaitc C s | |dS rX )rJ r" rG r% r% r&
put_nowait s zQueue.put_nowaitc C s2 d| _ z| j W 5 | j}|r,d | _| X d S )NT)r8 r9 r close)r" r\ r% r% r& r\ s z Queue.closec C s t d | jr| d S )NzQueue.join_thread())r
r6 r* r% r% r& join_thread s zQueue.join_threadc C s6 t d d| _z| j W n tk
r0 Y nX d S )NzQueue.cancel_join_thread()T)r
r7 r6 ZcancelAttributeErrorr* r% r% r& cancel_join_thread s zQueue.cancel_join_threadc
C s t d | j tjtj| j| j| j| j | j
j | j | j
| jfdd| _d| j_t d | j t d | jst| jtjt| jgdd| _t| tj| j| jgd d| _d S )
NzQueue._start_thread()ZQueueFeederThread)targetargsnameTzdoing self._thread.start()z ... done self._thread.start())Z exitpriority
)r
r4 clearr/ ZThreadr _feedr1 r; r r r\ r _on_queue_feeder_errorr r5 Zdaemonstartr7 r _finalize_joinweakrefrefr6 _finalize_closer9 r* r% r% r& rC s<
zQueue._start_threadc C s4 t d | }|d k r(| t d nt d d S )Nzjoining queue threadz... queue thread joinedz ... queue thread already dead)r
join)Ztwrthreadr% r% r& ri s
zQueue._finalize_joinc C s. t d | | t | W 5 Q R X d S )Nz telling queue thread to quit)r
rD _sentinelrE )buffernotemptyr% r% r& rl s
zQueue._finalize_closec
C sX t d |j}|j} |j}
| j} t} tjdkr<|j}
|j}nd }
z| z| sT|
W 5 | X zb| }|| krt d | W W d S t |}|
d kr|| qb|
z || W 5 | X qbW n t
k
r Y nX W q@ t k
rP } zV|rt |ddt
jkrW Y 6d S t r.td| W Y d S | ||| W 5 d }~X Y q@X q@d S )Nz$starting thread to feed data to piper z%feeder thread got sentinel -- exitingerrnor zerror in queue thread: %s)r
rB rK waitpopleftro r r rN dumps
IndexError Exceptiongetattrrr ZEPIPEr r )rp rq r: Z writelockr\ Z ignore_epipeonerrorZ queue_semZnacquireZnreleaseZnwaitZbpopleftsentinelZwacquireZwreleaserG er% r% r& rf sN
z Queue._feedc C s ddl }| dS )z
Private API hook called when feeding data in the background thread
raises an exception. For overriding by concurrent.futures.
r N) traceback print_exc)r{ rG r| r% r% r& rg
s z Queue._on_queue_feeder_error)r )TN)TN)__name__
__module__ __qualname__r' r+ r. r rJ rQ rS rU rW rY r[ r\ r] r_ rC staticmethodri rl rf rg 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 |d| _| | _d S )N)r$ r )r r' Z Semaphore_unfinished_tasksr0 _condr! r% r% r& r' # s zJoinableQueue.__init__c C s t | | j| jf S r( )r r+ r r r* r% r% r& r+ ( s zJoinableQueue.__getstate__c C s, t | |d d |dd \| _| _d S )N)r r. r r r, r% r% r& r. + s zJoinableQueue.__setstate__TNc
C s | j rtd| d| j||s(t| jJ | j8 | jd krJ| | j
| | j | j
W 5 Q R X W 5 Q R X d S r@ )r8 rA r rB r r1 r r5 rC r4 rD r rK rE rF r% r% r& rJ / s
zJoinableQueue.putc C s@ | j 0 | jds td| jj r2| j W 5 Q R X d S )NFz!task_done() called too many times)r r rB rA rR rV Z
notify_allr* r% r% r& task_done< s
zJoinableQueue.task_donec C s, | j | jj s | j W 5 Q R X d S r( )r r rR rV rs r* r% r% r& rm C s zJoinableQueue.join)r )TN) r~ r r r' r+ r. rJ r rm 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\| _| _| | _| jj| _tj dkr:d | _
n
| | _
d S )NFr r ) r r r r r r r> r? r r r )r" r$ r% r% r& r' N s
zSimpleQueue.__init__c C s
| S r( rT r* r% r% r& rU W s zSimpleQueue.emptyc C s t | | j| j| j| jfS r( )r r) r r r r r* r% r% r& r+ Z s
zSimpleQueue.__getstate__c C s" |\| _ | _| _| _| j j| _d S r( )r r r r r> r? r, r% r% r& r. ^ s zSimpleQueue.__setstate__c C s& | j | j }W 5 Q R X t|S r( )r r r< rN rO )r" rP r% r% r& rQ b s zSimpleQueue.getc C sD t |}| jd kr"| j| n | j | j| W 5 Q R X d S r( )rN ru r r r: rZ r% r% r& rJ h s
zSimpleQueue.putN) r~ r r r' rU r+ r. rQ rJ r% r% r% r& r L s ) __all__r r r/ r2 rL rj rr Zqueuer r Z_multiprocessing r r Z reductionZForkingPicklerrN utilr
r r r
r objectr ro r r r% r% r% r&
s$
v
+