B
\NM @ s d Z ddlZddlZddlmZ ddlZddlmZ ddlZddl m
Z
ddl m Z ddl
Z
ddlZddlmZ ddlZddlZddlZe ZdaG d d
d
Zd d Zd
ZdZG dd deZ G dd dZ dd Z G dd de Z G dd de Z!G dd de Z"G dd d e Z#d d Z$d d Z%d1d!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/d0 d0ej0Z1e2e dS )2z"Brian Quinlan (brian@sweetapp.com) N)_base)Full)wait)Queue)partialFc @ s, e Zd Zdd Zdd Zdd Zdd Zd S )
_ThreadWakeupc C s t jdd\| _| _d S )NF)Zduplex)mpZPipe_reader_writer)self r ?/opt/alt/python37/lib64/python3.7/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_bytes)r r r r
wakeupY s z_ThreadWakeup.wakeupc C s x| j r| j qW d S )N)r ZpollZ
recv_bytes)r 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 sH da tt } x| D ]\}}| qW x| D ]\}}| q0W 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 )N)tb)r r" r r r
r z s z_RemoteTraceback.__init__c C s | j S )N)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 )N) _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 )N)r! __cause__)r) r" r r r
r* s
r* c @ s e Zd Zdd ZdS ) _WorkItemc C s || _ || _|| _|| _d S )N)futurefnargskwargs)r r. r/ r0 r1 r r r
r s z_WorkItem.__init__N)r r r r r r r r
r- s r- c @ s e Zd ZdddZdS ) _ResultItemNc C s || _ || _|| _d S )N)work_id exceptionresult)r r3 r4 r5 r r r
r s z_ResultItem.__init__)NN)r r r r r r r r
r2 s r2 c @ s e Zd Zdd ZdS ) _CallItemc C s || _ || _|| _|| _d S )N)r3 r/ r0 r1 )r r3 r/ r0 r1 r r r
r s z_CallItem.__init__N)r r r r r r r r
r6 s r6 c s* e Zd Zd fdd Z fddZ ZS )
_SafeQueuer c s || _ t j||d d S )N)ctx)pending_work_itemssuperr )r max_sizer8 r9 ) __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% )
isinstancer6 r& r' r(
__traceback__r! formatr r, r9 popr3 r.
set_exceptionr: _on_queue_feeder_error)r eobjr" work_item)r< r r
rB s
z!_SafeQueue._on_queue_feeder_error)r )r r r r rB
__classcell__r r )r< r
r7 s r7 c g s0 t | }x"tt|| }|s"d S |V q
W d S )N)ziptuple itertoolsislice) chunksize iterablesitchunkr r r
_get_chunks s rO c s fdd|D S )Nc s g | ] } | qS r r ).0r0 )r/ r r
s z"_process_chunk..r )r/ rN r )r/ r
_process_chunk s rR c
C s^ y| t|||d W n@ tk
rX } z"t||j}| t||d W d d }~X Y nX d S )N)r5 r4 )r4 )putr2
BaseExceptionr$ r> ) result_queuer3 r5 r4 rC r) r r r
_sendback_result s
rV c
C s |d k r:y || W n$ t k
r8 tjjddd d S X x| jdd}|d krb|t d S y|j|j |j
}W n> t k
r } z t ||j }t
||j|d W d d }~X Y nX t
||j|d ~q rV r3 )
call_queuerU initializerinitargsZ call_itemrrC r) r r r
_process_worker s$
"r` c C sx xr| rd S y|jdd}W n tjk
r4 d S X | | }|j rh|jt||j|j |j
dd q| |= qqW d S )NF)rX T) ZfullrY queueZEmptyr. Z set_running_or_notify_cancelrS r6 r/ r0 r1 )r9 Zwork_idsr\ r3 rE r r r
_add_call_item_to_queue s
rb c
sF d fdd} fdd}|j } |j }
| |
g} x t|| dd D } t| | }
d }d}| |
kry| }d}W q tk
r } ztt|||j }W d d }~X Y qX n|
|
krd}d }|
|rt| d k rd _ d_ d t
d
}|d k r td d | d
|_x$| D ]\}}|j| ~q*W |
x D ]}| qXW | d S t|tr|}| s| d S nL|d k r||jd }|d k r|jr|j|j n|j|j ~~| | r:y$d k rd_ |s | d S W n t k
r8 Y nX d q6W d S )Nc s t p d kp jS )N)r _shutdown_threadr )executorr r
shutting_down? s z/_queue_management_worker..shutting_downc s t dd D } | }d}xn||k r| dkrxBt|| D ]2}y d |d7 }W q> tk
rn P Y q>X q>W t dd D } q W x D ] }| qW d S )Nc s s | ]}| V qd S )N)is_alive)rP pr r r
E s zD_queue_management_worker..shutdown_worker..r r c s s | ]}| V qd S )N)rf )rP rg r r r
rh Q s )sumvaluesrangeZ
put_nowaitr r r )Zn_children_aliveZn_children_to_stopZn_sentinels_sentirg )r\ processesr r
shutdown_workerC s
z1_queue_management_worker..shutdown_workerc S s g | ]
}|j qS r )sentinel)rP rg r r r
rQ h s z,_queue_management_worker..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 rb rj r ZrecvrT r& r' r( r> r _brokenrc BrokenProcessPoolr! r r, r r. rA Z terminater= intr@ r3 r4 Z
set_resultr5 r )Zexecutor_referencerm r9 Zwork_ids_queuer\ rU r re rn Z
result_readerZ
wakeup_readerZreadersZworker_sentinelsZreadycauseZ is_brokenZ result_itemrC Zbper3 rE rg r )r\ rd rm r
_queue_management_worker! s (
rt c C sh t rtrttda ytd} W n ttfk
r: d S X | dkrHd S | dkrTd S d| attd S )NTSC_SEM_NSEMS_MAX z@system provides too few semaphores (%d available, 256 necessary))_system_limits_checked_system_limitedNotImplementedErrorrZ sysconfAttributeError
ValueError)Z nsems_maxr r r
_check_system_limits s r~ c c s. x(| D ] }| x|r$| V qW qW d S )N)reverser@ )iterableZelementr r r
_chain_from_iterable_of_lists s
r c @ s e Zd ZdS )rq N)r r r r r r r
rq s rq c sh e Zd ZdddZdd Zdd Zd d
Zejjj 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 win32r z"max_workers must be greater than 0zmax_workers must be <= z initializer must be a callableF)r; r8 r9 T)$r~ rZ cpu_count _max_workerssysplatformmin_MAX_WINDOWS_WORKERSr} r Z get_context _mp_contextcallable TypeError _initializer _initargs_queue_management_thread
_processesrc threadingZLock_shutdown_lockrp _queue_count_pending_work_itemsEXTRA_QUEUED_CALLSr7 _call_queueZ
_ignore_epipeZ SimpleQueue
_result_queuera r _work_idsr _queue_management_thread_wakeup)r max_workersZ
mp_contextr] r^ Z
queue_sizer r r
r sF
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_cbA s zFProcessPoolExecutor._start_queue_management_thread..weakref_cbZQueueManagerThread)targetr0 nameT)r r _adjust_process_countr ZThreadrt weakrefrefr r r r r Zdaemonstartr )r r r r r
_start_queue_management_thread< s
z2ProcessPoolExecutor._start_queue_management_threadc C sT xNt t| j| jD ]8}| jjt| j| j| j | j
fd}| || j|j < qW d S )N)r r0 )
rk lenr r r ZProcessr` r r r r r pid)r r rg r r r
r W s z)ProcessPoolExecutor._adjust_process_countc O s t | dkr| ^}}} n>| s&tdn0d|krB|d}| ^}} ntdt | d |j |jrnt|j|jr|tdtrtdt
}t ||| |}||j |j
< |j|j
| j
d7 _
|j | |S Q R X d S )N zEdescriptor 'submit' of 'ProcessPoolExecutor' object needs an argumentr/ z6submit expected at least 1 positional argument, got %dr z*cannot schedule new futures after shutdownz6cannot schedule new futures after interpreter shutdown)r r r@ r rp rq rc RuntimeErrorr r ZFuturer- r r r rS r r r )r0 r1 r r/ fwr r r
submitb s0
zProcessPoolExecutor.submitr )timeoutrK c s: |dk rt dt jtt|t|d|i|d}t|S )Nr zchunksize must be >= 1.rK )r )r} r: mapr rR rO r )r r/ r rK rL results)r< r r
r s zProcessPoolExecutor.mapTc C s | j d| _W d Q R X | jr6| j |r6| j d | _| jd k rd| j |r^| j d | _d | _ d | _
| jr| j d | _d S )NT) r rc r r r r r r Z join_threadr r )r r r r r
shutdown s"
z ProcessPoolExecutor.shutdown)NNNr )T)
r r r r r r r r Executor__doc__r r rF r r )r< r
r s
J !
r )NN)3
__author__atexitrZ concurrent.futuresr ra r Zmultiprocessingr Zmultiprocessing.connectionr Zmultiprocessing.queuesr r r functoolsr rI r r& WeakKeyDictionaryr r r r r r Exceptionr! r$ r* objectr- r2 r6 r7 rO rR rV r` rb rt rx ry r~ r ZBrokenExecutorrq r r registerr r r r
. sT
(& ! L