3
\e @ s8 d dgZ ddlZddlZddlZddlZddlZddlZddlZddlm Z ddlm
Z
m Z dZ dZ
dZej Zdd Zd
d ZG d d
d
eZG dd dZdd ZG dd deZdf ddfddZdd ZG dd d eZG dd deZ e Z G d d d e Z G d d d eZ G d d! d!e Z G d"d deZ!dS )#Pool
ThreadPool N )util) get_context TimeoutError c C s t t| S )N)listmap)args r ,/usr/lib64/python3.6/multiprocessing/pool.pymapstar+ s r c C s t tj| d | d S )Nr r )r itertoolsstarmap)r r r r
starmapstar. s r c @ s e Zd Zdd Zdd ZdS )RemoteTracebackc C s
|| _ d S )N)tb)selfr r r r
__init__6 s zRemoteTraceback.__init__c C s | j S )N)r )r r r r
__str__8 s zRemoteTraceback.__str__N)__name__
__module__ __qualname__r r r r r r
r 5 s r c @ s e Zd Zdd Zdd ZdS )ExceptionWithTracebackc C s0 t jt|||}dj|}|| _d| | _d S )N z
"""
%s""") tracebackformat_exceptiontypejoinexcr )r r r r r r
r < s
z ExceptionWithTraceback.__init__c C s t | j| jffS )N) rebuild_excr r )r r r r
__reduce__A s z!ExceptionWithTraceback.__reduce__N)r r r r r" r r r r
r ; s r c C s t || _| S )N)r __cause__)r r r r r
r! D s
r! c s0 e Zd ZdZ fddZdd Zdd Z ZS )MaybeEncodingErrorzVWraps possible unpickleable errors, so they can be
safely sent through the socket.c s. t || _t || _tt| j| j| j d S )N)reprr valuesuperr$ r )r r r& ) __class__r r
r P s
zMaybeEncodingError.__init__c C s d| j | jf S )Nz(Error sending result: '%s'. Reason: '%s')r& r )r r r r
r U s zMaybeEncodingError.__str__c C s d| j j| f S )Nz<%s: %s>)r( r )r r r r
__repr__Y s zMaybeEncodingError.__repr__)r r r __doc__r r r)
__classcell__r r )r( r
r$ L s r$ Fc ' C s |j }| j}t| dr*| jj |jj |d k r:|| d}xH|d ksX|oT||k ry
| } W n$ ttfk
r tj d P Y nX | d krtj d P | \}
} } }
}yd| |
|f}W nB t
k
r } z&|r| t k rt ||j
}d|f}W Y d d }~X nX y||
| |f W nR t
k
rd } z4t||d }tj d| ||
| d|ff W Y d d }~X nX d } }
} } }
}|d7 }qBW tj d | d S )
N_writerr z)worker got EOFError or OSError -- exitingz worker got sentinel -- exitingTFr z0Possible encoding error while sending result: %sz worker exiting after %d tasks)putgethasattrr, close_readerEOFErrorOSErrorr debug Exception_helper_reraises_exceptionr
__traceback__r$ )inqueueoutqueue initializerinitargsZmaxtasksZwrap_exceptionr- r. Z completedtaskjobifuncr kwdsresultewrappedr r r
worker] sD
$ rD c C s | dS )z@Pickle-able helper function for use by _guarded_task_generation.Nr )Zexr r r
r6 s r6 c @ s6 e Zd ZdZdZdd Zddf ddfddZdd Zd
d Zd d
Z dd Z
f i fddZ d=ddZ d>ddZ
d?ddZdd Zd@dd ZdAd d Zf i ddfd d ZdBd!d"ZdCd#d$Zed%d& Zed'd( Zed)d* Zed+d, Zd-d. Zd/d0 Zd1d2 Z d3d4 Z ed5d6 Z e d7d8 Z d9d: Z!d;d< Z"dS )Dr zS
Class which supports an async version of applying functions to arguments.
Tc O s | j j||S )N)_ctxProcess)r r r@ r r r
rF s z Pool.ProcessNc C sn |pt | _| j tj | _i | _t| _|| _ || _
|| _ |d krPt j
pNd}|dk r`td|d k rzt| rztd|| _g | _| j tjtj| fd| _d| j_t| j_| jj tjtj| j| j | j | j| jfd| _ d| j _t| j _| j j tjtj | j | j | jfd| _!d| j!_t| j!_| j!j t"j#| | j$| j| j%| j | j| j| j | j!| jfdd| _&d S )Nr z&Number of processes must be at least 1z initializer must be a callable)targetr T )r Z exitpriority)'r rE
_setup_queuesqueueQueue
_taskqueue_cacheRUN_state_maxtasksperchild _initializer _initargsos cpu_count
ValueErrorcallable TypeError
_processes_pool_repopulate_pool threadingZThreadr _handle_workers_worker_handlerdaemonstart
_handle_tasks
_quick_put _outqueue
_task_handler_handle_results
_quick_get_result_handlerr ZFinalize_terminate_pool_inqueue
_terminate)r processesr: r; Zmaxtasksperchildcontextr r r
r sT
z
Pool.__init__c C sZ d}xPt tt| jD ]:}| j| }|jdk rtjd| |j 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)reversedrangelenrY exitcoder r4 r )r Zcleanedr> rD r r r
_join_exited_workers s
zPool._join_exited_workersc C s| xvt | jt| j D ]^}| jt| j| j| j| j | j
| j fd}| jj | |j
jdd|_
d|_|j tjd qW dS )zBring the number of pool processes up to the specified number,
for use after reaping workers which have exited.
)rG r rF Z
PoolWorkerTz added workerN)rm rX rn rY rF rD rh rb rQ rR rP _wrap_exceptionappendnamereplacer^ r_ r r4 )r r> wr r r
rZ s zPool._repopulate_poolc C s | j r| j dS )zEClean up any exited workers and start replacements for them.
N)rp rZ )r r r r
_maintain_pool s zPool._maintain_poolc C s4 | j j | _| j j | _| jjj| _| jjj| _ d S )N)
rE Z SimpleQueuerh rb r, sendra r1 recvre )r r r r
rI s zPool._setup_queuesc C s | j |||j S )z6
Equivalent of `func(*args, **kwds)`.
) apply_asyncr. )r r? r r@ r r r
apply s z
Pool.applyc C s | j ||t|j S )zx
Apply `func` to each element in `iterable`, collecting the results
in a list that is returned.
)
_map_asyncr r. )r r? iterable chunksizer r r
r
s zPool.mapc C s | j ||t|j 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{ r r. )r r? r| r} r r r
r s z Pool.starmapc C s | j ||t|||S )z=
Asynchronous version of `starmap()` method.
)r{ r )r r? r| r} callbackerror_callbackr r r
starmap_async s zPool.starmap_asyncc c sn y0d}x&t |D ]\}}||||fi fV qW W n8 tk
rh } z ||d t|fi fV W Y dd}~X nX dS )zProvides a generator of tasks for imap and imap_unordered with
appropriate handling for iterables which throw exceptions during
iteration.r N) enumerater5 r6 )r Z
result_jobr? r| r> xrB r r r
_guarded_task_generation s z Pool._guarded_task_generationr c C s | j tkrtd|dkrFt| j}| jj| j|j|||j f |S t
j |||}t| j}| jj| j|jt ||j f dd |D S dS )zP
Equivalent of `map()` -- can be MUCH slower than `Pool.map()`.
zPool not runningr c s s | ]}|D ]
}|V q
qd S )Nr ).0chunkitemr r r
@ s z Pool.imap..N)
rO rN rU IMapIteratorrM rL r- r _job _set_lengthr
_get_tasksr )r r? r| r} rA task_batchesr r r
imap' s
z Pool.imapc C s | j tkrtd|dkrFt| j}| jj| j|j|||j f |S t
j |||}t| j}| jj| j|jt ||j f dd |D S dS )zL
Like `imap()` method but ordering of results is arbitrary.
zPool not runningr c s s | ]}|D ]
}|V q
qd S )Nr )r r r r r r
r [ s z&Pool.imap_unordered..N)
rO rN rU IMapUnorderedIteratorrM rL r- r r r r r r )r r? r| r} rA r r r r
imap_unorderedB s
zPool.imap_unorderedc C sB | j tkrtdt| j||}| jj|jd|||fgdf |S )z;
Asynchronous version of `apply()` method.
zPool not runningr N)rO rN rU ApplyResultrM rL r- r )r r? r r@ r~ r rA r r r
ry ] s
zPool.apply_asyncc C s | j ||t|||S )z9
Asynchronous version of `map()` method.
)r{ r )r r? r| r} r~ r r r r
map_asynch s zPool.map_asyncc
C s | j tkrtdt|ds$t|}|dkrTtt|t| jd \}}|rT|d7 }t|dkrdd}tj |||}t
| j |t|||d} | j j
| j| j||df | S )zY
Helper function to implement map, starmap and their async counterparts.
zPool not running__len__N r r )r )rO rN rU r/ r divmodrn rY r r MapResultrM rL r- r r )
r r? r| Zmapperr} r~ r Zextrar rA r r r
r{ p s&
zPool._map_asyncc C sT t j }x0|jtks$| jr8|jtkr8| j tjd q
W | j j
d t j d d S )Ng?zworker handler exiting)
r[ current_threadrO rN rM TERMINATErv timesleeprL r- r r4 )poolthreadr r r
r\ s zPool._handle_workersc
C sj t j }xt| jd D ]\}}d }zx|D ]}|jr@tjd P y || W q* tk
r } zD|d d \}
} y||
j| d| f W n t k
r Y nX W Y d d } ~ X q*X q*W |rtjd |r|d nd } || d wP W d d } }}
X qW tjd y:tjd |j
d tjd x|D ]} |d q&W W n t k
rZ tjd Y nX tjd
d S ) Nz'task handler found thread._state != RUNr Fzdoing set_length()r 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 exitingr ) r[ r iterr. rO r r4 r5 _setKeyErrorr- r3 )
taskqueuer- r9 r cacher ZtaskseqZ
set_lengthr< rB r= idxpr r r
r` sB
zPool._handle_tasksc C s t j }xy
| }W n" ttfk
r6 tjd d S X |jrJtjd P |d kr^tjd P |\}}}y|| j|| W n tk
r Y nX d } }}q
W x|o|jt kr:y
| }W n" ttfk
r tjd d S X |d krtjd q|\}}}y|| j|| W n tk
r* Y nX d } }}qW t
| drtjd y,x&t dD ]}| j j
spP | q^W W n ttfk
r Y nX tjdt||j d S ) Nz.result handler got EOFError/OSError -- exitingz,result handler found thread._state=TERMINATEzresult handler got sentinelz&result handler ignoring extra sentinelr1 z"ensuring that outqueue is not full
z7result handler exiting: len(cache)=%s, thread._state=%s)r[ r r3 r2 r r4 rO r r r r/ rm r1 pollrn )r9 r. r r r< r= r> objr r r
rd sZ
zPool._handle_resultsc c s4 t |}x&ttj||}|s"d S | |fV q
W d S )N)r tupler islice)r? itsizer r r r
r s zPool._get_tasksc C s t dd S )Nz:pool objects cannot be passed between processes or pickled)NotImplementedError)r r r r
r" s zPool.__reduce__c C s& t jd | jtkr"t| _t| j_d S )Nz closing pool)r r4 rO rN CLOSEr] )r r r r
r0 s
z
Pool.closec C s$ t jd t| _t| j_| j d S )Nzterminating pool)r r4 r rO r] ri )r r r r
terminate s
zPool.terminatec C sD t jd | jj | jj | jj x| jD ] }|j q0W d S )Nz joining pool)r r4 r] r rc rf rY )r r r r r
r s
z Pool.joinc C sD t jd | jj x*|j r>| jj r>| jj tj d qW d S )Nz7removing tasks from inqueue until task handler finishedr )
r r4 Z_rlockacquireis_aliver1 r rx r r )r8 task_handlerr r r r
_help_stuff_finish( s
zPool._help_stuff_finishc
C s( t jd t|_t|_t jd | j||t| t|_|jd t jd tj |k r`|j |rt
|d drt jd x |D ]} | j d kr| j qW t jd tj |k r|j t jd tj |k r|j |ot
|d dr$t jd x,|D ]$} | j
rt jd
| j | j qW d S ) Nzfinalizing poolz&helping task handler/workers to finishzjoining worker handlerr r zterminating workerszjoining task handlerzjoining result handlerzjoining pool workerszcleaning up worker %d)r r4 r rO r rn r- r[ r r r/ ro r r pid)
clsr r8 r9 r Zworker_handlerr Zresult_handlerr r r r r
rg 1 s6
zPool._terminate_poolc C s | S )Nr )r r r r
__enter___ s zPool.__enter__c C s | j d S )N)r )r exc_typeZexc_valZexc_tbr r r
__exit__b s z
Pool.__exit__)N)N)NNN)r )r )NNN)NNN)#r r r r* rq rF r rp rZ rv rI rz r
r r r r r ry r r{ staticmethodr\ r` rd r r" r0 r r r classmethodrg r r r r r r
r sF 8
. < .c @ s@ e Zd Zdd Zdd Zdd Zddd Zdd
d Zd d
ZdS )r c C s4 t j | _tt| _|| _|| _|| _| || j< d S )N) r[ ZEvent_eventnext job_counterr rM _callback_error_callback)r r r~ r r r r
r k s
zApplyResult.__init__c C s
| j j S )N)r Zis_set)r r r r
readys s zApplyResult.readyc C s | j S )N)_success)r r r r
successfulv s zApplyResult.successfulNc C s | j j| d S )N)r wait)r timeoutr r r
r z s zApplyResult.waitc C s, | j | | j st| jr"| jS | jd S )N)r r r r _value)r r r r r
r. } s
zApplyResult.getc C sV |\| _ | _| jr$| j r$| j| j | jr>| j r>| j| j | jj | j| j= d S )N)r r r r r setrM r )r r> r r r r
r s
zApplyResult._set)N)N) r r r r r r r r. r r r r r
r i s
r c @ s e Zd Zdd Zdd ZdS )r c C sf t j| |||d d| _d g| | _|| _|dkrLd| _| jj || j= n|| t || | _d S )N)r Tr )
r r r r
_chunksize _number_leftr r r bool)r r r} lengthr~ r r r r
r s
zMapResult.__init__c C s | j d8 _ |\}}|rp| jrp|| j|| j |d | j < | j dkr| jrZ| j| j | j| j= | jj nH| r| jrd| _|| _| j dkr| j r| j | j | j| j= | jj d S )Nr r F)
r r r r r rM r r r r )r r> Zsuccess_resultsuccessrA r r r
r s"
zMapResult._setN)r r r r r r r r r
r s
r c @ s: e Zd Zdd Zdd Zd ddZeZdd Zd
d ZdS )
r c C sJ t jt j | _tt| _|| _tj | _
d| _ d | _ i | _
| || j< d S )Nr )r[ Z ConditionZLock_condr r r rM collectionsdeque_items_index_length _unsorted)r r r r r
r s
zIMapIterator.__init__c C s | S )Nr )r r r r
__iter__ s zIMapIterator.__iter__Nc C s | j ~ y| jj }W nh tk
r~ | j| jkr6t| j j| y| jj }W n( tk
rx | j| jkrpttY nX Y nX W d Q R X |\}}|r|S |d S )N) r r popleft
IndexErrorr r
StopIterationr r )r r r r r& r r r
r s" zIMapIterator.nextc
C s | j | j|krr| jj| | jd7 _x8| j| jkrd| jj| j}| jj| | jd7 _q.W | j j n
|| j|< | j| jkr| j| j = W d Q R X d S )Nr )
r r r rr r popnotifyr rM r )r r> r r r r
r s
zIMapIterator._setc C s< | j , || _| j| jkr.| j j | j| j= W d Q R X d S )N)r r r r rM r )r r r r r
r s
zIMapIterator._set_length)N) r r r r r r __next__r r r r r r
r s
r c @ s e Zd Zdd ZdS )r c
C sP | j @ | jj| | jd7 _| j j | j| jkrB| j| j= W d Q R X d S )Nr )r r rr r r r rM r )r r> r r r r
r s
zIMapUnorderedIterator._setN)r r r r r r r r
r s r c @ s@ e Zd ZdZedd Zddf fddZdd Zed d
ZdS ) r Fc O s ddl m} || |S )Nr )rF )ZdummyrF )r r@ rF r r r
rF s zThreadPool.ProcessNc C s t j| ||| d S )N)r r )r rj r: r; r r r
r s zThreadPool.__init__c C s, t j | _t j | _| jj| _| jj| _d S )N)rJ rK rh rb r- ra r. re )r r r r
rI s
zThreadPool._setup_queuesc
C s< | j , | jj | jjd g| | j j W d Q R X d S )N)Z not_emptyrJ clearextendZ
notify_all)r8 r r r r r
r s
z ThreadPool._help_stuff_finish) r r r rq r rF r rI r r r r r
r s
)"__all__r[ rJ r r rS r r r r r r rN r r countr r r r5 r r r! r$ rD r6 objectr r Z AsyncResultr r r r r r r r
s@ * \&)@