U :ezn@sdZddlZddlZddlmZddlZddlmZddlZddl Zddl m Z ddl Z ddl Z ddlmZddlZddlZddlZe ZdaGdd d Zd d Zd Zd ZGdddeZGdddZddZGdddeZGdddeZ GdddeZ!Gddde Z"ddZ#ddZ$d0d d!Z%d"d#Z&d$d%Z'd&d'Z(da)da*d(d)Z+d*d+Z,Gd,d-d-ej-Z.Gd.d/d/ej/Z0e1edS)1z"Brian Quinlan (brian@sweetapp.com)N)_base)Full)Queue)partialFc@s,eZdZddZddZddZddZd S) _ThreadWakeupcCstjdd\|_|_dS)NF)duplex)mpPipe_reader_writerselfr?/usr/local/python38/lib/python3.8/concurrent/futures/process.py__init__Rsz_ThreadWakeup.__init__cCs|j|jdSN)r closer r rrrrUs z_ThreadWakeup.closecCs|jddS)N)r send_bytesr rrrwakeupYsz_ThreadWakeup.wakeupcCs|jr|jqdSr)r poll recv_bytesr rrrclear\s z_ThreadWakeup.clearN)__name__ __module__ __qualname__rrrrrrrrrQsrcCs@datt}|D]\}}|q|D]\}}|q*dSNT)_global_shutdownlist_threads_wakeupsitemsrjoin)r _ thread_wakeuptrrr _python_exitas     r%=c@seZdZddZddZdS)_RemoteTracebackcCs ||_dSrtb)r r*rrrrzsz_RemoteTraceback.__init__cCs|jSrr)r rrr__str__|sz_RemoteTraceback.__str__N)rrrrr+rrrrr(ysr(c@seZdZddZddZdS)_ExceptionWithTracebackcCs0tt|||}d|}||_d||_dS)Nz """ %s""") tracebackformat_exceptiontyper!excr*)r r1r*rrrrs z _ExceptionWithTraceback.__init__cCst|j|jffSr) _rebuild_excr1r*r rrr __reduce__sz"_ExceptionWithTraceback.__reduce__N)rrrrr3rrrrr,sr,cCst||_|Sr)r( __cause__)r1r*rrrr2s r2c@seZdZddZdS) _WorkItemcCs||_||_||_||_dSr)futurefnargskwargs)r r6r7r8r9rrrrsz_WorkItem.__init__Nrrrrrrrrr5sr5c@seZdZdddZdS) _ResultItemNcCs||_||_||_dSr)work_id exceptionresult)r r<r=r>rrrrsz_ResultItem.__init__)NNr:rrrrr;sr;c@seZdZddZdS) _CallItemcCs||_||_||_||_dSr)r<r7r8r9)r r<r7r8r9rrrrsz_CallItem.__init__Nr:rrrrr?sr?cs*eZdZdfdd ZfddZZS) _SafeQueuercs||_tj||ddS)N)ctx)pending_work_itemssuperr)r max_sizerArB __class__rrrsz_SafeQueue.__init__cslt|trZtt|||j}tdd||_ |j |j d}|dk rh|j |nt||dS)Nz """ {}"""r-) isinstancer?r.r/r0 __traceback__r(formatr!r4rBpopr<r6 set_exceptionrC_on_queue_feeder_error)r eobjr* work_itemrErrrLs z!_SafeQueue._on_queue_feeder_error)r)rrrrrL __classcell__rrrErr@sr@cgs,t|}tt||}|s dS|VqdSr)ziptuple itertoolsislice) chunksize iterablesitchunkrrr _get_chunkss rYcsfdd|DS)Ncsg|] }|qSrr).0r8r7rr sz"_process_chunk..r)r7rXrr[r_process_chunks r]c Cs^z|t|||dWn@tk rX}z"t||j}|t||dW5d}~XYnXdS)N)r>r=r=)putr; BaseExceptionr,rH) result_queuer<r>r=rMr1rrr_sendback_results   rbc Cs|dk rtk r}z t ||j }t ||j|dW5d}~XYnXt ||j|d~~q)r`rLOGGERcriticalgetr_osgetpidr7r8r9r,rHrbr<) call_queuera initializerinitargs call_itemrrMr1rrr_process_workers$    "rpcCsv|r dSz|jdd}Wntjk r4YdSX||}|jrh|jt||j|j |j ddq||=qqdS)NFrdT) fullrhqueueEmptyr6set_running_or_notify_cancelr_r?r7r8r9)rBwork_idsrkr<rOrrr_add_call_item_to_queues" rvc s>dfdd}fdd}|j} |j} | | g} t||ddD} tj| | } d}d}| | krz| }d}Wqtk r}zt t |||j }W5d}~XYqXn| | krd}d}| |rl|dk rd _ d_dtd }|dk r td d |d |_|D]\}}|j|~q(| D]}|qR|dSt|tr|}|s|dSnL|dk r||jd}|dk r|jr|j|jn|j|j~~||r4z&dk rd_|s|WdSWntk r2YnXdq2dS)NcstpdkpjSr)r_shutdown_threadr)executorrr shutting_down@s z/_queue_management_worker..shutting_downc stddD}|}d}||kr|dkrt||D]6}zd|d7}Wq:tk rnYqrYq:Xq:tddD}qD] }|qdS)Ncss|]}|VqdSris_aliverZprrr FszD_queue_management_worker..shutdown_worker..rr&css|]}|VqdSrrzr|rrrr~Rs)sumvaluesrange put_nowaitrrr!)n_children_aliven_children_to_stopn_sentinels_sentir})rk processesrrshutdown_workerDs   z1_queue_management_worker..shutdown_workercSsg|] }|jqSr)sentinelr|rrrr\isz,_queue_management_worker..TFzKA child process terminated abruptly, the process pool is not usable anymorez^A process in the process pool was terminated abruptly while the future was running or pending.z ''' r-z''')r rvrr connectionwaitrecvr`r.r/r0rHr_brokenrwBrokenProcessPoolr(r!r4r r6rK terminaterGintrJr<r= set_resultr>r)executor_referencerrBwork_ids_queuerkrar#ryr result_reader wakeup_readerreadersworker_sentinelsreadycause is_broken result_itemrMbper<rOr}r)rkrxrr_queue_management_worker"s  (          rc Csjtrtrttdaztd}Wnttfk r<YdSX|dkrJdS|dkrVdSd|attdS)NTSC_SEM_NSEMS_MAXz@system provides too few semaphores (%d available, 256 necessary))_system_limits_checked_system_limitedNotImplementedErrorrisysconfAttributeError ValueError) nsems_maxrrr_check_system_limitss rccs&|D]}||r|VqqdSr)reverserJ)iterableelementrrr_chain_from_iterable_of_listssrc@s eZdZdS)rN)rrrrrrrrsrcsteZdZdddZddZddZd d Zejjj e_ ejjj e_ dd d fd d Z dddZ ejj j e _ Z S)ProcessPoolExecutorNrcCst|dkr6tpd|_tjdkrntt|j|_n8|dkrHtdn tjdkrh|tkrhtdt||_|dkr~t }||_ |dk rt |st d||_||_d|_i|_d|_t|_d|_d|_i|_|jt}t||j |jd|_d |j_||_t |_!t"|_#dS) Nr&win32rz"max_workers must be greater than 0zmax_workers must be <= zinitializer must be a callableF)rDrArBT)$rri cpu_count _max_workerssysplatformmin_MAX_WINDOWS_WORKERSrr get_context _mp_contextcallable TypeError _initializer _initargs_queue_management_thread _processesrw threadingLock_shutdown_lockr _queue_count_pending_work_itemsEXTRA_QUEUED_CALLSr@ _call_queue _ignore_epipe SimpleQueue _result_queuerrr _work_idsr_queue_management_thread_wakeup)r max_workers mp_contextrlrm queue_sizerrrrsP        zProcessPoolExecutor.__init__c Csv|jdkrr|jfdd}|tjtt|||j|j |j |j |j |jfdd|_d|j_ |j|jt|j<dS)NcSstjd|dS)Nz?Executor collected: triggering callback for QueueManager wakeup)rutildebugr)r"r#rrr weakref_cbBs zFProcessPoolExecutor._start_queue_management_thread..weakref_cbQueueManagerThread)targetr8nameT)rr_adjust_process_countrThreadrweakrefrefrrrrrdaemonstartr)r rrrr_start_queue_management_thread=s(     z2ProcessPoolExecutor._start_queue_management_threadcCsPtt|j|jD]8}|jjt|j|j|j |j fd}| ||j|j <qdS)N)rr8) rlenrrrProcessrprrrrrpid)r r"r}rrrrXsz)ProcessPoolExecutor._adjust_process_countc Os t|dkr|^}}}nV|s&tdnHd|krZ|d}|^}}ddl}|jdtddntdt|d|j|jrt|j|j rt d t rt d t }t||||}||j|j<|j|j|jd7_|j||W5QRSQRXdS) NzEdescriptor 'submit' of 'ProcessPoolExecutor' object needs an argumentr7rz.Passing 'fn' as keyword argument is deprecated) stacklevelz6submit expected at least 1 positional argument, got %dr&z*cannot schedule new futures after shutdownz6cannot schedule new futures after interpreter shutdown)rrrJwarningswarnDeprecationWarningrrrrw RuntimeErrorrrFuturer5rrrr_rrr)r8r9r r7rfwrrrsubmitcs<        zProcessPoolExecutor.submitr&)timeoutrUcs:|dkrtdtjtt|t|d|i|d}t|S)Nr&zchunksize must be >= 1.rU)r)rrCmaprr]rYr)r r7rrUrVresultsrErrrs zProcessPoolExecutor.mapTc Cs|j d|_W5QRX|jr6|j|r6|jd|_|jdk rd|j|r^|jd|_d|_ d|_ |jr|jd|_dSr) rrwrrrr!rr join_threadrr)r rrrrshutdowns"      zProcessPoolExecutor.shutdown)NNNr)T)rrrrrrrrExecutor__text_signature____doc__rrrPrrrErrs K $   r)NN)2 __author__atexitriconcurrent.futuresrrrrmultiprocessingrmultiprocessing.connectionmultiprocessing.queuesrrr functoolsrrSrr.WeakKeyDictionaryrrrr%rr Exceptionr(r,r2objectr5r;r?r@rYr]rbrprvrrrrrBrokenExecutorrrrregisterrrrr.sT        )&! P