B -_C-@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_MAXF)duplexwin32)Z synchronizer_maxsizerPipe_reader_writerLock_rlockosgetpid_opidsysplatform_wlockZBoundedSemaphore_sem _ignore_epipe _after_forkr r)selfmaxsizectxr$,/usr/lib/python3.7/multiprocessing/queues.py__init__$s       zQueue.__init__cCs.t||j|j|j|j|j|j|j|j fS)N) r assert_spawningrrrrrrrr)r!r$r$r% __getstate__9s zQueue.__getstate__c Cs0|\|_|_|_|_|_|_|_|_|dS)N) rrrrrrrrr )r!stater$r$r% __setstate__>s$zQueue.__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_poll)r!r$r$r%r Cs   zQueue._after_forkTNc Csd|jrtd||j||s&t|j.|jdkr@||j ||j WdQRXdS)NzQueue {0!r} has been closed) r4AssertionErrorformatracquirerr-r1 _start_threadr0appendnotify)r!objblocktimeoutr$r$r%putPs  z Queue.putc Cs|r2|dkr2|j|}WdQRX|jnr|rBt|}|j||sTtzB|rv|t}||stn |st|}|jWd|jXt |S)N) rr9rreleasetimeZ monotonicr>rr;_ForkingPicklerloads)r!rCrDresZdeadliner$r$r%get[s&      z Queue.getcCs|j|jjS)N)rr_semlockZ _get_value)r!r$r$r%qsizessz Queue.qsizecCs | S)N)r;)r!r$r$r%emptywsz Queue.emptycCs |jjS)N)rrL_is_zero)r!r$r$r%fullzsz Queue.fullcCs |dS)NF)rK)r!r$r$r% get_nowait}szQueue.get_nowaitcCs ||dS)NF)rE)r!rBr$r$r% put_nowaitszQueue.put_nowaitcCs2d|_z|jWd|j}|r,d|_|XdS)NT)r4rcloser5)r!rSr$r$r%rSsz Queue.closecCs.td|jstd||jr*|dS)NzQueue.join_thread()zQueue {0!r} not closed)r r4r<r=r2)r!r$r$r% join_threadszQueue.join_threadcCs6tdd|_y|jWntk r0YnXdS)NzQueue.cancel_join_thread()T)r r3r2ZcancelAttributeError)r!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 r0clearr+ZThreadr_feedr-r7rrrSr_on_queue_feeder_errorrr1Zdaemonstartr3r _finalize_joinweakrefrefr2_finalize_closer5)r!r$r$r%r?s*        zQueue._start_threadcCs4td|}|dk r(|tdntddS)Nzjoining queue threadz... queue thread joinedz... queue thread already dead)r join)ZtwrZthreadr$r$r%r`s  zQueue._finalize_joinc Cs.td||t|WdQRXdS)Nztelling queue thread to quit)r r@ _sentinelrA)buffernotemptyr$r$r%rcs zQueue._finalize_closec CsRtd|j}|j} |j} |j} t} tjdkr<|j} |j}nd} x y|z|sX| Wd| Xybx\| }|| krtd|dSt |}| dkr||qh| z ||Wd|XqhWWnt k rYnXWqDt k rH}zJ|rt |ddt jkrdStr&td|dS||||Wdd}~XYqDXqDWdS)Nz$starting thread to feed data to piperz%feeder thread got sentinel -- exitingerrnorzerror in queue thread: %s)r r>rFwaitpopleftrerrrHdumps IndexError ExceptiongetattrrhZEPIPErr )rfrgr6Z writelockrSZ ignore_epipeonerrorZ queue_semZnacquireZnreleaseZnwaitZbpopleftsentinelZwacquireZwreleaserBer$r$r%r]sR       z Queue._feedcCsddl}|dS)z Private API hook called when feeding data in the background thread raises an exception. For overriding by concurrent.futures. rN) traceback print_exc)rqrBrrr$r$r%r^ szQueue._on_queue_feeder_error)r)TN)TN)__name__ __module__ __qualname__r&r(r*r rErKrMrNrPrQrRrSrTrVr? staticmethodr`rcr]r^r$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_tasksr,_cond)r!r"r#r$r$r%r& s zJoinableQueue.__init__cCst||j|jfS)N)rr(ryrx)r!r$r$r%r(%szJoinableQueue.__getstate__cCs,t||dd|dd\|_|_dS)N)rr*ryrx)r!r)r$r$r%r*(szJoinableQueue.__setstate__TNc Cs|jrtd||j||s&t|jJ|j8|jdkrH| |j ||j |jWdQRXWdQRXdS)NzQueue {0!r} is closed)r4r<r=rr>rr-ryr1r?r0r@rxrFrA)r!rBrCrDr$r$r%rE,s   zJoinableQueue.putc Cs@|j0|jdstd|jjr2|jWdQRXdS)NFz!task_done() called too many times)ryrxr> ValueErrorrLrOZ notify_all)r!r$r$r% task_done8s   zJoinableQueue.task_donec Cs,|j|jjs|jWdQRXdS)N)ryrxrLrOri)r!r$r$r%rd?s zJoinableQueue.join)r)TN) rtrurvr&r(r*rEr|rdr$r$r$r%rs   c@s<eZdZddZddZddZddZd d Zd d Zd S)rcCsHtjdd\|_|_||_|jj|_tj dkr:d|_ n ||_ dS)NF)rr) rrrrrrr:r;rrr)r!r#r$r$r%r&Js    zSimpleQueue.__init__cCs | S)N)r;)r!r$r$r%rNSszSimpleQueue.emptycCst||j|j|j|jfS)N)r r'rrrr)r!r$r$r%r(Vs zSimpleQueue.__getstate__cCs"|\|_|_|_|_|jj|_dS)N)rrrrr:r;)r!r)r$r$r%r*ZszSimpleQueue.__setstate__c Cs&|j|j}WdQRXt|S)N)rrr8rHrI)r!rJr$r$r%rK^szSimpleQueue.getc CsDt|}|jdkr"|j|n|j|j|WdQRXdS)N)rHrkrrr6)r!rBr$r$r%rEds   zSimpleQueue.putN) rtrurvr&rNr(r*rKrEr$r$r$r%rHs  )__all__rrr+r.rGrarhZqueuerrZ_multiprocessingrr Z reductionZForkingPicklerrHutilr r r r robjectrrerrr$r$r$r% s$   s *