# TR5nU[R:XagUv M+7fr$)getrr)itemr1s rinput_queue_iterable%_worker..input_queue_iterable7s0"(<,,, s-0zReader worker z finishedr)logginginfoosgetpidr(r r r-put Exception exceptionrrclose join_threadget_lockvalue) r0r1r2r3r4r!workerr9processed_itemes ` r_workerrJ!s~ LL?9+VBIIK=AB!<##-f- hsm %nn-A-CDN   ^ ,E !  "  ( ( *  $ $ ) $+  ~i[ :;) -!++,,-"+ * !  "  ( ( *  $ $ ) $+ * *  ~i[ :;sOADE! E5EE2EE2! E/2AHG" H" G0 ,"Hc \rSrSrS\4S\S\\S\S-S\4Sjjr S\ S S4S jr S \ \ S \ S\ S \ \ 4S jr S \ \ S \ S\ S \ \ 4SjrS \ \ S \ S\ S \ \ 4SjrSSjrSS\S-S S4SjjrSSjrSSjrSrg)ParallelWorkerPoolWN num_workersrG start_methodmax_internal_batch_sizecX lXlSUlSUl[ U5Ul/UlURU-UlSUlSUl g)NF) r0rNr1r2rctx processes queue_sizeemergency_shutdownr3)r,rNrGrOrPs r__init__ParallelWorkerPool.__init__XsV#&)-*. +L 9,.**-DD"'48rr!r"c URRUR5UlURRUR5UlURR SUR 5n[U[5(deX l [SUR 5Hn[URS5(deURR[URURURURUUR54S9nUR!5 UR"R%U5 M g)NirProcess)targetr )rRrrTr1r2ValuerN isinstance BaseValuer3rangehasattrrZrJr0copyr(rSappend)r,r! ctx_valuer4r-s rr(ParallelWorkerPool.startis88>>$//: HHNN4??;HHNN3(8(89 )Y////"+q$"2"23I488Y// //hh&&%%$$%%++KKM ' G MMO NN ! !' *4rstreamr c/f# UR"S 0UD6 URcS5eURcS5eSnSnUHnUR5 XE- UR:aURR 5nOURR[S9nUb8U[R:XaUR5 [S5eUv US- nURRU5 US- nM [UR5H,n URR[R 5 M. XT:a\URR[S9nU[R:XaUR5 [S5eUv US- nXT:aM\URcS5eURcS5eUR#5 URR%5 URR%5 UR&(a5URR)5 URR)5 gURR+5 URR+5 g![ a SnGNf=f![ anUR5 UeSnAff=f!URcS5eURcS5eUR#5 URR%5 URR%5 UR&(a5URR)5 URR)5 fURR+5 URR+5 f=f7f) NzInput queue was not initializedz Output queue was not initializedrtimeoutzThread unexpectedly terminatedr;zInput queue is NonezOutput queue is Noner)r(r1r2check_worker_healthrT get_nowaitr r7processing_timeoutjoin_or_terminaterr RuntimeErrorr@r_rNrjoinrCrUcancel_join_threadrD) r,rer r!pushedreadr8out_itemrI_s r unordered_map ParallelWorkerPool.unordered_maps.4 0 JJ  ##/ R1R R/$$0 T2T T0FD((*=4??2(#'#4#4#?#?#A #'#4#4#8#8AS#8#T '<#5#55..0*+KLL"NAID  $$T*! +.4++,  $$\%6%67--,,009K0L|111**,&'GHH  -##/ F1F F/$$0 H2H H0 IIK    " " $    # # %&&  335!!446  ,,.!!--/O!(#'( ! ..0 0##/ F1F F/$$0 H2H H0 IIK    " " $    # # %&&  335!!446  ,,.!!--/shN1A%K*JKJ"#DK(C'N1 JKJK" K,J>>KKC(N..N1c@UR"[U5/UQ70UD6$r$)rt enumerate)r,rer r!s rsemi_ordered_map#ParallelWorkerPool.semi_ordered_maps"!!)F"3EdEfEErc/# [[5nSnUR"U/UQ70UD6H/upgXtU'XT;dMURU5v US- nXT;aMM1 g7f)Nrr;)rintrxpop)r,rer r!buffer next_expectedidxr8s r ordered_mapParallelWorkerPool.ordered_mapsdS! ..vGGGIC3K)jj//"  )Hs4AAAcURHfnUR5(aMURS:wdM,SUlUR 5 [ SUR SUR35e g)z: Checks if any worker process has terminated unexpectedly rTz Worker PID: z# terminated unexpectedly with code N)rSis_aliveexitcoderUrlrmpidr,r-s rri&ParallelWorkerPool.check_worker_healthsl~~G##%%'*:*:a*?*.'&&(""7;;-/RSZScScRde &rrhcSUlURH9nURUS9 UR5(dM)UR 5 M; URR 5 g)z- Emergency shutdown @param timeout: @return: TrgN)rUrSrnr terminateclear)r,rhr-s rrl$ParallelWorkerPool.join_or_terminatesU #'~~G LLL )!!!!#& rc~URHnUR5 M URR5 gr$)rSrnrrs rrnParallelWorkerPool.joins)~~G LLN& rcxURH*nUR5(dMUR5 M, g)a Terminate processes if the user hasn't joined. This is necessary as leaving stray processes running can corrupt shared state. In brief, we've observed shared memory counters being reused (when the memory was free from the perspective of the parent process) while the stray workers still held a reference to them. For a discussion of using destructors in Python in this manner, see https://eli.thegreenplace.net/2009/06/12/safely-using-destructors-in-python/. N)rSrrrs r__del__ParallelWorkerPool.__del__s-~~G!!!!#&r) rRrUr1r3rNr2rSrTr0)r"N)r;)rrrrMAX_INTERNAL_BATCH_SIZEr{r rstrrVr r(r rtrxrrirlrnrrrrrrLrLWs $('> 99V 9Dj 9 "% 9"+c+d+050HSM50#5050QYZ]Q^50nFx}FSFCFT\]`TaF#(3-##s#xX[}#  t D  $rrLr$)r<r> collectionsrenumrmultiprocessingrrmultiprocessing.contextrmultiprocessing.processrmultiprocessing.sharedctypesr r^queuer typingr r r rkrrrrr{dictrJrLrrrrs #.//B&&3 $$%) 3<v,3<3<3<" 3<  3< cNT ! 3< 3