U :e-@sdddgZddlZddlZddlZddlZddlZddlZddlZddlm Z m Z ddl Z ddl m Z ddl mZejjZdd lmZmZmZmZmZGd ddeZeZGd ddeZGd ddeZdS) Queue SimpleQueue JoinableQueueN)EmptyFull) connection)context)debuginfoFinalizeregister_after_fork is_exitingc@seZdZd*ddZddZddZdd Zd+d d Zd,ddZddZ ddZ ddZ ddZ ddZ ddZddZddZd d!Zed"d#Zed$d%Zed&d'Zed(d)Zd S)-rrcCs|dkrddlm}||_tjdd\|_|_||_t |_ t j dkrTd|_n ||_|||_d|_|t j dkrt|tjdS)Nrr) SEM_VALUE_MAXFduplexwin32) synchronizer_maxsizerPipe_reader_writerLock_rlockosgetpid_opidsysplatform_wlockBoundedSemaphore_sem _ignore_epipe _after_forkr rselfmaxsizectxr(;/usr/local/python38/lib/python3.8/multiprocessing/queues.py__init__$s       zQueue.__init__cCs.t||j|j|j|j|j|j|j|j fSN) r assert_spawningr"rrrrrr!rr%r(r(r) __getstate__9s zQueue.__getstate__c Cs0|\|_|_|_|_|_|_|_|_|dSr+) r"rrrrrr!rr#r%stater(r(r) __setstate__>szQueue.__setstate__cCsbtdtt|_t|_d|_d|_ d|_ d|_ d|_ |j j|_|jj|_|jj|_dS)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#Cs   zQueue._after_forkTNc Csf|jrtd|d|j||s(t|j.|jdkrB||j ||j W5QRXdSNzQueue z is closed) r; ValueErrorr!acquirerr4r8 _start_threadr7appendnotifyr%objblocktimeoutr(r(r)putPs  z Queue.putc Cs|jrtd|d|rH|dkrH|j|}W5QRX|jnr|rXt|}|j||sjt zB|r|t}| |st n | st |}|jW5|jXt |SrC) r;rDrr@r!releasetime monotonicrErrB_ForkingPicklerloads)r%rKrLresdeadliner(r(r)get\s*      z Queue.getcCs|j|jjSr+)rr!_semlock _get_valuer-r(r(r)qsizevsz Queue.qsizecCs | Sr+rBr-r(r(r)emptyzsz Queue.emptycCs |jjSr+)r!rV_is_zeror-r(r(r)full}sz Queue.fullcCs |dSNF)rUr-r(r(r) get_nowaitszQueue.get_nowaitcCs ||dSr])rMr%rJr(r(r) put_nowaitszQueue.put_nowaitcCs2d|_z|jW5|j}|r,d|_|XdS)NT)r;r<rclose)r%rar(r(r)rasz Queue.closecCstd|jr|dS)NzQueue.join_thread())r r9r-r(r(r) join_threadszQueue.join_threadcCs6tdd|_z|jWntk r0YnXdS)NzQueue.cancel_join_thread()T)r r:r9cancelAttributeErrorr-r(r(r)cancel_join_threads zQueue.cancel_join_threadc Cstd|jtjtj|j|j|j|j |j j |j |j |jfdd|_d|j_td|jtd|jst|jtjt|jgdd|_t|tj|j|jgd d|_dS) NzQueue._start_thread()QueueFeederThread)targetargsnameTzdoing self._thread.start()z... done self._thread.start()) exitpriority )r r7clearr2Threadr_feedr4r>rrrar"_on_queue_feeder_errorr!r8daemonstartr:r _finalize_joinweakrefrefr9_finalize_closer<r-r(r(r)rFs<     zQueue._start_threadcCs4td|}|dk r(|tdntddS)Nzjoining queue threadz... queue thread joinedz... queue thread already dead)r join)twrthreadr(r(r)rss  zQueue._finalize_joinc Cs.td||t|W5QRXdS)Nztelling queue thread to quit)r rG _sentinelrH)buffernotemptyr(r(r)rvs zQueue._finalize_closec CsXtd|j}|j} |j} |j} t} tjdkr<|j} |j}nd} z|z|sT| W5| Xzb| }|| krtd|WWdSt |}| dkr||qb| z ||W5|XqbWnt k rYnXWq@t k rP}zV|rt |ddt jkrWY6dStr.td|WYdS||||W5d}~XYq@Xq@dS)Nz$starting thread to feed data to piperz%feeder thread got sentinel -- exitingerrnorzerror in queue thread: %s)r rErNwaitpopleftrzrrrQdumps IndexError Exceptiongetattrr}EPIPErr )r{r|r= writelockra ignore_epipeonerror queue_semnacquirenreleasenwaitbpopleftsentinelwacquirewreleaserJer(r(r)rosN         z Queue._feedcCsddl}|dS)Nr) traceback print_exc)rrJrr(r(r)rp szQueue._on_queue_feeder_error)r)TN)TN)__name__ __module__ __qualname__r*r.r1r#rMrUrXrZr\r^r`rarbrerF staticmethodrsrvrorpr(r(r(r)r"s.       =c@s@eZdZdddZddZddZdd d Zd d ZddZd S)rrcCs*tj|||d|d|_||_dS)N)r'r)rr* Semaphore_unfinished_tasksr3_condr$r(r(r)r*#s zJoinableQueue.__init__cCst||j|jfSr+)rr.rrr-r(r(r)r.(szJoinableQueue.__getstate__cCs,t||dd|dd\|_|_dS)N)rr1rrr/r(r(r)r1+szJoinableQueue.__setstate__TNc Cs|jrtd|d|j||s(t|jJ|j8|jdkrJ||j ||j |j W5QRXW5QRXdSrC)r;rDr!rErr4rr8rFr7rGrrNrHrIr(r(r)rM/s   zJoinableQueue.putc Cs@|j0|jdstd|jjr2|jW5QRXdS)NFz!task_done() called too many times)rrrErDrVr[ notify_allr-r(r(r) task_done<s   zJoinableQueue.task_donec Cs,|j|jjs|jW5QRXdSr+)rrrVr[r~r-r(r(r)rwCs zJoinableQueue.join)r)TN) rrrr*r.r1rMrrwr(r(r(r)r!s   c@s<eZdZddZddZddZddZd d Zd d Zd S)rcCsHtjdd\|_|_||_|jj|_tj dkr:d|_ n ||_ dS)NFrr) rrrrrrrArBrrr)r%r'r(r(r)r*Ns    zSimpleQueue.__init__cCs | Sr+rYr-r(r(r)rZWszSimpleQueue.emptycCst||j|j|j|jfSr+)r r,rrrrr-r(r(r)r.Zs zSimpleQueue.__getstate__cCs"|\|_|_|_|_|jj|_dSr+)rrrrrArBr/r(r(r)r1^szSimpleQueue.__setstate__c Cs&|j|j}W5QRXt|Sr+)rrr?rQrR)r%rSr(r(r)rUbszSimpleQueue.getc CsDt|}|jdkr"|j|n|j|j|W5QRXdSr+)rQrrrr=r_r(r(r)rMhs   zSimpleQueue.putN) rrrr*rZr.r1rUrMr(r(r(r)rLs  )__all__rrr2r5rOrtr}queuerr_multiprocessingrr reductionForkingPicklerrQutilr r r r robjectrrzrrr(r(r(r) s$   v +