U e5d-@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_MAXFZduplexwin32)Z synchronizer_maxsizerPipe_reader_writerLock_rlockosgetpid_opidsysplatform_wlockZBoundedSemaphore_sem _ignore_epipe _after_forkr rselfmaxsizectxr%./usr/lib64/python3.8/multiprocessing/queues.py__init__$s       zQueue.__init__cCs.t||j|j|j|j|j|j|j|j fSN) r assert_spawningrrrrrrrrr"r%r%r& __getstate__9s zQueue.__getstate__c Cs0|\|_|_|_|_|_|_|_|_|dSr() rrrrrrrrr 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) r8 ValueErrorracquirerr1r5 _start_threadr4appendnotifyr"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 |Sr@) r8rArr=rreleasetime monotonicrBrr?_ForkingPicklerloads)r"rHrIresZdeadliner%r%r&get\s*      z Queue.getcCs|j|jjSr()rr_semlockZ _get_valuer*r%r%r&qsizevsz Queue.qsizecCs | Sr(r?r*r%r%r&emptyzsz Queue.emptycCs |jjSr()rrR_is_zeror*r%r%r&full}sz Queue.fullcCs |dSNF)rQr*r%r%r& get_nowaitszQueue.get_nowaitcCs ||dSrX)rJr"rGr%r%r& put_nowaitszQueue.put_nowaitcCs2d|_z|jW5|j}|r,d|_|XdS)NT)r8r9rclose)r"r\r%r%r&r\sz Queue.closecCstd|jr|dS)NzQueue.join_thread())r r6r*r%r%r& join_threadszQueue.join_threadcCs6tdd|_z|jWntk r0YnXdS)NzQueue.cancel_join_thread()T)r r7r6ZcancelAttributeErrorr*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()ZQueueFeederThread)targetargsnameTzdoing self._thread.start()z... done self._thread.start())Z exitpriority )r r4clearr/ZThreadr_feedr1r;rrr\r_on_queue_feeder_errorrr5Zdaemonstartr7r _finalize_joinweakrefrefr6_finalize_closer9r*r%r%r&rCs<     zQueue._start_threadcCs4td|}|dk r(|tdntddS)Nzjoining queue threadz... queue thread joinedz... queue thread already dead)r join)Ztwrthreadr%r%r&ris  zQueue._finalize_joinc Cs.td||t|W5QRXdS)Nztelling queue thread to quit)r rD _sentinelrE)buffernotemptyr%r%r&rls 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 rBrKwaitpopleftrorrrNdumps IndexError ExceptiongetattrrrZEPIPErr )rprqr:Z writelockr\Z ignore_epipeonerrorZ queue_semZnacquireZnreleaseZnwaitZbpopleftsentinelZwacquireZwreleaserGer%r%r&rfsN         z Queue._feedcCsddl}|dS)Nr) traceback print_exc)r{rGr|r%r%r&rg szQueue._on_queue_feeder_error)r)TN)TN)__name__ __module__ __qualname__r'r+r.r rJrQrSrUrWrYr[r\r]r_rC staticmethodrirlrfrgr%r%r%r&r"s.       =c@s@eZdZdddZddZddZdd d Zd d ZddZd S)rrcCs*tj|||d|d|_||_dS)N)r$r)rr'Z Semaphore_unfinished_tasksr0_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)rr.rrr,r%r%r&r.+szJoinableQueue.__setstate__TNc Cs|jrtd|d|j||s(t|jJ|j8|jdkrJ||j ||j |j W5QRXW5QRXdSr@)r8rArrBrr1rr5rCr4rDrrKrErFr%r%r&rJ/s   zJoinableQueue.putc Cs@|j0|jdstd|jjr2|jW5QRXdS)NFz!task_done() called too many times)rrrBrArRrVZ notify_allr*r%r%r& task_done<s   zJoinableQueue.task_donec Cs,|j|jjs|jW5QRXdSr()rrrRrVrsr*r%r%r&rmCs zJoinableQueue.join)r)TN) r~rrr'r+r.rJrrmr%r%r%r&r!s   c@s<eZdZddZddZddZddZd d Zd d Zd S)rcCsHtjdd\|_|_||_|jj|_tj dkr:d|_ n ||_ dS)NFrr) rrrrrrr>r?rrr)r"r$r%r%r&r'Ns    zSimpleQueue.__init__cCs | Sr(rTr*r%r%r&rUWszSimpleQueue.emptycCst||j|j|j|jfSr()r r)rrrrr*r%r%r&r+Zs zSimpleQueue.__getstate__cCs"|\|_|_|_|_|jj|_dSr()rrrrr>r?r,r%r%r&r.^szSimpleQueue.__setstate__c Cs&|j|j}W5QRXt|Sr()rrr<rNrO)r"rPr%r%r&rQbszSimpleQueue.getc CsDt|}|jdkr"|j|n|j|j|W5QRXdSr()rNrurrr:rZr%r%r&rJhs   zSimpleQueue.putN) r~rrr'rUr+r.rQrJr%r%r%r&rLs  )__all__rrr/r2rLrjrrZqueuerrZ_multiprocessingrr Z reductionZForkingPicklerrNutilr r r r robjectrrorrr%r%r%r& s$   v +