B ^ ÇcÒ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 ú%lib/python3.7/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_exceptionr1r2Z completedÚtaskÚjobÚiÚfuncr ÚkwdsÚresultÚeÚwrappedr r r Úworker]sJ       $ rHcCs|‚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?ZmaxtasksperchildÚ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)rZcleanedrBrHr 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. )rLr rKZ PoolWorkerTz added workerN)rrr]rsr^rKrHrmrgrVrWrUÚ_wrap_exceptionÚappendÚnameÚreplacercrdrr8)rrBÚwr r r r_ãs  zPool._repopulate_poolcCs| ¡r| ¡dS)zEClean up any exited workers and start replacements for them. N)rur_)rr r r Ú_maintain_poolôszPool._maintain_poolcCs4|j ¡|_|j ¡|_|jjj|_|jjj|_ dS)N) rJrPrmrgr,Úsendrfr5Úrecvrj)rr r r rNús   zPool._setup_queuescCs| |||¡ ¡S)zT Equivalent of `func(*args, **kwds)`. Pool must be running. )Ú apply_asyncr2)rrCr rDr 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)rrCÚ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)rrCrr‚r r r rsz Pool.starmapcCs| ||t|||¡S)z= Asynchronous version of `starmap()` method. )r€r)rrCrr‚Ú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_jobrCrrBÚxrFr 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)rTrSrZÚ IMapIteratorrRrQr1r‰Ú_jobÚ _set_lengthr0rÚ _get_tasksr)rrCrr‚rEÚ 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 )rŠr‹rŒr r r rbsz&Pool.imap_unordered..N)rTrSrZÚIMapUnorderedIteratorrRrQr1r‰rrr0rr‘r)rrCrr‚rEr’r r r Úimap_unorderedGs&      zPool.imap_unorderedcCsB|jtkrtdƒ‚t|j||ƒ}|j |jd|||fgdf¡|S)z; Asynchronous version of `apply()` method. zPool not runningrN)rTrSrZÚ ApplyResultrRrQr1r)rrCr rDrƒr„rEr r r r~ds  zPool.apply_asynccCs| ||t|||¡S)z9 Asynchronous version of `map()` method. )r€r)rrCrr‚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„)rTrSrZr3r Údivmodrsr^rr‘Ú MapResultrRrQr1r‰r) rrCrZmapperr‚rƒr„Zextrar’rEr r r r€ws&   zPool._map_asynccCsTt ¡}x0|jtks$|jr8|jtkr8| ¡t d¡q W|j   d¡t   d¡dS)Ngš™™™™™¹?zworker handler exiting) r`Úcurrent_threadrTrSrRÚ TERMINATEr{ÚtimeÚsleeprQr1rr8)ÚpoolÚthreadr r r ra•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) r`rœÚiterr2rTrr8r9Ú_setÚKeyErrorr1r7) Ú taskqueuer1r=r Úcacher¡ZtaskseqZ set_lengthr@rFrAÚidxÚpr r r re¢sB          zPool._handle_tasksc CsÈt ¡}xªy |ƒ}Wn"ttfk r6t d¡dSX|jr\|jtksPtdƒ‚t d¡P|dkrpt d¡P|\}}}y||  ||¡Wnt k r¢YnXd}}}q Wxœ|rP|jtkrPy |ƒ}Wn"ttfk rôt d¡dSX|dkr t d¡q¶|\}}}y||  ||¡Wnt k r@YnXd}}}q¶Wt |dƒr°t d¡y,x&t dƒD]}|j  ¡s†P|ƒqtWWnttfk r®YnXt d t|ƒ|j¡dS) Nz.result handler got EOFError/OSError -- exitingzThread not in TERMINATEz,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)r`rœr7r6rr8rTrr/r£r¤r3rrr5Úpollrs)r=r2r¦r¡r@rArBÚobjr r r riÐs\             zPool._handle_resultsccs4t|ƒ}x&tt ||¡ƒ}|s"dS||fVq WdS)N)r¢ÚtuplerÚislice)rCÚ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)rr8rTrSÚCLOSErb)rr r r r4s  z Pool.closecCs$t d¡t|_t|j_| ¡dS)Nzterminating pool)rr8rrTrbrn)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) rr8rTrSrZr±rrbrrhrkr^)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)rr8rrTr¶rsr´r/r1r`rœrr3rtr²Úpid) Úclsr¥r<r=r Zworker_handlerrµZresult_handlerr¦r¨r r r rl;s<                zPool._terminate_poolcCs|S)Nr )rr r r Ú __enter__kszPool.__enter__cCs | ¡dS)N)r²)rÚexc_typeZexc_valZexc_tbr r r Ú__exit__nsz Pool.__exit__)NNr NN)N)N)NNN)r)r)NNN)NNN)#rrrr*rvrKrrur_r{rNrr rr…r‰r“r•r~r—r€Ú staticmethodrarerir‘r"r4r²rr¶Ú classmethodrlr¹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) r`ZEventÚ_eventÚnextÚ job_counterrrRÚ _callbackÚ_error_callback)rr¦rƒr„r r r rws   zApplyResult.__init__cCs |j ¡S)N)r¾Zis_set)rr r r ÚreadyszApplyResult.readycCs| ¡std |¡ƒ‚|jS)Nz{0!r} not ready)rÃrZr0Ú_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¾ÚsetrRr)rrBr«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ÁrRrr¾rÉrÂ)rrBZsuccess_resultÚsuccessrEr 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)r`Z ConditionZLockÚ_condr¿rÀrrRÚ 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ÒrwrÕÚpopÚnotifyrÔrRr)rrBr«r r r r£ðs      zIMapIterator._setc Cs<|j,||_|j|jkr.|j ¡|j|j=WdQRXdS)N)rÏrÔrÓrÛrRr)rrÍr r r rs   zIMapIterator._set_length)N) rrrrrÖr¿Ú__next__r£rr r r r rŽËs   rŽc@seZdZdd„ZdS)r”c CsP|j@|j |¡|jd7_|j ¡|j|jkrB|j|j=WdQRXdS)Nr)rÏrÒrwrÓrÛrÔrRr)rrBr«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)rK)ZdummyrK)r rDrKr r r rKs zThreadPool.ProcessNr cCst ||||¡dS)N)rr)rror>r?r r r r!szThreadPool.__init__cCs,t ¡|_t ¡|_|jj|_|jj|_dS)N)rOrPrmrgr1rfr2rj)rr r r rN$s   zThreadPool._setup_queuescCsNyx|jddqWWntjk r,YnXxt|ƒD]}| d¡q8WdS)NF)Úblock)r2rOZEmptyrrr1)r<rµr¯rBr r r r¶*szThreadPool._help_stuff_finish)NNr ) rrrrvr¼rKrrNr¶r r r r rs   )Nr NF)"Ú__all__r`rOrrÐrXržrrrrrrSr±rÚcountrÀrrr9rrr!r$rHr:Úobjectrr–Z AsyncResultr›rŽr”rr r r r Ú s@   ,f')@