B 18™\Òhã@s0ddgZddlZddlZddlZddlZddlZddlZddlZddlm Z ddlm Z m Z dZ dZ dZe ¡Zdd „Zd d „ZGd d „d eƒZGdd„dƒZdd„ZGdd„deƒZd$dd„Zdd„ZGdd„deƒZGdd„deƒZeZGdd„deƒZGdd „d eƒZGd!d"„d"eƒZ Gd#d„deƒZ!dS)%ÚPoolÚ ThreadPooléNé)Úutil)Ú get_contextÚ TimeoutErrorécCs tt|ŽƒS)N)ÚlistÚmap)Úargs©r úX/oak/stanford/groups/akundaje/marinovg/programs/Python-3.7.3/Lib/multiprocessing/pool.pyÚmapstar+srcCstt |d|d¡ƒS)Nrr)r Ú itertoolsÚstarmap)r r r r Ú starmapstar.src@seZdZdd„Zdd„ZdS)ÚRemoteTracebackcCs ||_dS)N)Útb)Úselfrr r r Ú__init__6szRemoteTraceback.__init__cCs|jS)N)r)rr r r Ú__str__8szRemoteTraceback.__str__N)Ú__name__Ú __module__Ú __qualname__rrr r r r r5src@seZdZdd„Zdd„ZdS)ÚExceptionWithTracebackcCs0t t|ƒ||¡}d |¡}||_d||_dS)NÚz """ %s""")Ú tracebackÚformat_exceptionÚtypeÚjoinÚexcr)rr rr r r r<s zExceptionWithTraceback.__init__cCst|j|jffS)N)Ú rebuild_excr r)rr r r Ú __reduce__Asz!ExceptionWithTraceback.__reduce__N)rrrrr"r r r r r;srcCst|ƒ|_|S)N)rÚ __cause__)r rr r r r!Ds r!cs0eZdZdZ‡fdd„Zdd„Zdd„Z‡ZS)ÚMaybeEncodingErrorzVWraps possible unpickleable errors, so they can be safely sent through the socket.cs.t|ƒ|_t|ƒ|_tt|ƒ |j|j¡dS)N)Úreprr ÚvalueÚsuperr$r)rr r&)Ú __class__r r rPs  zMaybeEncodingError.__init__cCsd|j|jfS)Nz(Error sending result: '%s'. Reason: '%s')r&r )rr r r rUszMaybeEncodingError.__str__cCsd|jj|fS)Nz<%s: %s>)r(r)rr r r Ú__repr__YszMaybeEncodingError.__repr__)rrrÚ__doc__rrr)Ú __classcell__r r )r(r r$Ls r$r Fc CsÌ|dk r(t|tƒr|dks(td |¡ƒ‚|j}|j}t|dƒrR|j ¡|j  ¡|dk rb||Žd}xP|dks‚|r¸||kr¸y |ƒ} Wn$t t fk r°t   d¡PYnX| dkrÆt   d¡P| \} } } } }yd| | |Žf}WnHtk r.}z(|r| tk rt||jƒ}d|f}Wdd}~XYnXy|| | |fƒWnRtk r”}z2t||dƒ}t   d |¡|| | d|ffƒWdd}~XYnXd} } }} } }|d7}qjWt   d |¡dS) NrzMaxtasks {!r} is not validÚ_writerrz)worker got EOFError or OSError -- exitingzworker got sentinel -- exitingTFz0Possible encoding error while sending result: %szworker exiting after %d tasks)Ú isinstanceÚintÚAssertionErrorÚformatÚputÚgetÚhasattrr,ÚcloseÚ_readerÚEOFErrorÚOSErrorrÚdebugÚ ExceptionÚ_helper_reraises_exceptionrÚ __traceback__r$)ÚinqueueÚoutqueueÚ initializerÚinitargsZmaxtasksZwrap_exceptionr1r2Ú completedÚtaskÚjobÚiÚfuncr ÚkwdsÚresultÚeÚwrappedr r r Úworker]sJ       $ rIcCs|‚dS)z@Pickle-able helper function for use by _guarded_task_generation.Nr )Úexr r r r:Šsr:c@s,eZdZdZdZdd„Zd>dd„Zd d „Zd d „Zd d„Z dd„Z difdd„Z d?dd„Z d@dd„Z dAdd„Zdd„ZdBdd„ZdCdd„Zdiddfd d!„ZdDd"d#„ZdEd$d%„Zed&d'„ƒZed(d)„ƒZed*d+„ƒZed,d-„ƒZd.d/„Zd0d1„Zd2d3„Zd4d5„Zed6d7„ƒZed8d9„ƒZ d:d;„Z!dr?ÚmaxtasksperchildÚcontextr r r r›sT         z Pool.__init__cCsZd}xPttt|jƒƒƒD]:}|j|}|jdk rt d|¡| ¡d}|j|=qW|S)zCleanup after any worker processes which have exited due to reaching their specified lifetime. Returns True if any workers were cleaned up. FNzcleaning up worker %dT)ÚreversedÚrangeÚlenr`Úexitcoderr8r)rÚcleanedrCrIr r r Ú_join_exited_workersÔs   zPool._join_exited_workersc Cs|xvt|jt|jƒƒD]^}|jt|j|j|j|j |j |j fd}|j  |¡|j  dd¡|_ d|_| ¡t d¡qWdS)z€Bring the number of pool processes up to the specified number, for use after reaping workers which have exited. )rMr rLZ PoolWorkerTz added workerN)rwr_rxr`rLrIrqrjrXrYrWÚ_wrap_exceptionÚappendÚnameÚreplacerfrgrr8)rrCÚwr r r raãs  zPool._repopulate_poolcCs| ¡r| ¡dS)zEClean up any exited workers and start replacements for them. N)r{ra)rr r r Ú_maintain_poolôszPool._maintain_poolcCs4|j ¡|_|j ¡|_|jjj|_|jjj|_ dS)N) rKrRrqrjr,Úsendrir5Úrecvrm)rr r r rPús   zPool._setup_queuescCs| |||¡ ¡S)zT Equivalent of `func(*args, **kwds)`. Pool must be running. )Ú apply_asyncr2)rrDr rEr r r Úapplysz Pool.applycCs| ||t|¡ ¡S)zx Apply `func` to each element in `iterable`, collecting the results in a list that is returned. )Ú _map_asyncrr2)rrDÚiterableÚ chunksizer r r r szPool.mapcCs| ||t|¡ ¡S)zÌ Like `map()` method but the elements of the `iterable` are expected to be iterables as well and will be unpacked as arguments. Hence `func` and (a, b) becomes func(a, b). )r†rr2)rrDr‡rˆr r r rsz Pool.starmapcCs| ||t|||¡S)z= Asynchronous version of `starmap()` method. )r†r)rrDr‡rˆÚcallbackÚerror_callbackr r r Ú starmap_asyncs zPool.starmap_asyncc csny0d}x&t|ƒD]\}}||||fifVqWWn8tk rh}z||dt|fifVWdd}~XYnXdS)zšProvides a generator of tasks for imap and imap_unordered with appropriate handling for iterables which throw exceptions during iteration.éÿÿÿÿrN)Ú enumerater9r:)rZ result_jobrDr‡rCÚxrGr r r Ú_guarded_task_generations zPool._guarded_task_generationrcCs¤|jtkrtdƒ‚|dkrFt|jƒ}|j | |j||¡|j f¡|S|dkr\td  |¡ƒ‚t   |||¡}t|jƒ}|j | |jt |¡|j f¡dd„|DƒSdS)zP Equivalent of `map()` -- can be MUCH slower than `Pool.map()`. zPool not runningrzChunksize must be 1+, not {0:n}css|]}|D] }|Vq qdS)Nr )Ú.0ÚchunkÚitemr r r ú EszPool.imap..N)rVrUr\Ú IMapIteratorrTrSr1rÚ_jobÚ _set_lengthr0rÚ _get_tasksr)rrDr‡rˆrFÚ task_batchesr r r Úimap)s(     z Pool.imapcCs¤|jtkrtdƒ‚|dkrFt|jƒ}|j | |j||¡|j f¡|S|dkr\td  |¡ƒ‚t   |||¡}t|jƒ}|j | |jt |¡|j f¡dd„|DƒSdS)zL Like `imap()` method but ordering of results is arbitrary. zPool not runningrzChunksize must be 1+, not {0!r}css|]}|D] }|Vq qdS)Nr )rr‘r’r r r r“bsz&Pool.imap_unordered..N)rVrUr\ÚIMapUnorderedIteratorrTrSr1rr•r–r0rr—r)rrDr‡rˆrFr˜r r r Úimap_unorderedGs&      zPool.imap_unorderedcCsB|jtkrtdƒ‚t|j||ƒ}|j |jd|||fgdf¡|S)z; Asynchronous version of `apply()` method. zPool not runningrN)rVrUr\Ú ApplyResultrTrSr1r•)rrDr rEr‰rŠrFr r r r„ds  zPool.apply_asynccCs| ||t|||¡S)z9 Asynchronous version of `map()` method. )r†r)rrDr‡rˆr‰rŠr r r Ú map_asyncoszPool.map_asyncc Csª|jtkrtdƒ‚t|dƒs$t|ƒ}|dkrTtt|ƒt|jƒdƒ\}}|rT|d7}t|ƒdkrdd}t  |||¡}t |j |t|ƒ||d} |j   | | j||¡df¡| S)zY Helper function to implement map, starmap and their async counterparts. zPool not runningÚ__len__Nérr)rŠ)rVrUr\r3r Údivmodrxr`rr—Ú MapResultrTrSr1rr•) rrDr‡Zmapperrˆr‰rŠÚextrar˜rFr r r r†ws&   zPool._map_asynccCsTt ¡}x0|jtks$|jr8|jtkr8| ¡t d¡q W|j   d¡t   d¡dS)Ngš™™™™™¹?zworker handler exiting) rbÚcurrent_threadrVrUrTÚ TERMINATErÚtimeÚsleeprSr1rr8)ÚpoolÚthreadr r r rd•s  zPool._handle_workersc Csjt ¡}xöt|jdƒD]Ü\}}d}zÀx¸|D]„}|jr@t d¡Py ||ƒWq*tk r¬} zB|dd…\} } y||  | d| f¡Wnt k ršYnXWdd} ~ XYq*Xq*W|rÜt d¡|rÌ|dnd} || dƒwPWdd}}} XqWt d¡y:t d¡|  d¡t d ¡x|D]} |dƒq&WWn t k rZt d ¡YnXt d ¡dS) Nz'task handler found thread._state != RUNrFzdoing set_length()rrŒztask handler got sentinelz/task handler sending sentinel to result handlerz(task handler sending sentinel to workersz/task handler got OSError when sending sentinelsztask handler exiting) rbr£Úiterr2rVrr8r9Ú_setÚKeyErrorr1r7) Ú taskqueuer1r=r§Úcacher¨ZtaskseqZ set_lengthrArGrBÚidxÚpr r r rh¢sB          zPool._handle_tasksc Cs¶t ¡}x˜y |ƒ}Wn"ttfk r6t d¡dSX|jrJt d¡P|dkr^t d¡P|\}}}y|| ||¡Wntk rYnXd}}}q Wxœ|r>|jt kr>y |ƒ}Wn"ttfk rât d¡dSX|dkrút d¡q¤|\}}}y|| ||¡Wntk r.YnXd}}}q¤Wt |dƒržt d¡y,x&t dƒD]}|j   ¡stP|ƒqbWWnttfk rœYnXt dt|ƒ|j¡dS) Nz.result handler got EOFError/OSError -- exitingz,result handler found thread._state=TERMINATEzresult handler got sentinelz&result handler ignoring extra sentinelr5z"ensuring that outqueue is not fullé z7result handler exiting: len(cache)=%s, thread._state=%s)rbr£r7r6rr8rVrªr«r¤r3rwr5Úpollrx)r=r2r­r¨rArBrCÚobjr r r rlÐsZ             zPool._handle_resultsccs4t|ƒ}x&tt ||¡ƒ}|s"dS||fVq WdS)N)r©ÚtuplerÚislice)rDÚitÚsizerŽr r r r— s zPool._get_taskscCs tdƒ‚dS)Nz:pool objects cannot be passed between processes or pickled)ÚNotImplementedError)rr r r r"szPool.__reduce__cCs&t d¡|jtkr"t|_t|j_dS)Nz closing pool)rr8rVrUÚCLOSEre)rr r r r4s  z Pool.closecCs$t d¡t|_t|j_| ¡dS)Nzterminating pool)rr8r¤rVrerr)rr r r Ú terminate s zPool.terminatecCsnt d¡|jtkrtdƒ‚n|jttfkr4tdƒ‚|j ¡|j  ¡|j  ¡x|j D] }| ¡qZWdS)Nz joining poolzPool is still runningzIn unknown state) rr8rVrUr\r¸r¤rerrkrnr`)rr¯r r r r&s       z Pool.joincCsDt d¡|j ¡x*| ¡r>|j ¡r>|j ¡t  d¡qWdS)Nz7removing tasks from inqueue until task handler finishedr) rr8Z_rlockÚacquireÚis_aliver5r±rƒr¥r¦)r<Ú task_handlerr¶r r r Ú_help_stuff_finish2s    zPool._help_stuff_finishc CsJt d¡t|_t|_t d¡| ||t|ƒ¡| ¡sNt|ƒdkrNtdƒ‚t|_| d¡t d¡t   ¡|k r||  ¡|r¸t |ddƒr¸t d¡x|D]} | j dkrž|  ¡qžWt d¡t   ¡|k rÖ|  ¡t d ¡t   ¡|k rô|  ¡|rFt |ddƒrFt d ¡x0|D](} |  ¡rt d | j¡|   ¡qWdS) Nzfinalizing poolz&helping task handler/workers to finishrz.Cannot have cache with result_hander not alivezjoining worker handlerr¹zterminating workerszjoining task handlerzjoining result handlerzjoining pool workerszcleaning up worker %d)rr8r¤rVr½rxr»r/r1rbr£rr3ryr¹Úpid) Úclsr¬r<r=r§Zworker_handlerr¼Zresult_handlerr­r¯r r r rp;s<                zPool._terminate_poolcCs|S)Nr )rr r r Ú __enter__kszPool.__enter__cCs | ¡dS)N)r¹)rÚexc_typeÚexc_valÚexc_tbr r r Ú__exit__nsz Pool.__exit__)NNr NN)N)N)NNN)r)r)NNN)NNN)#rrrr*r|rLrr{rarrPr…r rr‹rr™r›r„rr†Ú staticmethodrdrhrlr—r"r4r¹rr½Ú classmethodrprÀrÄr r r r r’sF 8         . <  0c@s@eZdZdd„Zdd„Zdd„Zddd „Zdd d „Zd d „ZdS)rœcCs4t ¡|_ttƒ|_||_||_||_|||j<dS)N) rbÚEventÚ_eventÚnextÚ job_counterr•rTÚ _callbackÚ_error_callback)rr­r‰rŠr r r rws   zApplyResult.__init__cCs |j ¡S)N)rÈÚis_set)rr r r ÚreadyszApplyResult.readycCs| ¡std |¡ƒ‚|jS)Nz{0!r} not ready)rÎr\r0Ú_success)rr r r Ú successful‚szApplyResult.successfulNcCs|j |¡dS)N)rÈÚwait)rÚtimeoutr r r rчszApplyResult.waitcCs,| |¡| ¡st‚|jr"|jS|j‚dS)N)rÑrÎrrÏÚ_value)rrÒr r r r2Šs  zApplyResult.getcCsT|\|_|_|jr$|jr$| |j¡|jr<|js<| |j¡|j ¡|j|j=dS)N)rÏrÓrËrÌrÈÚsetrTr•)rrCr²r r r rª“s      zApplyResult._set)N)N) rrrrrÎrÐrÑr2rªr r r r rœus   rœc@seZdZdd„Zdd„ZdS)r¡cCsftj||||dd|_dg||_||_|dkrLd|_|j ¡||j=n||t ||ƒ|_dS)N)rŠTr) rœrrÏrÓÚ _chunksizeÚ _number_leftrÈrÔr•Úbool)rr­rˆÚlengthr‰rŠr r r r¤s    zMapResult.__init__cCsº|jd8_|\}}|rp|jrp||j||j|d|j…<|jdkr¶|jrZ| |j¡|j|j=|j ¡nF|s†|jr†d|_||_|jdkr¶|j r¢|  |j¡|j|j=|j ¡dS)NrrF) rÖrÏrÓrÕrËrTr•rÈrÔrÌ)rrCZsuccess_resultÚsuccessrFr r r rª±s"         zMapResult._setN)rrrrrªr r r r r¡¢s r¡c@s:eZdZdd„Zdd„Zd dd„ZeZdd „Zd d „ZdS) r”cCsJt t ¡¡|_ttƒ|_||_t  ¡|_ d|_ d|_ i|_ |||j<dS)Nr)rbÚ ConditionÚLockÚ_condrÉrÊr•rTÚ collectionsÚdequeÚ_itemsÚ_indexÚ_lengthÚ _unsorted)rr­r r r rÍs  zIMapIterator.__init__cCs|S)Nr )rr r r Ú__iter__×szIMapIterator.__iter__Nc Cs¨|j„y|j ¡}Wnntk r„|j|jkr8td‚|j |¡y|j ¡}Wn,tk r~|j|jkrttd‚td‚YnXYnXWdQRX|\}}|r |S|‚dS)N) rÜrßÚpopleftÚ IndexErrorràráÚ StopIterationrÑr)rrÒr’rÙr&r r r rÉÚs"   zIMapIterator.nextc Cs |j|j|krr|j |¡|jd7_x8|j|jkrd|j |j¡}|j |¡|jd7_q.W|j ¡n ||j|<|j|jkr’|j|j =WdQRXdS)Nr) rÜràrßr}râÚpopÚnotifyrárTr•)rrCr²r r r rªðs      zIMapIterator._setc Cs<|j,||_|j|jkr.|j ¡|j|j=WdQRXdS)N)rÜráràrèrTr•)rrØr r r r–s   zIMapIterator._set_length)N) rrrrrãrÉÚ__next__rªr–r r r r r”Ës   r”c@seZdZdd„ZdS)ršc CsP|j@|j |¡|jd7_|j ¡|j|jkrB|j|j=WdQRXdS)Nr)rÜrßr}ràrèrárTr•)rrCr²r r r rª s    zIMapUnorderedIterator._setN)rrrrªr r r r rš sršc@s:eZdZdZedd„ƒZd dd„Zdd „Zed d „ƒZdS) rFcOsddlm}|||ŽS)Nr)rL)ÚdummyrL)r rErLr r r rLs zThreadPool.ProcessNr cCst ||||¡dS)N)rr)rrsr>r?r r r r!szThreadPool.__init__cCs,t ¡|_t ¡|_|jj|_|jj|_dS)N)rQrRrqrjr1rir2rm)rr r r rP$s   zThreadPool._setup_queuescCsNyx|jddqWWntjk r,YnXxt|ƒD]}| d¡q8WdS)NF)Úblock)r2rQÚEmptyrwr1)r<r¼r¶rCr r r r½*szThreadPool._help_stuff_finish)NNr ) rrrr|rÅrLrrPr½r r r r rs   )Nr NF)"Ú__all__rbrQrrÝrZr¥rrrrrrUr¸r¤ÚcountrÊrrr9rrr!r$rIr:ÚobjectrrœÚ AsyncResultr¡r”ršrr r r r Ú s@   ,f')@