U
e5dzn � @ s� d Z ddlZddlZddlmZ ddlZddlmZ ddlZddl Zddl
m Z ddl Z ddl
Z
ddlmZ ddlZddlZddlZe
�� ZdaG dd � d �Zd
d � Zd Zd
ZG dd� de�ZG dd� d�Z 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/�Z0e�1e� dS )1z"Brian Quinlan ([email protected])� 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�_writer��self� r �2/usr/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_shutdown�list�_threads_wakeups�itemsr �join)r �_�
thread_wakeup�tr 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""")� traceback�format_exception�typer �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 )�future�fn�args�kwargs)r r1 r2 r3 r4 r r r
r � s z_WorkItem.__init__N�r r r r r r r r
r0 � s r0 c @ s e Zd Zddd�ZdS )� _ResultItemNc C s || _ || _|| _d S r )�work_id� exception�result)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� fdd�Z� ZS )�
_SafeQueuer c s || _ t� j||d� d S )N)�ctx)�pending_work_items�superr )r �max_sizer< r= �� __class__r r
r � s z_SafeQueue.__init__c sl t |t�rZt�t|�||j�}td�d�|���|_ | 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 �e�objr% � 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 )�zip�tuple� itertools�islice)� chunksize� iterables�it�chunkr r r
� _get_chunks� s
rT c s � fdd�|D �S )Nc s g | ] }� |� �qS r r )�.0r3 �r2 r r
�
� s z"_process_chunk..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 t k
r� } z t ||j �}t
||j|d� W 5 d }~X Y nX t
||j|d� ~~q d ��fdd�}� �fdd�}|j } |j }
| |
g} t||� � dd� ��� D �} tj�| | �}
d }d}| |
kr�z| �� }d}W q� tk
r� } zt� t
|�||j �}W 5 d }~X Y q�X n|
|
kr�d}d }|� � |�rl| � ��d k r�d �_
d�_d �td
�}|d k �r td d �|�� d
��|_|�� D ]\}}|j�|� ~�q(|� � ��� D ]}|�� �qR|� d S t|t��r���|�}|�� ��s�|� d S nL|d k �r�|�|jd �}|d k �r�|j�r�|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..shutting_downc s� t dd� ��� D ��} | }d}||k r�| dkr�t|| �D ]6}z� �d � |d7 }W q: tk
rn Y qrY q:X q:t dd� ��� D ��} q � �� ��� D ] }|�� q�d S )Nc s s | ]}|� � V qd S r �Zis_alive�rU �pr r r
� F s zD_queue_management_worker..shutdown_worker..r r! c s s | ]}|� � V qd S r rn ro r r r
rq R s )�sum�values�rangeZ
put_nowaitr r r )Zn_children_aliveZn_children_to_stopZn_sentinels_sent�irp )rd � processesr r
�shutdown_workerD s
z1_queue_management_worker..shutdown_workerc S s g | ]
}|j �qS r )�sentinelro r r r
rW i 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 rj rs r Z
connection�waitZrecvr[ 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_sentinelsZready�causeZ is_brokenZ result_itemrH Zbper7 rJ rp r )rd rl rv r
�_queue_management_worker" s� � (
�
r~ c C sj t rtrtt��da zt�d�} W n ttfk
r< Y d S X | dkrJd S | dkrVd S d| att��d S )NT�SC_SEM_NSEMS_MAX������ z@system provides too few semaphores (%d available, 256 necessary))�_system_limits_checked�_system_limited�NotImplementedErrorrb �sysconf�AttributeError�
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 Zddd�Zdd� Zdd� Zd d
� Zejjj e_ ejjj
e_
dd d �� fd
d�
Z ddd�Z ejj j
e _
� Z
S )�ProcessPoolExecutorNr c C s t � |d kr6t�� pd| _tjdkrntt| j�| _n8|dkrHtd��n tjdkrh|tkrhtdt� ���|| _|d kr~t �
� }|| _ |d k r�t |�s�t
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_workers�sys�platform�min�_MAX_WINDOWS_WORKERSr� r Z get_context� _mp_context�callable� TypeError� _initializer� _initargs�_queue_management_thread�
_processesrk � threadingZLock�_shutdown_lockrz � _queue_count�_pending_work_items�EXTRA_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 j�d� |�� d S )Nz?Executor collected: triggering callback for QueueManager wakeup)r �util�debugr )r r r r r
�
weakref_cbB s zFProcessPoolExecutor._start_queue_management_thread..weakref_cbZQueueManagerThread)�targetr3 �nameT)r� r� �_adjust_process_countr� ZThreadr~ �weakref�refr� r� r� r� r� Zdaemon�startr )r r� r r r
� _start_queue_management_thread= s(
�
��
�z2ProcessPoolExecutor._start_queue_management_threadc C sP t t| j�| j�D ]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&td��nHd|krZ|�d�}| ^}} dd l}|jdtdd� ntdt | �d ��|j�� |jr�t|j��|j r�t
d ��t r�t
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 �warnings�warn�DeprecationWarningr� rz r{ rk � RuntimeErrorr r ZFuturer0 r� r� r� rZ r� r r� )r3 r4 r r2 r� �f�wr r r
�submitc s<
�
�
zProcessPoolExecutor.submitr! )�timeoutrP c s: |dk rt d��t� 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
�. sT
)& ! P