U
,azn @ s d Z ddlZddlZddlmZ ddlZddlmZ ddlZddl Zddl
m Z ddl Z ddl
Z
ddlmZ ddlZddlZddlZe
ZdaG dd d Zd
d Zd Zd
ZG dd deZG dd dZ dd Z G dd de Z G dd de Z G dd de Z!G dd de Z"d d Z#d d Z$d0d d!Z%d"d# Z&d$d% Z'd&d' Z(da)da*d(d) Z+d*d+ Z,G d,d- d-ej-Z.G d.d/ d/ej/Z0e1e dS )1z"Brian Quinlan (brian@sweetapp.com) N)_base)Full)Queue)partialFc @ s, e Zd Zdd Zdd Zdd Zdd Zd S )
_ThreadWakeupc C s t jdd\| _| _d S )NF)Zduplex)mpZPipe_reader_writerself r ?/opt/alt/python38/lib64/python3.8/concurrent/futures/process.py__init__R s z_ThreadWakeup.__init__c C s | j | j d S N)r closer r
r r r
r U s
z_ThreadWakeup.closec C s | j d d S )N )r Z
send_bytesr
r r r
wakeupY s z_ThreadWakeup.wakeupc C s | j r| j q d S r )r ZpollZ
recv_bytesr
r r r
clear\ s
z_ThreadWakeup.clearN)__name__
__module__ __qualname__r r r r r r r r
r Q s r c C s@ da tt } | D ]\}}| q| D ]\}}| q*d S NT)_global_shutdownlist_threads_wakeupsitemsr join)r _
thread_wakeuptr r r
_python_exita s
r = c @ s e Zd Zdd Zdd ZdS )_RemoteTracebackc C s
|| _ d S r tb)r r% r r r
r z s z_RemoteTraceback.__init__c C s | j S r r$ r
r r r
__str__| s z_RemoteTraceback.__str__N)r r r r r& r r r r
r# y s r# c @ s e Zd Zdd Zdd ZdS )_ExceptionWithTracebackc C s0 t t|||}d|}|| _d| | _d S )N z
"""
%s""") tracebackformat_exceptiontyper excr% )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__ 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- s
r- c @ s e Zd Zdd ZdS ) _WorkItemc C s || _ || _|| _|| _d S r )futurefnargskwargs)r r1 r2 r3 r4 r r r
r s z_WorkItem.__init__Nr r r r r r r r
r0 s r0 c @ s e Zd ZdddZdS ) _ResultItemNc C s || _ || _|| _d S r )work_id exceptionresult)r r7 r8 r9 r r r
r s z_ResultItem.__init__)NNr5 r r r r
r6 s r6 c @ s e Zd Zdd ZdS ) _CallItemc C s || _ || _|| _|| _d S r )r7 r2 r3 r4 )r r7 r2 r3 r4 r r r
r s z_CallItem.__init__Nr5 r r r r
r: s r: c s* e Zd Zd fdd Z fddZ ZS )
_SafeQueuer c s || _ t j||d d S )N)ctx)pending_work_itemssuperr )r max_sizer< r= __class__r r
r s z_SafeQueue.__init__c sl t |trZtt|||j}tdd||_ | j
|j d }|d k rh|j
| nt || d S )Nz
"""
{}"""r( )
isinstancer: r) r* r+
__traceback__r# formatr r/ r= popr7 r1
set_exceptionr> _on_queue_feeder_error)r eobjr% work_itemr@ r r
rG s
z!_SafeQueue._on_queue_feeder_error)r )r r r r rG
__classcell__r r r@ r
r; s r; c g s, t | }tt|| }|s d S |V qd S r )ziptuple itertoolsislice) chunksize iterablesitchunkr r r
_get_chunks s
rT c s fdd|D S )Nc s g | ] } | qS r r ).0r3 r2 r r
<listcomp> s z"_process_chunk.<locals>.<listcomp>r )r2 rS r rV r
_process_chunk s rX c
C s^ z| t|||d W n@ tk
rX } z"t||j}| t||d W 5 d }~X Y nX d S )N)r9 r8 r8 )putr6
BaseExceptionr' rC ) result_queuer7 r9 r8 rH r, r r r
_sendback_result s
r] c
C s |d k r<z || W n& t k
r: tjjddd Y d S X | jdd}|d krb|t d S z|j|j |j
}W n> t k
r } z t ||j }t
||j|d W 5 d }~X Y nX t
||j|d ~~q<d S )NzException in initializer:T)exc_infoblockrY )r9 )r[ r ZLOGGERZcriticalgetrZ osgetpidr2 r3 r4 r' rC r] r7 )
call_queuer\ initializerinitargsZ call_itemrrH r, r r r
_process_worker s$
"rh c C sv | r d S z|jdd}W n tjk
r4 Y d S X | | }|j rh|jt||j|j |j
dd q | |= q q d S )NFr_ T) Zfullra queueZEmptyr1 Z set_running_or_notify_cancelrZ r: r2 r3 r4 )r= Zwork_idsrd r7 rJ r r r
_add_call_item_to_queue s"
rj c
s> d fdd} fdd}|j } |j }
| |
g} t|| dd D } tj| | }
d }d}| |
krz| }d}W q tk
r } zt t
|||j }W 5 d }~X Y qX n|
|
krd}d }| |rl| d k rd _
d_d td
}|d k r td d | d
|_| D ]\}}|j| ~q(| D ]}| qR| d S t|tr|}| s| d S nL|d k r||jd }|d k r|jr|j|j n|j |j ~~| | r4z&d k rd_|s| W d S W n t k
r2 Y nX d q2d S )Nc s t p d kp jS r )r _shutdown_threadr )executorr r
shutting_down@ s z/_queue_management_worker.<locals>.shutting_downc s t dd D } | }d}||k r| dkrt|| D ]6}z d |d7 }W q: tk
rn Y qrY q:X q:t dd D } q D ] }| qd S )Nc s s | ]}| V qd S r Zis_aliverU pr r r
<genexpr>F s zD_queue_management_worker.<locals>.shutdown_worker.<locals>.<genexpr>r r! c s s | ]}| V qd S r rn ro r r r
rq R s )sumvaluesrangeZ
put_nowaitr r r )Zn_children_aliveZn_children_to_stopZn_sentinels_sentirp )rd processesr r
shutdown_workerD s
z1_queue_management_worker.<locals>.shutdown_workerc S s g | ]
}|j qS r )sentinelro r r r
rW i s z,_queue_management_worker.<locals>.<listcomp>TFzKA child process terminated abruptly, the process pool is not usable anymorez^A process in the process pool was terminated abruptly while the future was running or pending.z
'''
r( z''') r rj rs r Z
connectionwaitZrecvr[ r) r* r+ rC r _brokenrk BrokenProcessPoolr# r r/ r r1 rF Z terminaterB intrE r7 r8 Z
set_resultr9 r )Zexecutor_referencerv r= Zwork_ids_queuerd r\ r rm rw Z
result_readerZ
wakeup_readerZreadersZworker_sentinelsZreadycauseZ is_brokenZ result_itemrH Zbper7 rJ rp r )rd rl rv r
_queue_management_worker" s (
r~ c C sj t rtrttda ztd} W n ttfk
r< Y d S X | dkrJd S | dkrVd S d| attd S )NTSC_SEM_NSEMS_MAX z@system provides too few semaphores (%d available, 256 necessary))_system_limits_checked_system_limitedNotImplementedErrorrb sysconfAttributeError
ValueError)Z nsems_maxr r r
_check_system_limits s r c c s& | D ] }| |r| V qqd S r )reverserE )iterableZelementr r r
_chain_from_iterable_of_lists s r c @ s e Zd ZdS )r{ N)r r r r r r r
r{ s r{ c st e Zd ZdddZdd Zdd Zd d
Zejjj e_ ejjj
e_
dd d fd
d
Z dddZ ejj j
e _
Z
S )ProcessPoolExecutorNr c C s t |d kr6t pd| _tjdkrntt| j| _n8|dkrHtdn tjdkrh|tkrhtdt || _|d kr~t
}|| _ |d k rt |st
d|| _|| _d | _i | _d| _t | _d| _d| _i | _| jt }t|| j | jd| _d | j_ | | _ t | _!t" | _#d S )
Nr! Zwin32r z"max_workers must be greater than 0zmax_workers must be <= z initializer must be a callableF)r? r< r= T)$r rb cpu_count _max_workerssysplatformmin_MAX_WINDOWS_WORKERSr r Z get_context _mp_contextcallable TypeError _initializer _initargs_queue_management_thread
_processesrk threadingZLock_shutdown_lockrz _queue_count_pending_work_itemsEXTRA_QUEUED_CALLSr; _call_queueZ
_ignore_epipeZ SimpleQueue
_result_queueri r _work_idsr _queue_management_thread_wakeup)r Z max_workersZ
mp_contextre rf Z
queue_sizer r r
r sP
z ProcessPoolExecutor.__init__c C sv | j d krr| jfdd}| tjtt| || j| j | j
| j | j | jfdd| _ d| j _
| j | jt| j < d S )Nc S s t jd | d S )Nz?Executor collected: triggering callback for QueueManager wakeup)r utildebugr )r r r r r
weakref_cbB s zFProcessPoolExecutor._start_queue_management_thread.<locals>.weakref_cbZQueueManagerThread)targetr3 nameT)r r _adjust_process_countr ZThreadr~ weakrefrefr r r r r Zdaemonstartr )r r r r r
_start_queue_management_thread= s(
z2ProcessPoolExecutor._start_queue_management_threadc C sP t t| j| jD ]8}| jjt| j| j| j | j
fd}| || j|j < qd S )N)r r3 )
rt lenr r r ZProcessrh r r r r r pid)r r rp r r r
r X s z)ProcessPoolExecutor._adjust_process_countc
O s
t | dkr| ^}}} nV| s&tdnHd|krZ|d}| ^}} dd l}|jdtdd ntdt | d |j |jrt|j|j rt
d t rt
d
t
}t||| |}||j|j< |j|j | jd7 _|j | |W 5 Q R S Q R X d S ) N zEdescriptor 'submit' of 'ProcessPoolExecutor' object needs an argumentr2 r z.Passing 'fn' as keyword argument is deprecated)
stacklevelz6submit expected at least 1 positional argument, got %dr! z*cannot schedule new futures after shutdownz6cannot schedule new futures after interpreter shutdown)r r rE warningswarnDeprecationWarningr rz r{ rk RuntimeErrorr r ZFuturer0 r r r rZ r r r )r3 r4 r r2 r fwr r r
submitc s<
zProcessPoolExecutor.submitr! )timeoutrP c s: |dk rt dt jtt|t|d|i|d}t|S )Nr! zchunksize must be >= 1.rP )r )r r> mapr rX rT r )r r2 r rP rQ Zresultsr@ r r
r s zProcessPoolExecutor.mapTc C s | j d| _W 5 Q R X | jr6| j |r6| j d | _| jd k rd| j |r^| j d | _d | _ d | _
| jr| j d | _d S r ) r rk r r r r r r Z join_threadr r )r ry r r r
shutdown s"
z ProcessPoolExecutor.shutdown)NNNr )T)r r r r r r r r Executor__text_signature____doc__r r rK r r r@ r
r s
K $
r )NN)2
__author__atexitrb Zconcurrent.futuresr ri r Zmultiprocessingr Zmultiprocessing.connectionZmultiprocessing.queuesr r r functoolsr rN r r) WeakKeyDictionaryr r r r r r Exceptionr# r' r- objectr0 r6 r: r; rT rX r] rh rj r~ r r r r ZBrokenExecutorr{ r r registerr r r r
<module>. sT
)& ! P