3
\ P @ s^ d Z ddlZddlZddlmZ ddlZddlmZ ddlZddlmZ ddl m
Z
ddl Z ddl Z ddl
mZ ddlZddlZe j Zdad d
Zd ZG d d
d
eZG dd dZdd ZG dd deZG dd deZ G dd deZ dd Z dd Z d d Z d d Z!d d! Z"da#da$d"d# Z%d$d% Z&G d&d' d'e'Z(G d(d) d)ej)Z*ej+e dS )*z"Brian Quinlan (brian@sweetapp.com) N)_base)Full) SimpleQueue)wait)partialFc C sJ da ttj } x| D ]\}}|jd qW x| D ]\}}|j q2W d S )NT) _shutdownlist_threads_queuesitemsputjoin)r
tq r 2/usr/lib64/python3.6/concurrent/futures/process.py _python_exitO s r c @ s e Zd Zdd Zdd ZdS )_RemoteTracebackc C s
|| _ d S )N)tb)selfr r r r __init__a s z_RemoteTraceback.__init__c C s | j S )N)r )r r r r __str__c s z_RemoteTraceback.__str__N)__name__
__module__ __qualname__r r r r r r r ` s r c @ s e Zd Zdd Zdd ZdS )_ExceptionWithTracebackc C s0 t jt|||}dj|}|| _d| | _d S )N z
"""
%s""") tracebackformat_exceptiontyper excr )r r r r r r r g s
z _ExceptionWithTraceback.__init__c C s t | j| jffS )N) _rebuild_excr r )r r r r
__reduce__l s z"_ExceptionWithTraceback.__reduce__N)r r r r r" r r r r r f s r c C s t || _| S )N)r __cause__)r r r r r r! o s
r! c @ s e Zd Zdd ZdS ) _WorkItemc C s || _ || _|| _|| _d S )N)futurefnargskwargs)r r% r& r' r( r r r r t s z_WorkItem.__init__N)r r r r r r r r r$ s s r$ c @ s e Zd ZdddZdS ) _ResultItemNc C s || _ || _|| _d S )N)work_id exceptionresult)r r* r+ r, r r r r { s z_ResultItem.__init__)NN)r r r r r r r r r) z s r) c @ s e Zd Zdd ZdS ) _CallItemc C s || _ || _|| _|| _d S )N)r* r& r' r( )r r* r& r' r( r r r r s z_CallItem.__init__N)r r r r r r r r r- s r- c g s0 t | }x"ttj|| }|s"d S |V q
W d S )N)ziptuple itertoolsislice) chunksize iterablesitchunkr r r _get_chunks s r6 c s fdd|D S )Nc s g | ] } | qS r r ).0r' )r& r r
s z"_process_chunk..r )r& r5 r )r& r _process_chunk s r9 c C s x| j dd}|d kr(|jtj d S y|j|j|j}W nB tk
r~ } z&t||j }|jt
|j |d W Y d d }~X qX |jt
|j |d qW d S )NT)block)r+ )r, ) getr osgetpidr& r' r(
BaseExceptionr
__traceback__r) r* )
call_queue result_queueZ call_itemrer r r r _process_worker s
&
rD c C sx xr|j rd S y|jdd}W n tjk
r4 d S X | | }|jj rh|jt||j|j |j
dd q| |= qqW d S )NF)r: T) Zfullr; queueZEmptyr% Z set_running_or_notify_cancelr r- r& r' r( )pending_work_itemsZwork_idsr@ r* work_itemr r r _add_call_item_to_queue s
rH c s d fdd} fdd}|j }xlt|| dd j D } t|g| }
||
krf|j } nr| d k rd_d_d x&|j D ]\} }
|
jj t
d ~
qW |j xj D ] }|j qW | d S t
| trj| }|j sR| d S nJ| d k rR|j| jd }
|
d k rR| jrB|
jj | j n|
jj| j ~
| | ry|sr| d S W n tk
r Y nX d q(W d S ) Nc s t p d kp jS )N)r _shutdown_threadr )executorr r
shutting_down s z/_queue_management_worker..shutting_downc sZ t dd j D } x td| D ]} jd q"W j xj D ] }|j qFW d S )Nc s s | ]}|j V qd S )N)Zis_alive)r7 pr r r s zD_queue_management_worker..shutdown_worker..r )sumvaluesrangeZ
put_nowaitcloser )Znb_children_aliveirL )r@ processesr r shutdown_worker s z1_queue_management_worker..shutdown_workerc S s g | ]
}|j qS r )sentinel)r7 rL r r r r8 s z,_queue_management_worker..Tz^A process in the process pool was terminated abruptly while the future was running or pending.)Z_readerrH rO r Zrecv_brokenrI r
r% Z
set_exceptionBrokenProcessPoolclearZ terminate
isinstanceintpopr r* r+ Z
set_resultr, r )Zexecutor_referencerS rF Zwork_ids_queuer@ rA rK rT readerZ sentinelsZreadyZ result_itemr* rG rL r )r@ rJ rS r _queue_management_worker sb
r] c C sh t rtrttda ytjd} W n ttfk
r: d S X | dkrHd S | dkrTd S d| attd S )NTSC_SEM_NSEMS_MAXr z@system provides too few semaphores (%d available, 256 necessary))_system_limits_checked_system_limitedNotImplementedErrorr< sysconfAttributeError
ValueError)Z nsems_maxr r r _check_system_limitsQ s rg c c s. x(| D ] }|j x|r$|j V qW qW d S )N)reverser[ )iterableelementr r r _chain_from_iterable_of_listsh s
rk c @ s e Zd ZdS )rW N)r r r r r r r rW t s rW c sh e Zd ZdddZdd Zdd Zdd Zejjj e_ dd
d fd d
Z
dddZ ejj j e _ Z S )ProcessPoolExecutorNc C s t |d kr tj pd| _n|dkr.td|| _tj| jt | _d| j_ t
| _ t j | _
d | _i | _d| _tj | _d| _d| _i | _d S )Nr r z"max_workers must be greater than 0TF)rg r< cpu_count _max_workersrf multiprocessingZQueueEXTRA_QUEUED_CALLS _call_queueZ
_ignore_epiper
_result_queuerE _work_ids_queue_management_thread
_processesrI threadingZLock_shutdown_lockrV _queue_count_pending_work_items)r Z max_workersr r r r | s$
z ProcessPoolExecutor.__init__c C sp | j fdd}| jd krl| j tjttj| || j| j | j
| j | j fd| _d| j_ | jj
| j t| j< d S )Nc S s |j d d S )N)r )_r r r r
weakref_cb s zFProcessPoolExecutor._start_queue_management_thread..weakref_cb)targetr' T)rr rt _adjust_process_countrv ZThreadr] weakrefrefru ry rs rq Zdaemonstartr )r r{ r r r _start_queue_management_thread s
z2ProcessPoolExecutor._start_queue_management_threadc C sJ xDt t| j| jD ].}tjt| j| jfd}|j || j|j
< qW d S )N)r| r' ) rP lenru rn ro ZProcessrD rq rr r pid)r rz rL r r r r} s z)ProcessPoolExecutor._adjust_process_countc O s | j t | jrtd| jr$tdtj }t||||}|| j| j < | j
j | j | j d7 _ | j j d | j
|S Q R X d S )NzKA child process terminated abruptly, the process pool is not usable anymorez*cannot schedule new futures after shutdownr )rw rV rW rI RuntimeErrorr ZFuturer$ ry rx rs r rr r )r r& r' r( fwr r r submit s zProcessPoolExecutor.submitr )timeoutr2 c s: |dk rt dt jtt|t|d|i|d}t|S )Nr zchunksize must be >= 1.r2 )r )rf supermapr r9 r6 rk )r r& r r2 r3 Zresults) __class__r r r s zProcessPoolExecutor.mapTc C sT | j d| _W d Q R X | jr8| jjd |r8| jj d | _d | _d | _d | _d S )NT)rw rI rt rr r r rq ru )r r r r r shutdown s
z ProcessPoolExecutor.shutdown)N)T)
r r r r r r} r r Executor__doc__r r
__classcell__r r )r r rl { s
(
rl ),
__author__atexitr< Zconcurrent.futuresr rE r ro r Zmultiprocessing.connectionr rv r~ functoolsr r0 r WeakKeyDictionaryr r r rp Exceptionr r r! objectr$ r) r- r6 r9 rD rH r] ra rb rg rk r rW r rl registerr r r r . sH
%s