֦i d dgZ ddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl
m Z ddl
m Z m
Z
ddlmZ dZd Zd
Zd Z ej( Zd Zd
Z G d de Z G d d Zd Z G d de Z d#dZ d Z G d de Z G d d e! Z" G d de! Z#e#Z$ G d d e# Z% G d d e! Z& G d d!e& Z' G d" de" Z(y)$Pool
ThreadPool N )util) get_context TimeoutError)waitINITRUNCLOSE TERMINATEc $ t t | S N)listmapargss -/usr/lib64/python3.12/multiprocessing/pool.pymapstarr / s T
c L t t j | d | d S )Nr r )r itertoolsstarmapr s r starmapstarr 2 s" !!$q'473 44r c e Zd Zd Zd Zy)RemoteTracebackc || _ y r tb)selfr s r __init__zRemoteTraceback.__init__: s r c | j S r r r s r __str__zRemoteTraceback.__str__< s wwr N)__name__
__module__ __qualname__r! r$ r r r r 9 s r r c e Zd Zd Zd Zy)ExceptionWithTracebackc t j t | || }dj | }|| _ d|z | _ y )N z
"""
%s""") tracebackformat_exceptiontypejoinexcr )r r1 r s r r! z ExceptionWithTraceback.__init__@ s:
'
'S 3
;
WWR[ 2%r c > t | j | j ffS r ) rebuild_excr1 r r# s r
__reduce__z!ExceptionWithTraceback.__reduce__E s TXXtww ///r N)r% r& r' r! r4 r( r r r* r* ? s &
0r r* c & t | | _ | S r )r __cause__)r1 r s r r3 r3 H s #B'CM Jr c . e Zd ZdZ fdZd Zd Z xZS )MaybeEncodingErrorzVWraps possible unpickleable errors, so they can be
safely sent through the socket.c t | | _ t | | _ t t | | j | j y r )reprr1 valuesuperr8 r! )r r1 r; __class__s r r! zMaybeEncodingError.__init__T s3 9%[
$04::Fr c < d| j d| j dS )NzError sending result: 'z '. Reason: '')r; r1 r# s r r$ zMaybeEncodingError.__str__Y s =AZZ=AXXG Gr c < d| j j d| dS )N)r= r% r# s r __repr__zMaybeEncodingError.__repr__] s !^^ 4 4d;;r )r% r& r' __doc__r! r$ rC
__classcell__r= s @r r8 r8 P s 'G
G}t% ||d }t j d |z ||
| d|ff Y d }~d }~ww xY w) 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)
isinstanceintAssertionErrorformatputgethasattrrH close_readerEOFErrorOSErrorr debug Exception_helper_reraises_exceptionr*
__traceback__r8 )inqueueoutqueue initializerinitargsmaxtaskswrap_exceptionrM rN completedtaskjobifuncr kwdsresultewrappeds r workerrg a s z(C'@+3q= 9 @ @ JKK
,,C
++Cw " X I
xI,@ 5D
< JJ7 8 #' QdD D$ /$ /0F
, a ! 4877s7V7d7TDQ 7
xI,@8 JJ.:;3 '" JJB C $.H"H*1aoo>QZF ,(F1I6G JJI
a%)* + + ,sB D
D: E4 $D76D7: E1$E,,E14 F;=4F66F;c | )z@Pickle-able helper function for use by _guarded_task_generation.r( )exs r rV rV s
Hr c 2 e Zd ZdZdd fd
Z fdZ xZS )
_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 2 || _ t | |i | y r )rm r< r! )r rm r rc r= s r r! z_PoolCache.__init__ s
$'$'r c ` t | | | s | j j d y y r )r< __delitem__rm rM )r itemr= s r rp z_PoolCache.__delitem__ s,
D! MM d # r )r% r& r' rD r! rp rE rF s @r rk rk s +/ (
$
$r rk c e Zd ZdZdZed Z d(dZej e
fdZ d Z d Z
ed
Zed Zd Zed
Zed Zd Zd Zdi fdZd)dZd)dZ d*dZd Zd+dZd+dZdi ddfdZ d*dZ d*dZ ed)d Z e d Z!ed Z"ed Z#ed Z$d Z%d! Z&d" Z'd# Z(ed$ Z)e d% Z*d& Z+d' Z,y),r zS
Class which supports an async version of applying functions to arguments.
Tc & | j |i |S r Process)ctxr rc s r ru z Pool.Process s s{{D)D))r Nr( c g | _ t | _ |xs
t | _ | j t
j | _ | j j | _ t | j | _ || _ || _
|| _ |t j xs d}|dk r t# d | t% |t& r|dk r t# d |t) | s t+ d || _ | j/ | j9 }t; j< t> j@ | j | j | j | jB | j, | j | jD | jF | j | j | j | jH || j f | _% d| jJ _& tN | jJ _ | jJ jQ t; j< t> jR | j | jT | jF | j | j f | _+ d| jV _& tN | jV _ | jV jQ t; j< t> jX | jF | jZ | j f | _. d| j\ _& tN | j\ _ | j\ jQ t_ j` | | jb | j | jD | jF | j | j | jJ | jV | j\ | j f d
| _2 tN | _ y # t0 $ rQ | j D ] }|j2 |j5 ! | j D ] }|j7 w xY w) Nrl r z&Number of processes must be at least 1r z/maxtasksperchild must be a positive int or Nonez initializer must be a callabletargetr T )r exitpriority)3_poolr
_stater _ctx
_setup_queuesqueue SimpleQueue
_taskqueue_change_notifierrk _cache_maxtasksperchild _initializer _initargsos cpu_count
ValueErrorrI rJ callable TypeError
_processes_repopulate_poolrU exitcode terminater0 _get_sentinels threadingThreadr _handle_workersru _inqueue _outqueue_wrap_exception_worker_handlerdaemonr start
_handle_tasks
_quick_put
_task_handler_handle_results
_quick_get_result_handlerr Finalize_terminate_pool
_terminate)r processesrZ r[ maxtasksperchildcontextp sentinelss r r! z
Pool.__init__ sL
,{} ++- !% 5 5 7 $*?*?@ !1 '! +!I q= EF F ' .48HA8M !RSS "8K+@ <= =# ! ! # '') ( / /''++t 4<<//4::t}}dnn##T^^T5K5K&& 43H3HJ '+ #&) # ""$ ' - -%%//4??DNN**dkk+
%)!$'! " ( / /''..$//4;;? '+ #&) # ""$ -- $&&//4==$..$**'')=)=t?Q?Q&& 5 ] ZZ::%KKM ZZ
s
O' '$Q 5Qc | j |k( r< |d| t | t | dd | j j d y y y )Nz&unclosed running multiprocessing pool )sourcer )r} ResourceWarninggetattrr rM )r _warnr s r __del__z Pool.__del__
sT ;;# :4(C!$
0t /6B%%))$/ C r c
| j }d|j d|j d| j dt | j
d S )NrA .z state=z pool_size=rB )r= r& r' r} lenr| )r clss r rC z
Pool.__repr__ sQ nnCNN#1S%5%5$6 7
& _ -Q0 1r c j | j j g}| j j g}g ||S r )r rQ r )r task_queue_sentinelsself_notifier_sentinelss r r zPool._get_sentinels s< $ 6 6 7 #'#8#8#@#@"A @%@(?@@r c X | D cg c] }t |d r |j c}S c c}w )Nsentinel)rO r )workersrg s r _get_worker_sentinelszPool._get_worker_sentinels s: 8 %+"6:6 8 8 8s 'c d}t t t | D ]A }| | }|j t j
d|z |j
d}| |= C |S )zCleanup after any worker processes which have exited due to reaching
their specified lifetime. Returns True if any workers were cleaned up.
Fcleaning up worker %dT)reversedranger r r rT r0 )poolcleanedra rg s r _join_exited_workerszPool._join_exited_workers! sb
%D *+A!WF*
2Q67
G , r c | j | j | j | j | j | j
| j | j | j | j | j
S r ) _repopulate_pool_staticr~ ru r r| r r r r r r r# s r r zPool._repopulate_pool1 s_ ++DIIt||,0OO,0JJ
,0NNDbI I Js) A $ A AAA AA c | j |dk( rOt | }| j j | j |j
|| |j f |S |dk rt dj | t j ||| }t | }| j j | j |j
t | |j f d |D S )zP
Equivalent of `map()` -- can be MUCH slower than `Pool.map()`.
r z Chunksize must be 1+, not {0:n}c 3 . K | ]
}|D ] }| y wr r( .0chunkrq s r z Pool.imap.. ?fUDDf ) r IMapIteratorr rM r _job _set_lengthr rL r
_get_tasksr r rb r r rd task_batchess r imapz Pool.imap s
>!$'F OO 11&++tXN&&
M1} 5<<!#$ $ ??49EL!$'F OO 11&++292>@ &&
@f? ?r c | j |dk( rOt | }| j j | j |j
|| |j f |S |dk rt dj | t j ||| }t | }| j j | j |j
t | |j f d |D S )zL
Like `imap()` method but ordering of results is arbitrary.
r z Chunksize must be 1+, not {0!r}c 3 . K | ]
}|D ] }| y wr r( r s r r z&Pool.imap_unordered.. r r ) r IMapUnorderedIteratorr rM r r r r rL r r r r s r imap_unorderedzPool.imap_unordered s
>*40F OO 11&++tXN&&
M1} 5<@ &&
@f? ?r c | j t | || }| j j |j d|||fgdf |S )z;
Asynchronous version of `apply()` method.
r N)r ApplyResultr rM r )r rb r rc r r rd s r r zPool.apply_async sL
T8^< v{{AtT4 @ A4 HI
r c 6 | j ||t ||| S )z9
Asynchronous version of `map()` method.
)r r r s r map_asynczPool.map_async s"
tXw 8 r c | j t |d s t | }|5t t | t | j
dz \ }}|r|dz
}t | dk( rd}t j ||| }t | |t | || } | j j | j | j || df | S )zY
Helper function to implement map, starmap and their async counterparts.
__len__N r r r )
r rO r divmodr r| r r MapResultr rM r r )
r rb r mapperr r r extrar rd s
r r zPool._map_async s
x+H~H %c(mS_q5H I IuQ x=A ItXyA 4CM8*8: --fkk.4.:<
r c t | | |j s"|j |j s!y y )N)timeout)r emptyrN )r change_notifierr s r _wait_for_updateszPool._wait_for_updates s2 Y(!'') ! "'')r c t j }|j t k( s|r~|j t k7 rk| j ||||||| |
| |
g | j
| |
}| j || |j t k( rU|r|j t k7 rk|j d t j d y )Nzworker handler exiting) r current_threadr} r r
r r r rM r rT )r cache taskqueuerv ru r r rX rY rZ r[ r r] r r threadcurrent_sentinelss r r zPool._handle_workers s
))+ mms"u )1K sGYg 'h /
A !O#";";D"A NI N ! !"3_ E mms"u )1K
d
+,r c 0 t j }t | j d D ]} \ }}d } |D ]5 }|j t
k7 rt
j d n> || 7 |r)t
j d |r|d nd} || dz d x}x}}
v d x}x}}
n t
j d t
j d |j d t
j d |D ]
} |d t
j d y # t $ r;} |d d \ }
} ||
j | d| f n# t $ r Y nw xY wY d } ~ +d } ~ ww xY w# d x}x}}
w xY w# t $ r t
j d
Y w xY w) 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)
r r iterrN r} r r rT rU _setKeyErrorrM rS )
r rM rY r r
r taskseq
set_lengthr_ re r` idxr s
r r zPool._handle_tasks s ))+#'
t#< GZD
, #D }}+
#LM !D $ "
#78)-d1g2"37+ (,++w '+++w1 $=4
JJ2 3
J JJH I LL
JJA BD
)*= % !#'8S! !#JOOC% < ' ! ! ! (,++w J JJH I Jsl /E)$D",-E)A
E4 " E&+E!4E E! E E!E E!E)!E&&E))E14 FFc t j } | }|j t k7 rt j
d n;|t j
d n#|\ }}} || j || d x}x}}l|rk|j t k7 rX | }|t j
d 5|\ }}} || j || d x}x}}|r|j t k7 rXt | d rIt j
d t d D ]% }| j j s n | ' t j
dt | |j y # t t f$ r t j
d Y y w xY w# t $ r Y *w xY w# t t f$ r t j
d Y y w xY w# t $ r Y w xY w# t t f$ r Y w xY w) Nz.result handler got EOFError/OSError -- exitingz,result handler found thread._state=TERMINATEzresult handler got sentinelz&result handler ignoring extra sentinelrQ z"ensuring that outqueue is not full
z7result handler exiting: len(cache)=%s, thread._state=%s)r r rS rR r rT r} r r r r
rO r rQ pollr )rY rN r
r r_ r` ra objs r r zPool._handle_results= s ))+
u
}}#
IJ|
89 KCC
c
3' $ #D #3+ .
2
u
|
CD KCC
c
3' $ #D #3
2" 8Y ' JJ; <
rA#++002 E #
L%j&-- )e X&
KL
X&
KL
X&
sY E/ &F F) >G 3G# /$FF F&%F&)$GG G G #G54G5c # t K t | } t t j || }|sy | |f )wr )r tupler islice)rb itsizer s r r zPool._get_tasksy s=
"Xi&&r401A)O s 68c t d )Nz:pool objects cannot be passed between processes or pickled)NotImplementedErrorr# s r r4 zPool.__reduce__ s !J r c t j d | j t k( r" ;;# DK*/D ' ! ! % %d + r c d t j d t | _ | j y )Nzterminating pool)r rT r
r} r r# s r r zPool.terminate s
%& r c t j d | j t k( r t d | j t
t fvr t d | j j | j j | j j | j D ] }|j y )Nz joining poolzPool is still runningzIn unknown state) r rT r} r r r r
r r0 r r r| )r r s r r0 z Pool.join s
>" ;;# 45 5
[[ 2
2 /0 0 !!# ! !!#A
FFH r c p t j d | j j |j rw| j
j
r\| j
j t j d |j r | j
j
rZy y y y )Nz7removing tasks from inqueue until task handler finishedr )
r rT _rlockacquireis_aliverQ r r timesleep)rX task_handlerr s r _help_stuff_finishzPool._help_stuff_finish s}
LM ##%'//*>*>*@ OO " JJqM ##%'//*>*>*@%*@%r c
d t j d t |_ |j d t |_ t j d | j ||t
| |j st
| dk7 r t d t |_ |j d |j d t j d t j |ur|j |rHt |d d r9t j d |D ] }
|
j |
j ! t j d t j |ur|j t j d t j |ur|j |rpt |d d r`t j d
|D ]E }
|
j st j d |
j z |
j G y y y ) Nzfinalizing poolz&helping task handler/workers to finishr z/Cannot have cache with result_handler not alivezjoining worker handlerr zterminating workerszjoining task handlerzjoining result handlerzjoining pool workersr )r rT r
r} rM r+ r r' rK r r r0 rO r r pid) r r rX rY r r worker_handlerr* result_handlerr
r s r r zPool._terminate_pool s
$%
!* D! '
;< w c$i@ '')E
a AC
C !* D! T
+, # # %^ ; ! GDG[1 JJ, -::%KKM
)* # # %\ 9
+, # # %^ ; ! GDG[1 JJ- .::<JJ 6 >?FFH 24r c & | j | S r )r r# s r __enter__zPool.__enter__ s r c $ | j y r )r )r exc_typeexc_valexc_tbs r __exit__z
Pool.__exit__ s r )NNr( NNr )NNN)r )-r% r& r' rD r staticmethodru r! warningswarnr r rC r r r r r r r r r r r r r r r r r r r classmethodr r r r r4 rP r r0 r+ r r1 r6 r( r r r r s O* * CE04Of %MMs 01 A
8 8
B ' '$ 9 96 1 " 8IM FJ 9 J@:@8 &(b4 BF KO : " "
- -( ++ ++Z 9) 9)v
,
3 3 jr c X e Zd Zd Zd Zd ZddZddZd Z e e
j Z y) r c || _ t j | _ t t
| _ |j | _ || _ || _ | | j | j < y r )
r| r Event_eventnext job_counterr r _callback_error_callback)r r r r s r r! zApplyResult.__init__ sO
oo' % kk ! - !% DII r c 6 | j j S r )r> is_setr# s r readyzApplyResult.ready s {{!!##r c n | j st dj | | j S )Nz{0!r} not ready)rE r rL _successr# s r
successfulzApplyResult.successful s, zz| . 5 5d ;< <}} r Nc : | j j | y r )r> r r r s r r zApplyResult.wait s !r c | j | | j st | j r | j S | j r )r rE r rG _valuerJ s r rN zApplyResult.get s8 'zz| ==;; ++ r c ^ |\ | _ | _ | j r'| j r| j | j | j r'| j s| j | j | j j | j | j = d | _ y r ) rG rL rA rB r> setr r r| r ra r s r r zApplyResult._set ss %("
t{ >>dmm NN4;; '
- KK "
r r )
r% r& r' r! rE rH r rN r r: types GenericAlias__class_getitem__r( r r r r s3 &$
" $E$6$67r r c e Zd Zd Zd Zy)r c t j | ||| d| _ d g|z | _ || _ |dk r9d| _ | j j | j | j = y ||z t ||z z | _ y )Nr Tr ) r r! rG rL
_chunksize _number_leftr> rN r r bool)r r r lengthr r s r r! zMapResult.__init__ s T4,: <
fvo # > !D KKOO DII& & 1D)9K4L LD r c | xj dz c_ |\ }}|r| j r|| j || j z |dz | j z | j dk( r`| j r| j | j | j
| j = | j j d | _ y y |s| j rd| _ || _ | j dk( r`| j r| j | j | j
| j = | j j d | _ y y )Nr r F) rV rG rL rU rA r r r> rN r| rB )r ra success_resultsuccessrd s r r zMapResult._set) s Q ( t}}CIDKK$//)1Q3*? @ A%>>NN4;;/KK * ! !
& t}} %
$ A%''((5KK * ! !
&r N)r% r& r' r! r r( r r r r s
M"r r c 0 e Zd Zd Zd ZddZeZd Zd Zy)r c B || _ t j t j | _ t t | _ |j | _ t j | _ d| _ d | _
i | _ | | j | j < y )Nr )r| r ConditionLock_condr? r@ r r collectionsdeque_items_index_length _unsorted)r r s r r! zIMapIterator.__init__G sp
(( )9:
% kk !'') !% DII r c | S r r( r# s r __iter__zIMapIterator.__iter__R s r Nc | j 5 | j j }d d d \ }}|r|S |# t $ r | j | j
k( rd | _ t d | j j | | j j }n;# t $ r/ | j | j
k( rd | _ t d t d w xY wY w xY w# 1 sw Y xY wr )
r` rc popleft
IndexErrorrd re r|
StopIterationr r )r r rq r[ r; s r r? zIMapIterator.nextU s
ZZ
1{{**, L !
1;;$,,.!%DJ'T1
(1 ;;..0D! 1{{dll2%)
+5&D0 1
1 Zs? C$<A C!B#"C!#8CC! C$ C!!C$$C-c | j 5 | j |k( r| j j | | xj dz
c_ | j | j v rn| j j | j }| j j | | xj dz
c_ | j | j v rn| j j
n|| j |< | j | j k( r | j | j = d | _
d d d y # 1 sw Y y xY wNr ) r` rd rc r rf popnotifyre r r r| rO s r r zIMapIterator._setm s
ZZ{{a ""3' q kkT^^3 ..,,T[[9CKK&&s+KK1$K kkT^^3
!!#$'q!{{dll*KK * !
ZZs CD=A!D==Ec | j 5 || _ | j | j k( r8| j j | j | j
= d | _ d d d y # 1 sw Y y xY wr )r` re rd rp r r r| )r rX s r r zIMapIterator._set_length~ sQ
ZZ!DL{{dll*
!!#KK * !
ZZs AA//A8r ) r% r& r' r! rh r? __next__r r r( r r r r E s" &, H"""r r c e Zd Zd Zy)r c H | j 5 | j j | | xj dz
c_ | j j | j | j
k( r | j | j = d | _ d d d y # 1 sw Y y xY wrn ) r` rc r rd rp re r r r| rO s r r zIMapUnorderedIterator._set sj
ZZ KK s # KK1 K JJ {{dll*KK * !
ZZs BBB!N)r% r& r' r r( r r r r s "r r c Z e Zd ZdZed Zd
dZd Zd Zed Z ed Z
d Z y) r Fc ddl m} ||i |S )Nr rt )dummyru )rv r rc ru s r ru zThreadPool.Process s "%%%r Nc 4 t j | ||| y r )r r! )r r rZ r[ s r r! zThreadPool.__init__ s
dI{H=r c t j | _ t j | _ | j j | _ | j j | _ y r )r r r r rM r rN r r# s r r zThreadPool._setup_queues sD ))+
**, --++ ..,,r c 0 | j j gS r )r rQ r# s r r zThreadPool._get_sentinels s %%--..r c g S r r( )r s r r z ThreadPool._get_worker_sentinels s r c | j d # t j $ r Y nw xY wt | D ] }| j d y )NF)block)rN r Emptyr rM )rX r* r ra s r r+ z ThreadPool._help_stuff_finish sM % ( {{ tA KK s ,,c . t j | y r )r( r) )r r r r s r r z ThreadPool._wait_for_updates s
7r )NNr( ) r% r& r' r r7 ru r! r r r r+ r r( r r r r sU O& &>- / r )Nr( NF))__all__ra r r r r r( r- rP r8 r, r r r
connectionr r
r r r
countr@ r r rU r r* r3 r8 rg rV dictrk objectr r AsyncResultr r r r r( r r r s <
' ioo 5i 0 0< <" GK +