U
e5d~ @ sd d dgZ ddlZddlZddlZddlZddlZddlZddlZddlZddlm Z ddl
m Z ddl
m Z m
Z
ddlmZ d Zd
Zd Zd Ze Zd
d Zdd ZG dd deZG dd dZdd ZG dd deZ d+dd Z d d Z G d d d e Z G d!d d e!Z"G d"d# d#e!Z#e#Z$G d$d% d%e#Z%G d&d' d'e!Z&G d(d) d)e&Z'G d*d de"Z(dS ),Pool
ThreadPool N)Empty )util) get_context TimeoutError)waitINITRUNCLOSE TERMINATEc C s t t| S N)listmapargs r ,/usr/lib64/python3.8/multiprocessing/pool.pymapstar/ s r c C s t t| d | d S )Nr r )r itertoolsstarmapr r r r starmapstar2 s r c @ s e Zd Zdd Zdd ZdS )RemoteTracebackc C s
|| _ d S r tb)selfr r r r __init__: s zRemoteTraceback.__init__c C s | j S r r r r r r __str__< s zRemoteTraceback.__str__N)__name__
__module__ __qualname__r r r r r r r 9 s r c @ s e Zd Zdd Zdd ZdS )ExceptionWithTracebackc C s0 t t|||}d|}|| _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 r ) rebuild_excr) r r r r r
__reduce__E s z!ExceptionWithTraceback.__reduce__N)r r! r" r r+ r r r r r# ? s r# c C s t || _| S r )r __cause__)r) r r r r r* H 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 d S r )reprr) valuesuperr- r )r r) r/ __class__r r r T s
zMaybeEncodingError.__init__c C s d| j | jf S )Nz(Error sending result: '%s'. Reason: '%s')r/ r) r r r r r Y s zMaybeEncodingError.__str__c C s d| j j| f S )Nz<%s: %s>)r2 r r r r r __repr__] s zMaybeEncodingError.__repr__)r r! r" __doc__r r r3
__classcell__r r r1 r r- P s r- r Fc
C s |d k r(t |tr|dks(td||j}| j}t| drR| j |j |d k rb|| d}|d ks~|r||k rz
| } W n( t
t fk
r t
d Y qY nX | d krt
d q| \}
} } }
}zd| |
|f}W nH tk
r0 } z(|r| tk rt||j}d|f}W 5 d }~X Y nX z||
| |f W nR tk
r } z2t||d }t
d | ||
| d|ff W 5 d }~X Y nX d } }
} } }
}|d7 }qft
d
| d S ) Nr zMaxtasks {!r} is not valid_writerr z)worker got EOFError or OSError -- exitingz worker got sentinel -- exitingTFz0Possible encoding error while sending result: %sz worker exiting after %d tasks)
isinstanceintAssertionErrorformatputgethasattrr6 close_readerEOFErrorOSErrorr debug Exception_helper_reraises_exceptionr#
__traceback__r- )inqueueoutqueue initializerinitargsZmaxtaskswrap_exceptionr; r< Z completedtaskjobifuncr kwdsresultewrappedr r r workera sN
$
rS c C s | dS )z@Pickle-able helper function for use by _guarded_task_generation.Nr )Zexr r r rD s rD c s2 e Zd ZdZdd fdd
Z fddZ ZS )
_PoolCachez
Class that implements a cache for the Pool class that will notify
the pool management threads every time the cache is emptied. The
notification is done by the use of a queue that is provided when
instantiating the cache.
Nnotifierc s || _ t j|| d S r )rV r0 r )r rV r rO r1 r r r s z_PoolCache.__init__c s t | | s | jd d S r )r0 __delitem__rV r; )r itemr1 r r rW s z_PoolCache.__delitem__)r r! r" r4 r rW r5 r r r1 r rT s rT c @ s e Zd ZdZdZedd ZdLddZej e
fd d
Z d d Z d
d Z
edd Zedd Zdd Zedd Zedd Zdd Zdd Zdi fd d ZdMd d ZdNd!d"ZdOd#d$Zd%d& ZdPd(d)ZdQd*d+Zdi ddfd,d-Z dRd.d/Z dSd0d1Z edTd2d3Z e d4d5 Z!ed6d7 Z"ed8d9 Z#ed:d; Z$dd? Z&d@dA Z'dBdC Z(edDdE Z)e dFdG Z*dHdI Z+dJdK Z,dS )Ur zS
Class which supports an async version of applying functions to arguments.
Tc O s | j ||S r Process)ctxr rO r r r rZ s z Pool.ProcessNr c C s g | _ t| _|pt | _| t | _| j | _ t
| j d| _ || _ || _
|| _|d krjt phd}|dk rztd|d k rt|std|| _z | W nH tk
r | j D ]}|jd kr| q| j D ] }| q Y nX | }tj t j | j | j| j| j | j| j | j | j!| j
| j| j | j"|| j fd| _#d| j#_$t%| j#_| j#& tj t j'| j| j(| j!| j | j fd| _)d| j)_$t%| j)_| j)& tj t j*| j!| j+| j fd| _,d| j,_$t%| j,_| j,& t-j.| | j/| j| j | j!| j | j | j#| j)| j,| j f dd| _0t%| _d S ) NrU r z&Number of processes must be at least 1z initializer must be a callabletargetr T )r Z exitpriority)1_poolr
_stater _ctx
_setup_queuesqueue SimpleQueue
_taskqueue_change_notifierrT _cache_maxtasksperchild _initializer _initargsos cpu_count
ValueErrorcallable TypeError
_processes_repopulate_poolrC exitcode terminater( _get_sentinels threadingZThreadr _handle_workersrZ _inqueue _outqueue_wrap_exception_worker_handlerdaemonr start
_handle_tasks
_quick_put
_task_handler_handle_results
_quick_get_result_handlerr ZFinalize_terminate_pool
_terminate)r processesrH rI maxtasksperchildcontextp sentinelsr r r r s
z
Pool.__init__c C s> | j |kr:|d| t| d t| dd d k r:| jd d S )Nz&unclosed running multiprocessing pool )sourcerf )r` ResourceWarninggetattrrf r; )r Z_warnr r r r __del__ s
z Pool.__del__c C s0 | j }d|j d|j d| j dt| j d S )N<.z state=z pool_size=>)r2 r! r" r` lenr_ )r clsr r r r3 s z
Pool.__repr__c C s | j jg}| jjg}||S r )rx r? rf )r Ztask_queue_sentinelsZself_notifier_sentinelsr r r rt s
zPool._get_sentinelsc C s dd | D S )Nc S s g | ]}t |d r|jqS )sentinel)r= r ).0rS r r r
s
z.Pool._get_worker_sentinels..r Zworkersr r r _get_worker_sentinels s zPool._get_worker_sentinelsc C sP d}t tt| D ]6}| | }|jdk rtd| | d}| |= q|S )zCleanup after any worker processes which have exited due to reaching
their specified lifetime. Returns True if any workers were cleaned up.
FNcleaning up worker %dT)reversedranger rr r rB r( )poolZcleanedrM rS r r r _join_exited_workers s
zPool._join_exited_workersc C s0 | | j| j| j| j| j| j| j| j| j | j
S r ) _repopulate_pool_staticra rZ rp r_ rw rx ri rj rh ry r r r r rq . s zPool._repopulate_poolc
C sf t |t| D ]P}
|| t|||||| fd} | jdd| _d| _| || t d qdS )zBring the number of pool processes up to the specified number,
for use after reaping workers which have exited.
r\ rZ Z
PoolWorkerTz added workerN)
r r rS namereplacer{ r| appendr rB ) r[ rZ r r rF rG rH rI r rJ rM wr r r r 7 s
z Pool._repopulate_pool_staticc
C s* t |r&t | |||||||||
dS )zEClean up any exited workers and start replacements for them.
N)r r r )
r[ rZ r r rF rG rH rI r rJ r r r _maintain_poolJ s
zPool._maintain_poolc C s4 | j | _| j | _| jjj| _| jjj| _ d S r )
ra rd rw rx r6 sendr~ r? recvr r r r r rb V s zPool._setup_queuesc C s | j tkrtdd S )NzPool not running)r` r rm r r r r _check_running\ s
zPool._check_runningc C s | ||| S )zT
Equivalent of `func(*args, **kwds)`.
Pool must be running.
) apply_asyncr< )r rN r rO r r r apply` s z
Pool.applyc C s | ||t| S )zx
Apply `func` to each element in `iterable`, collecting the results
in a list that is returned.
)
_map_asyncr r< r rN iterable chunksizer r r r g s zPool.mapc C s | ||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 r r< r r r r r n s z Pool.starmapc C s | ||t|||S )z=
Asynchronous version of `starmap()` method.
)r r r rN r r callbackerror_callbackr r r
starmap_asyncv s zPool.starmap_asyncc
c sj z,d}t |D ]\}}||||fi fV qW n8 tk
rd } z||d t|fi fV W 5 d}~X Y nX dS )zProvides a generator of tasks for imap and imap_unordered with
appropriate handling for iterables which throw exceptions during
iteration.r N) enumeraterC rD )r Z
result_jobrN r rM xrQ r r r _guarded_task_generation~ s z Pool._guarded_task_generationr c C s | |dkr:t| }| j| |j|||jf |S |dk rPtd|t
|||}t| }| j| |jt ||jf dd |D S dS )zP
Equivalent of `map()` -- can be MUCH slower than `Pool.map()`.
r z Chunksize must be 1+, not {0:n}c s s | ]}|D ]
}|V q
qd S r r r chunkrX r r r s z Pool.imap..N) r IMapIteratorre r; r _job _set_lengthrm r: r
_get_tasksr r rN r r rP task_batchesr r r imap s4 z Pool.imapc C s | |dkr:t| }| j| |j|||jf |S |dk rPtd|t
|||}t| }| j| |jt ||jf dd |D S dS )zL
Like `imap()` method but ordering of results is arbitrary.
r z Chunksize must be 1+, not {0!r}c s s | ]}|D ]
}|V q
qd S r r r r r r r s z&Pool.imap_unordered..N) r IMapUnorderedIteratorre r; r r r rm r: r r r r r r r imap_unordered s0 zPool.imap_unorderedc C s6 | t| ||}| j|jd|||fgdf |S )z;
Asynchronous version of `apply()` method.
r N)r ApplyResultre r; r )r rN r rO r r rP r r r r s zPool.apply_asyncc C s | ||t|||S )z9
Asynchronous version of `map()` method.
)r r r r r r map_async s zPool.map_asyncc
C s | t|dst|}|dkrJtt|t| jd \}}|rJ|d7 }t|dkrZd}t|||}t| |t|||d} | j
| | j ||df | S )zY
Helper function to implement map, starmap and their async counterparts.
__len__N r r r )
r r= r divmodr r_ r r MapResultre r; r r )
r rN r Zmapperr r r Zextrar rP r r r r s,
zPool._map_asyncc C s" t | |d | s | q d S )N)timeout)r emptyr< )r change_notifierr r r r _wait_for_updates s zPool._wait_for_updatesc C sp t }|jtks |rX|jtkrX| ||||||| |
| |
| ||
}| || q|d t
d d S )Nzworker handler exiting) ru current_threadr` r r
r r r r; r rB )r cache taskqueuer[ rZ r r rF rG rH rI r rJ r r threadZcurrent_sentinelsr r r rv s
zPool._handle_workersc
C sp t }t| jd D ]\}}d }z|D ]}|jtkrBtd qz || W q& tk
r }
zB|d d \} } z|| | d|
f W n t
k
r Y nX W 5 d }
~
X Y q&X q&|rtd |r|d nd} || d W qW
q
W 5 d } }} X qtd z6td | d td |D ]} |d q.W n t k
r` td
Y nX td d S ) Nz'task handler found thread._state != RUN Fzdoing set_length()r 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 exiting)
ru r iterr< r` r r rB rC _setKeyErrorr; rA )
r r; rG r r r ZtaskseqZ
set_lengthrK rL rQ idxr r r r r} sB
zPool._handle_tasksc C s t }z
| }W n$ ttfk
r6 td Y d S X |jtkrNtd q|d krbtd q|\}}}z|| || W n t k
r Y nX d } }}q|r@|jt
kr@z
| }W n$ ttfk
r td Y d S X |d k rtd q|\}}}z|| || W n t k
r0 Y nX d } }}qt | drtd z,t dD ] }| j
sv q| q`W n ttfk
r Y nX tdt||j d S ) 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)ru r rA r@ r rB r` r r r r
r= r r? pollr )rG r< r r rK rL rM objr r r r : s\
zPool._handle_resultsc c s0 t |}tt||}|s d S | |fV qd S r )r tupler islice)rN itsizer r r r r v s
zPool._get_tasksc C s t dd S )Nz:pool objects cannot be passed between processes or pickled)NotImplementedErrorr r r r r+ s zPool.__reduce__c C s2 t d | jtkr.t| _t| j_| jd d S )Nz closing pool)r rB r` r r rz rf r; r r r r r> s
z
Pool.closec C s t d t| _| d S )Nzterminating pool)r rB r
r` r r r r r rs s
zPool.terminatec C sj t d | jtkr tdn| jttfkr4td| j | j | j
| j D ] }| qXd S )Nz joining poolzPool is still runningzIn unknown state) r rB r` r rm r r
rz r( r r r_ )r r r r r r( s
z Pool.joinc C s@ t d | j | r<| j r<| j t d qd S )Nz7removing tasks from inqueue until task handler finishedr )
r rB Z_rlockacquireis_aliver? r r timesleep)rF task_handlerr r r r _help_stuff_finish s
zPool._help_stuff_finishc
C sX t d t|_|d t|_t d | ||t| | sXt| dkrXtdt|_|d |d t d t
|k r| |rt |d drt d |D ]}
|
j
d kr|
qt d t
|k r| t d t
|k r| |rTt |d drTt d
|D ](}
|
r*t d |
j |
q*d S ) Nzfinalizing poolz&helping task handler/workers to finishr z.Cannot have cache with result_hander not alivezjoining worker handlerrs zterminating workerszjoining task handlerzjoining result handlerzjoining pool workersr )r rB r
r` r; r r r r9 ru r r( r= rr rs pid) r r rF rG r r Zworker_handlerr Zresult_handlerr r r r r r sB
zPool._terminate_poolc C s | | S r )r r r r r __enter__ s zPool.__enter__c C s | d S r )rs )r exc_typeZexc_valZexc_tbr r r __exit__ s z
Pool.__exit__)NNr NN)N)N)NNN)r )r )NNN)NNN)N)-r r! r" r4 ry staticmethodrZ r warningswarnr r r3 rt r r rq r r rb r r r r r r r r r r r r classmethodrv r} r r r+ r> rs r( r r r r r r r r r sx
P
-
;
5c @ s@ e Zd Zdd Zdd Zdd Zddd Zdd
d Zd d
ZdS )r c C s> || _ t | _tt| _|j| _|| _|| _ | | j| j< d S r )
r_ ru ZEvent_eventnext job_counterr rg _callback_error_callback)r r r r r r r r s
zApplyResult.__init__c C s
| j S r )r Zis_setr r r r ready s zApplyResult.readyc C s | std| | jS )Nz{0!r} not ready)r rm r: _successr r r r
successful s zApplyResult.successfulNc C s | j | d S r )r r r r r r r r s zApplyResult.waitc C s, | | | st| jr"| jS | jd S r )r r r r _valuer r r r r< s
zApplyResult.getc C sZ |\| _ | _| jr$| j r$| | j | jr<| j s<| | j | j | j| j= d | _d S r ) r r r r r setrg r r_ r rM r r r r r s
zApplyResult._set)N)N) r r! r" r r r r r< r r r r r r s
r c @ s e Zd Zdd Zdd ZdS )r c C sh t j| |||d d| _d g| | _|| _|dkrNd| _| j | j| j = n|| t
|| | _d S )Nr Tr ) r r r r
_chunksize _number_leftr r rg r bool)r r r lengthr r r r r r s
zMapResult.__init__c C s | j d8 _ |\}}|rv| jrv|| j|| j |d | j < | j dkr| jrZ| | j | j| j= | j d | _ nL|s| jrd| _|| _| j dkr| j
r|
| j | j| j= | j d | _ d S )Nr r F) r r r r r rg r r r r_ r )r rM Zsuccess_resultsuccessrP 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 sT || _ tt | _tt| _|j| _t
| _ d| _ d | _
i | _| | j| j< d S )Nr )r_ ru Z ConditionZLock_condr r r rg collectionsdeque_items_index_length _unsorted)r r r r r r B s
zIMapIterator.__init__c C s | S r r r r r r __iter__M s zIMapIterator.__iter__Nc C s | j z| j }W nz tk
r | j| jkr>d | _td | j | z| j }W n2 tk
r | j| jkrd | _td t d Y nX Y nX W 5 Q R X |\}}|r|S |d S r )
r r popleft
IndexErrorr r r_
StopIterationr r )r r rX r r/ r r r r P s& zIMapIterator.nextc C s | j | j|krn| j| | jd7 _| j| jkrb| j| j}| j| | jd7 _q,| j n
|| j|< | j| jkr| j| j = d | _
W 5 Q R X d S Nr ) r r r r r popnotifyr rg r r_ r r r r r h s
zIMapIterator._setc C sB | j 2 || _| j| jkr4| j | j| j= d | _W 5 Q R X d S r )r r r r rg r r_ )r r r r r r y 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 sV | j F | j| | jd7 _| j | j| jkrH| j| j= d | _W 5 Q R X d S r ) r r r r r r rg r r_ r r r r r s
zIMapUnorderedIterator._setN)r r! r" r r r r r r s r c @ sV e Zd ZdZedd ZdddZdd Zd
d Zed d
Z edd Z
dd Z dS )r Fc O s ddl m} |||S )Nr rY )ZdummyrZ )r[ r rO rZ r r r rZ s zThreadPool.ProcessNr c C s t | ||| d S r )r r )r r rH rI r r r r s zThreadPool.__init__c C s, t | _t | _| jj| _| jj| _d S r )rc rd rw rx r; r~ r< r r r r r rb s
zThreadPool._setup_queuesc C s
| j jgS r )rf r? r r r r rt s zThreadPool._get_sentinelsc C s g S r r r r r r r s z ThreadPool._get_worker_sentinelsc C sF z| j dd qW n tjk
r( Y nX t|D ]}| d q2d S )NF)block)r< rc r r r; )rF r r rM r r r r s z ThreadPool._help_stuff_finishc C s t | d S r )r r )r r r r r r r r s z ThreadPool._wait_for_updates)NNr ) r r! r" ry r rZ r rb rt r r r r r r r r s
)Nr NF))__all__r r rk rc ru r r% r r r$ r r r Z
connectionr r
r r r
countr r r rC r r# r* r- rS rD dictrT objectr r Z AsyncResultr r r r r r r r
sN
- =)+E