]d@sddgZddlZddlZddlZddlZddlZddlZddlZddlm Z ddlm Z m Z dZ dZ dZejZdd Zd d ZGd d d eZGdddZddZGdddeZdfddddZddZGdddeZGdddeZeZGdddeZGdddeZGd d!d!eZ Gd"ddeZ!dS)#Pool ThreadPoolN)util) get_context TimeoutErrorcCstt|S)N)listmap)argsr )/opt/alt/python35/lib64/python3.5/pool.pymapstar+srcCsttj|d|dS)Nrr)r itertoolsstarmap)r r r r starmapstar.src@s(eZdZddZddZdS)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 r5s  rc@s(eZdZddZddZdS)ExceptionWithTracebackcCsDtjt|||}dj|}||_d||_dS)Nz """ %s""") tracebackformat_exceptiontypejoinexcr)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;s  rcCst||_|S)N)r __cause__)r rr r r r!Dsr!cs:eZdZfddZddZddZS)MaybeEncodingErrorcsAt||_t||_tt|j|j|jdS)N)reprr valuesuperr$r)rr r&) __class__r r rPszMaybeEncodingError.__init__cCsd|j|jfS)Nz(Error sending result: '%s'. Reason: '%s')r&r )rr r r rUs zMaybeEncodingError.__str__cCsd|jj|fS)Nz<%s: %s>)r(r)rr r r __repr__YszMaybeEncodingError.__repr__)rrrrrr)r r )r(r r$Ls  r$Fc'Cs|j}|j}t|dr;|jj|jj|dk rQ||d}x|dksx|r||kry |} Wn&ttfk rtj dPYnX| dkrtj dP| \} } } } }yd| | |f}WnUt k rM}z5|r/| t k r/t ||j }d|f}WYdd}~XnXy|| | |fWnbt k r}zBt||d}tj d||| | d|ffWYdd}~XnXd} } }} } }|d7}qZWtj d |dS) N_writerrz)worker got EOFError or OSError -- exitingzworker got sentinel -- exitingTFrz0Possible encoding error while sending result: %szworker exiting after %d tasks)putgethasattrr*close_readerEOFErrorOSErrorrdebug Exception_helper_reraises_exceptionr __traceback__r$)inqueueoutqueue initializerinitargsZmaxtasksZwrap_exceptionr+r,Z completedtaskjobifuncr kwdsresultewrappedr r r worker]sD      !     ,rBcCs |dS)Nr )exr r r r4sr4c@seZdZdZddZddfddddZddZd d Zd d Zd dZ fiddZ dddZ dddZ dddddZ ddZdddZdddZfiddddZdddd d!Zdddd"d#Zed$d%Zed&d'Zed(d)Zed*d+Zd,d-Zd.d/Zd0d1Zd2d3Zed4d5Zed6d7Zd8d9Z d:d;Z!dS)r r r rEsz Pool.ProcessNc Cs#|p t|_|jtj|_i|_t|_||_ ||_ ||_ |dkrvt j psd}|dkrtd|dk rt| rtd||_g|_|jtjdtjd|f|_d|j_t|j_|jjtjdtjd|j|j|j|j|jf|_d|j_t|j_|jjtjdtjd|j|j |jf|_!d|j!_t|j!_|j!jt"j#||j$d|j|j%|j|j|j|j|j!|jfdd|_&dS) Nrz&Number of processes must be at least 1zinitializer must be a callabletargetr T exitpriority)'rrD _setup_queuesqueueQueue _taskqueue_cacheRUN_state_maxtasksperchild _initializer _initargsos cpu_count ValueErrorcallable TypeError _processes_pool_repopulate_pool threadingThreadr_handle_workers_worker_handlerdaemonstart _handle_tasks _quick_put _outqueue _task_handler_handle_results _quick_get_result_handlerrFinalize_terminate_pool_inqueue _terminate)r processesr8r9maxtasksperchildcontextr r r rsT                              z Pool.__init__cCswd}xjttt|jD]M}|j|}|jdk r"tjd||jd}|j|=q"W|S)NFzcleaning up worker %dT)reversedrangelenrYexitcoderr2r)rZcleanedr<rBr r r _join_exited_workerss"  zPool._join_exited_workersc Csxt|jt|jD]}|jdtd|j|j|j|j |j |j f}|jj ||j jdd|_ d|_|jtjdqWdS)NrFr rEZ PoolWorkerTz added worker)rprXrqrYrErBrjrcrQrRrP_wrap_exceptionappendnamereplacer_r`rr2)rr<wr r r rZs#    zPool._repopulate_poolcCs|jr|jdS)N)rsrZ)rr r r _maintain_pools zPool._maintain_poolcCsL|jj|_|jj|_|jjj|_|jjj|_ dS)N) rD SimpleQueuerjrcr*sendrbr/recvrf)rr r r rIszPool._setup_queuescCs|j|||jS)N) apply_asyncr,)rr=r r>r r r applysz Pool.applycCs|j||t|jS)N) _map_asyncrr,)rr=iterable chunksizer r r r szPool.mapcCs|j||t|jS)N)rrr,)rr=rrr r r r sz Pool.starmapcCs|j||t|||S)N)rr)rr=rrcallbackerror_callbackr r r starmap_asyncszPool.starmap_asyncccsy>d}x1t|D]#\}}||||fifVqWWn@tk r}z ||dt|fifVWYdd}~XnXdS)Nr) enumerater3r4)rZ result_jobr=rr<xr@r r r _guarded_task_generations zPool._guarded_task_generationrcCs|jtkrtd|dkret|j}|jj|j|j|||j f|St j |||}t|j}|jj|j|jt ||j fdd|DSdS)NzPool not runningrcss"|]}|D] }|Vq qdS)Nr ).0chunkitemr r r @szPool.imap..) rOrNrU IMapIteratorrMrLr+r_job _set_lengthr _get_tasksr)rr=rrr? task_batchesr r r imap's        z Pool.imapcCs|jtkrtd|dkret|j}|jj|j|j|||j f|St j |||}t|j}|jj|j|jt ||j fdd|DSdS)NzPool not runningrcss"|]}|D] }|Vq qdS)Nr )rrrr r r r[sz&Pool.imap_unordered..) rOrNrUIMapUnorderedIteratorrMrLr+rrrrrr)rr=rrr?rr r r imap_unorderedBs        zPool.imap_unorderedcCs_|jtkrtdt|j||}|jj|jd|||fgdf|S)NzPool not runningr)rOrNrU ApplyResultrMrLr+r)rr=r r>rrr?r r r r}]s  +zPool.apply_asynccCs|j||t|||S)N)rr)rr=rrrrr r r map_asynchszPool.map_asyncc Cs|jtkrtdt|ds6t|}|dkrztt|t|jd\}}|rz|d7}t|dkrd}tj |||}t |j |t||d|} |j j |j| j||df| S)NzPool not running__len__rrr)rOrNrUr-r divmodrqrYrr MapResultrMrLr+rr) rr=rZmapperrrrextrarr?r r r rps&   (     zPool._map_asynccCsrtj}xB|jtks6|jrP|jtkrP|jtjdqW|j j dt j ddS)Ng?zworker handler exiting) r[current_threadrOrNrM TERMINATErytimesleeprLr+rr2)poolthreadr r r r]s  * zPool._handle_workersc Cstj}x+t|jdD] \}}d}zx|D]}|jrXtjdPy||Wq;tk r} zN|dd\} } y|| j| d| fWnt k rYnXWYdd} ~ Xq;Xq;W|rtjd|r|dnd } || dwPWdd}}} XqWtjdyFtjd|j dtjdx|D]} |dqkWWnt k rtjd YnXtjd dS) Nz'task handler found thread._state != RUNrFzdoing set_length()rztask handler got sentinelz/task handler sending sentinel to result handlerz(task handler sending sentinel to workersz/task handler got OSError when sending sentinelsztask handler exitingr) r[riterr,rOrr2r3_setKeyErrorr+r1) taskqueuer+r7rcacherZtaskseqZ set_lengthr:r@r;idxpr r r rasB            zPool._handle_taskscCs tj}xy |}Wn)ttfk rGtjddSYnX|jr_tjdP|dkrytjdP|\}}}y||j||Wntk rYnXd}}}qWx|r|jt kry |}Wn)ttfk rtjddSYnX|dkr4tjdq|\}}}y||j||Wntk roYnXd}}}qWt |drtjdy2x+t dD]}|j j sP|qWWnttfk rYnXtjdt||jdS) Nz.result handler got EOFError/OSError -- exitingz,result handler found thread._state=TERMINATEzresult handler got sentinelz&result handler ignoring extra sentinelr/z"ensuring that outqueue is not full z7result handler exiting: len(cache)=%s, thread._state=%s)r[rr1r0rr2rOrrrr-rpr/pollrq)r7r,rrr:r;r<objr r r resZ                 zPool._handle_resultsccsDt|}x1ttj||}|s1dS||fVqWdS)N)rtuplerislice)r=itsizerr r r rs  zPool._get_taskscCstddS)Nz:pool objects cannot be passed between processes or pickled)NotImplementedError)rr r r r"szPool.__reduce__cCs5tjd|jtkr1t|_t|j_dS)Nz closing pool)rr2rOrNCLOSEr^)rr r r r.s  z Pool.closecCs0tjdt|_t|j_|jdS)Nzterminating pool)rr2rrOr^rk)rr r r terminates   zPool.terminatecCsVtjd|jj|jj|jjx|jD]}|jq>WdS)Nz joining pool)rr2r^rrdrgrY)rrr r r rs     z Pool.joincCsZtjd|jjx9|jrU|jjrU|jjtj dqWdS)Nz7removing tasks from inqueue until task handler finishedr) rr2Z_rlockacquireis_aliver/rr|rr)r6 task_handlerrr r r _help_stuff_finish(s    zPool._help_stuff_finishc Cstjdt|_t|_tjd|j||t|t|_|jdtjdtj|k r|j |rt |ddrtjdx'|D]} | j dkr| j qWtjdtj|k r|j tjdtj|k r&|j |rt |ddrtjd x8|D]0} | j rStjd | j| j qSWdS) Nzfinalizing poolz&helping task handler/workers to finishzjoining worker handlerrrzterminating workerszjoining task handlerzjoining result handlerzjoining pool workerszcleaning up worker %d)rr2rrOrrqr+r[rrr-rrrrpid) clsrr6r7rZworker_handlerrZresult_handlerrrr r r ri1s6                 zPool._terminate_poolcCs|S)Nr )rr r r __enter___szPool.__enter__cCs|jdS)N)r)rexc_typeexc_valexc_tbr r r __exit__bsz Pool.__exit__)"rrrrtrErrsrZryrIr~r rrrrrr}rr staticmethodr]rarerr"r.rrr classmethodrirrr r r r rsD   8        .<     . c@s^eZdZddZddZddZddd Zdd d Zd d ZdS)rcCsJtj|_tt|_||_||_||_|||jrEr r r rE szThreadPool.ProcessNcCstj||||dS)N)rr)rrlr8r9r r r rszThreadPool.__init__cCs@tj|_tj|_|jj|_|jj|_dS)N)rJrKrjrcr+rbr,rf)rr r r rIszThreadPool._setup_queuesc CsF|j6|jj|jjdg||jjWdQRXdS)N) not_emptyrJclearextend notify_all)r6rrr r r rs  zThreadPool._help_stuff_finish) rrrrtrrErrIrr r r r rs  )"__all__r[rJrrrSrrrrrrrNrrcountrrrr3rrr!r$rBr4objectrrZ AsyncResultrrrrr r r r  s<             * &%@