← Back
Editing: pool.cpython-39.pyc
a XC?h+ � @ s` d dgZ ddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl m Z m Z ddlmZ dZd Zd Zd Ze�� Zd d � Zdd� ZG dd� de�ZG dd� d�Zdd� ZG dd� de�Z d*dd�Z d d � Z G d d � d e �Z G d d � d e!�Z"G d!d"� d"e!�Z#e#Z$G d#d$� d$e#�Z%G d%d&� d&e!�Z&G d'd(� d(e&�Z'G d)d� de"�Z(dS )+�Pool� ThreadPool� N� )�util)� get_context� TimeoutError)�wait�INIT�RUN�CLOSE� TERMINATEc C s t t| � �S �N)�list�map��args� r �9/opt/alt/python39/lib64/python3.9/multiprocessing/pool.py�mapstar/ s r c C s t t�| d | d ��S )Nr r )r � itertools�starmapr r r r � starmapstar2 s r c @ s e Zd Zdd� Zdd� ZdS )�RemoteTracebackc C s || _ d S r ��tb)�selfr r r r �__init__: s zRemoteTraceback.__init__c C s | j S r r �r r r r �__str__< s zRemoteTraceback.__str__N)�__name__� __module__� __qualname__r r r r r r r 9 s r c @ s e Zd Zdd� Zdd� ZdS )�ExceptionWithTracebackc C s0 t �t|�||�}d�|�}|| _d| | _d S )N� z """ %s""")� traceback�format_exception�type�join�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__E 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) H s r) c s0 e Zd ZdZ� fdd�Zdd� Zdd� Z� ZS )�MaybeEncodingErrorzVWraps possible unpickleable errors, so they can be safely sent through the socket.c s. t |�| _t |�| _tt| ��| j| j� d S r )�reprr( �value�superr, r )r r( r. �� __class__r r r T s zMaybeEncodingError.__init__c C s d| j | jf S )Nz(Error sending result: '%s'. Reason: '%s')r. r( r r r r r Y s �zMaybeEncodingError.__str__c C s d| j j| f S )Nz<%s: %s>)r1 r r r r r �__repr__] s zMaybeEncodingError.__repr__)r r r! �__doc__r r r2 � __classcell__r r r0 r r, P s r, r Fc C s� |d ur(t |t�r|dks(td�|���|j}| j}t| d�rR| j�� |j �� |d urb||� d}|d u s~|�r�||k �r�z |� } W n& t t fy� t � d� Y �q�Y n0 | d u r�t � d� �q�| \} } } } }zd| | i |��f}W nJ t�y4 } z0|�r| tu�rt||j�}d|f}W Y d }~n d }~0 0 z|| | |f� W nT t�y� } z:t||d �}t � d | � || | d|ff� W Y d }~n d }~0 0 d } } } } } }|d7 }qft � d | � d S ) 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)� isinstance�int�AssertionError�format�put�get�hasattrr5 �close�_reader�EOFError�OSErrorr �debug� Exception�_helper_reraises_exceptionr" � __traceback__r, )�inqueue�outqueue� initializer�initargsZmaxtasks�wrap_exceptionr: r; Z completed�task�job�i�funcr �kwds�result�e�wrappedr r r �workera sN � �( rR c C s | �dS )z@Pickle-able helper function for use by _guarded_task_generation.Nr )Zexr r r rC � s rC c s2 e Zd ZdZdd�� fdd� Z� fdd�Z� ZS )� _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. N��notifierc s || _ t� j|i |�� d S r )rU r/ r )r rU r rN r0 r r r � s z_PoolCache.__init__c s t � �|� | s | j�d � d S r )r/ � __delitem__rU r: )r �itemr0 r r rV � s z_PoolCache.__delitem__)r r r! r3 r rV r4 r r r0 r rS � s rS c @ s� e Zd ZdZdZedd� �ZdLdd�Zej e fd d �Z d d � Z d d� Z edd� �Zedd� �Zdd� Zedd� �Zedd� �Zdd� Zdd � Zdi fd d �ZdMd d �ZdNd!d"�ZdOd#d$�Zd%d&� ZdPd(d)�ZdQd*d+�Zdi ddfd,d-�Z dRd.d/�Z dSd0d1�Z edTd2d3��Z e d4d5� �Z!ed6d7� �Z"ed8d9� �Z#ed:d;� �Z$d<d=� Z%d>d?� Z&d@dA� Z'dBdC� Z(edDdE� �Z)e dFdG� �Z*dHdI� Z+dJdK� Z,dS )Ur zS Class which supports an async version of applying functions to arguments. Tc O s | j |i |��S r ��Process)�ctxr rN r r r rY � s z Pool.ProcessNr c C s g | _ t| _|pt� | _| �� t�� | _| j�� | _ t | j d�| _ || _ || _ || _|d u rjt�� phd}|dk rztd��|d ur�t|�s�td��|| _z | �� W nF ty� | j D ]}|jd u r�|�� q�| j D ] }|�� qւ Y n0 | �� }tj t j | j | j| j| j | j| j | j | j!| j | j| j | j"|| j fd�| _#d| j#_$t%| j#_| j#�&� tj t j'| j| j(| j!| j | j fd�| _)d| j)_$t%| j)_| j)�&� tj t j*| j!| j+| j fd�| _,d| j,_$t%| j,_| j,�&� t-j.| | j/| j| j | j!| j | j | j#| j)| j,| j f dd�| _0t%| _d S ) NrT r z&Number of processes must be at least 1z initializer must be a callable��targetr T� )r Z exitpriority)1�_poolr �_stater �_ctx� _setup_queues�queue� SimpleQueue� _taskqueue�_change_notifierrS �_cache�_maxtasksperchild� _initializer� _initargs�os� cpu_count� ValueError�callable� TypeError� _processes�_repopulate_poolrB �exitcode� terminater' �_get_sentinels� threadingZThreadr �_handle_workersrY �_inqueue� _outqueue�_wrap_exception�_worker_handler�daemonr �start� _handle_tasks� _quick_put� _task_handler�_handle_results� _quick_get�_result_handlerr ZFinalize�_terminate_pool� _terminate)r � processesrG rH �maxtasksperchild�context�p� sentinelsr r r r � s~ �� �� � ��z Pool.__init__c C s> | j |kr:|d| ��t| d� t| dd �d ur:| j�d � d S )Nz&unclosed running multiprocessing pool )�sourcere )r_ �ResourceWarning�getattrre r: )r Z_warnr r r r �__del__ s �z Pool.__del__c C s0 | j }d|j� d|j� d| j� dt| j�� d� S )N�<�.z state=z pool_size=�>)r1 r r! r_ �lenr^ )r �clsr r r r2 s ��z Pool.__repr__c C s | j jg}| jjg}g |�|�S r )rw r> re )r Ztask_queue_sentinelsZself_notifier_sentinelsr r r rs s zPool._get_sentinelsc C s dd� | D �S )Nc S s g | ]}t |d �r|j�qS )�sentinel)r< r� )�.0rR r r r � <listcomp> s �z.Pool._get_worker_sentinels.<locals>.<listcomp>r �Zworkersr r r �_get_worker_sentinels s �zPool._get_worker_sentinelsc C sP d}t tt| ���D ]6}| | }|jdurt�d| � |�� d}| |= q|S )z�Cleanup after any worker processes which have exited due to reaching their specified lifetime. Returns True if any workers were cleaned up. FN�cleaning up worker %dT)�reversed�ranger� rq r rA r' )�poolZcleanedrL rR r r r �_join_exited_workers s zPool._join_exited_workersc C s0 | � | j| j| j| j| j| j| j| j| j | j � S r ) �_repopulate_pool_staticr` rY ro r^ rv rw rh ri rg rx r r r r rp . s �zPool._repopulate_poolc C sf t |t|� �D ]P} || t|||||| fd�} | j�dd�| _d| _| �� |�| � t� d� qdS )z�Bring the number of pool processes up to the specified number, for use after reaping workers which have exited. r[ rY Z PoolWorkerTz added workerN) r� r� rR �name�replacerz r{ �appendr rA ) rZ rY r� r� rE rF rG rH r� rI rL �wr r r r� 7 s �� z Pool._repopulate_pool_staticc C s* t �|�r&t �| ||||||||| � dS )zEClean up any exited workers and start replacements for them. N)r r� r� ) rZ rY r� r� rE rF rG rH r� rI r r r �_maintain_poolJ s �zPool._maintain_poolc C s4 | j �� | _| j �� | _| jjj| _| jjj| _ d S r ) r` rc rv rw r5 �sendr} r> �recvr� r r r r ra V s zPool._setup_queuesc C s | j tkrtd��d S )NzPool not running)r_ r rl r r r r �_check_running\ s zPool._check_runningc C s | � |||��� S )zT Equivalent of `func(*args, **kwds)`. Pool must be running. )� apply_asyncr; )r rM r rN r r r �apply` s z Pool.applyc C s | � ||t|��� S )zx Apply `func` to each element in `iterable`, collecting the results in a list that is returned. )� _map_asyncr r; �r rM �iterable� chunksizer r r r g s zPool.mapc C s | � ||t|��� S )z� Like `map()` method but the elements of the `iterable` are expected to be iterables as well and will be unpacked as arguments. Hence `func` and (a, b) becomes func(a, b). )r� r r; r� r r r r n s z Pool.starmapc C s | � ||t|||�S )z= Asynchronous version of `starmap()` method. )r� r �r rM r� r� �callback�error_callbackr r r � starmap_asyncv s �zPool.starmap_asyncc c sl z,d}t |�D ]\}}||||fi fV qW n: tyf } z"||d t|fi fV W Y d}~n d}~0 0 dS )z�Provides a generator of tasks for imap and imap_unordered with appropriate handling for iterables which throw exceptions during iteration.���r N)� enumeraterB rC )r Z result_jobrM r� rL �xrP r r r �_guarded_task_generation~ s z Pool._guarded_task_generationr c C s� | � � |dkr:t| �}| j�| �|j||�|jf� |S |dk rPtd�|���t � |||�}t| �}| j�| �|jt |�|jf� dd� |D �S dS )zP Equivalent of `map()` -- can be MUCH slower than `Pool.map()`. r z Chunksize must be 1+, not {0:n}c s s | ]}|D ] }|V q qd S r r �r� �chunkrW r r r � <genexpr>� � z Pool.imap.<locals>.<genexpr>N) r� � IMapIteratorrd r: r� �_job� _set_lengthrl r9 r � _get_tasksr �r rM r� r� rO � task_batchesr r r �imap� s4 �������z Pool.imapc C s� | � � |dkr:t| �}| j�| �|j||�|jf� |S |dk rPtd�|���t � |||�}t| �}| j�| �|jt |�|jf� dd� |D �S dS )zL Like `imap()` method but ordering of results is arbitrary. r z Chunksize must be 1+, not {0!r}c s s | ]}|D ] }|V q qd S r r r� r r r r� � r� z&Pool.imap_unordered.<locals>.<genexpr>N) r� �IMapUnorderedIteratorrd r: r� r� r� rl r9 r r� r r� r r r �imap_unordered� s0 ������zPool.imap_unorderedc C s6 | � � t| ||�}| j�|jd|||fgdf� |S )z; Asynchronous version of `apply()` method. r N)r� � ApplyResultrd r: r� )r rM r rN r� r� rO r r r r� � s zPool.apply_asyncc C s | � ||t|||�S )z9 Asynchronous version of `map()` method. )r� r r� r r r � map_async� s �zPool.map_asyncc C s� | � � t|d�st|�}|du rJtt|�t| j�d �\}}|rJ|d7 }t|�dkrZd}t�|||�}t| |t|�||d�} | j � | � | j ||�df� | S )zY Helper function to implement map, starmap and their async counterparts. �__len__N� r r �r� ) r� r< r �divmodr� r^ r r� � MapResultrd r: r� r� ) r rM r� Zmapperr� r� r� Zextrar� rO r r r r� � s, ����zPool._map_asyncc C s" t | |d� |�� s |�� q d S )N)�timeout)r �emptyr; )r� �change_notifierr� r r r �_wait_for_updates� s zPool._wait_for_updatesc C st t �� }|jtks |r\|jtkr\| �||||||| | | | � g | �|��| �}| �||� q|�d � t � d� d S )Nzworker handler exiting) rt �current_threadr_ r r r� r� r� r: r rA )r� �cache� taskqueuerZ rY r� r� rE rF rG rH r� rI r� r� �threadZcurrent_sentinelsr r r ru � s � zPool._handle_workersc C s� t �� }t| jd �D �] \}}d }z�|D ]�}|jtkrDt�d� q�z ||� W q( ty� } zH|d d� \} } z|| � | d| f� W n t y� Y n0 W Y d } ~ q(d } ~ 0 0 q(|r�t�d� |r�|d nd} || d � W d } }} qW d } }} �q,W d } }} qd } }} 0 qt�d� z6t�d� |� d � t�d � |D ]} |d � �qPW n t �y� t�d � Y n0 t�d � d S ) 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) rt r� �iterr; r_ r r rA rB �_set�KeyErrorr: r@ ) r� r: rF r� r� r� ZtaskseqZ set_lengthrJ rP rK �idxr� r r r r| sJ � � zPool._handle_tasksc C s� t �� }z |� }W n" ttfy4 t�d� Y d S 0 |jtkr^|jtksRJ d��t�d� q�|d u rrt�d� q�|\}}}z|| � ||� W n t y� Y n0 d } }}q|�rJ|jtk�rJz |� }W n" ttfy� t�d� Y d S 0 |d u �rt�d� q�|\}}}z|| � ||� W n t �y: Y n0 d } }}q�t | d��r�t�d� z,t d�D ] }| j �� �s� �q�|� �qjW n ttf�y� Y n0 t�d t|�|j� d S ) Nz.result handler got EOFError/OSError -- exitingzThread not in TERMINATEz,result handler found thread._state=TERMINATEzresult handler got sentinelz&result handler ignoring extra sentinelr> z"ensuring that outqueue is not full� z7result handler exiting: len(cache)=%s, thread._state=%s)rt r� r@ r? r rA r_ r r r� r� r< r� r> �pollr� )rF r; r� r� rJ rK rL �objr r r r : s\ �zPool._handle_resultsc c s0 t |�}tt�||��}|s d S | |fV qd S r )r� �tupler �islice)rM �it�sizer� r r r r� v s zPool._get_tasksc C s t d��d S )Nz:pool objects cannot be passed between processes or pickled)�NotImplementedErrorr r r r r* s �zPool.__reduce__c C s2 t �d� | jtkr.t| _t| j_| j�d � d S )Nz closing pool)r rA r_ r r ry re r: r r r r r= � s z Pool.closec C s t �d� t| _| �� d S )Nzterminating pool)r rA r r_ r� r r r r rr � s zPool.terminatec C sj t �d� | jtkr td��n| jttfvr4td��| j�� | j �� | j �� | j D ] }|�� qXd S )Nz joining poolzPool is still runningzIn unknown state) r rA r_ r rl r r ry r' r~ r� r^ )r r� r r r r' � s z Pool.joinc C s@ t �d� | j�� |�� r<| j�� r<| j�� t� d� qd S )Nz7removing tasks from inqueue until task handler finishedr ) r rA Z_rlock�acquire�is_aliver> r� r� �time�sleep)rE � task_handlerr� r r r �_help_stuff_finish� s zPool._help_stuff_finishc C sX t �d� t|_|�d � t|_t �d� | �||t|�� |�� sXt| �dkrXtd��t|_|�d � |�d � t �d� t � � |ur�|� � |r�t |d d�r�t �d� |D ]} | j d u r�| �� q�t �d� t � � |ur�|� � t �d � t � � |u�r|� � |�rTt |d d��rTt �d � |D ](} | �� �r*t �d | j � | � � �q*d S ) Nzfinalizing poolz&helping task handler/workers to finishr z.Cannot have cache with result_hander not alivezjoining worker handlerrr zterminating workerszjoining task handlerzjoining result handlerzjoining pool workersr� )r rA r r_ r: r� r� r� r8 rt r� r' r< rq rr �pid) r� r� rE rF r� r� Zworker_handlerr� Zresult_handlerr� r� r r r r� � sB � zPool._terminate_poolc C s | � � | S r )r� r r r r � __enter__� s zPool.__enter__c C s | � � d S r )rr )r �exc_typeZexc_valZexc_tbr r r �__exit__� s z Pool.__exit__)NNr NN)N)N)NNN)r )r )NNN)NNN)N)-r r r! r3 rx � staticmethodrY r �warnings�warnr r� r2 rs r� r� rp r� r� ra r� r� r r r� r� r� r� r� r� r� r� � classmethodru r| r r� r* r= rr r' r� r� r� r� r r r r r � sv � P � � � � - ; 5c @ sJ e Zd Zdd� Zdd� Zdd� Zddd �Zdd d �Zd d � Ze e j �Z dS )r� c C s> || _ t�� | _tt�| _|j| _|| _|| _ | | j| j< d S r ) r^ rt ZEvent�_event�next� job_counterr� rf � _callback�_error_callback)r r� r� r� r r r r � s zApplyResult.__init__c C s | j �� S r )r� Zis_setr r r r �ready� s zApplyResult.readyc C s | � � std�| ���| jS )Nz{0!r} not ready)r� rl r9 �_successr r r r � successful� s zApplyResult.successfulNc C s | j �|� d S r )r� r �r r� r r r r � s zApplyResult.waitc C s, | � |� | �� st�| jr"| jS | j�d S r )r r� r r� �_valuer� r r r r; � s zApplyResult.getc C sZ |\| _ | _| jr$| j r$| �| j� | jr<| j s<| �| j� | j�� | j| j= d | _d S r ) r� r� r� r� r� �setrf r� r^ �r rL r� r r r r� s zApplyResult._set)N)N) r r r! r r� r� r r; r� r� �types� GenericAlias�__class_getitem__r r r r r� � s r� c @ s e Zd Zdd� Zdd� ZdS )r� c C sh t j| |||d� d| _d g| | _|| _|dkrNd| _| j�� | j| j = n|| t || � | _d S )Nr� Tr ) r� r r� r� � _chunksize� _number_leftr� r� rf r� �bool)r r� r� �lengthr� r� r r r r s � zMapResult.__init__c C s� | j d8 _ |\}}|rv| jrv|| j|| j |d | j �< | j dkr�| jrZ| �| j� | j| j= | j�� d | _ nL|s�| jr�d| _|| _| j dkr�| j r�| � | j� | j| j= | j�� d | _ d S )Nr r F) r� r� r� r� r� rf r� r� r� r^ r� )r rL Zsuccess_result�successrO r r r r� & s& zMapResult._setN)r r r! r r� r r r r r� s r� c @ s: e Zd Zdd� Zdd� Zd dd�ZeZdd � Zd d � ZdS ) r� c C sT || _ t�t�� �| _tt�| _|j| _t � � | _ d| _ d | _ i | _| | j| j< d S )Nr )r^ rt Z ConditionZLock�_condr� r� r� rf � collections�deque�_items�_index�_length� _unsorted)r r� r r r r D s zIMapIterator.__init__c C s | S r r r r r r �__iter__O s zIMapIterator.__iter__Nc C s� | j �� z| j�� }W nv ty� | j| jkr<d | _td �| j �|� z| j�� }W n0 ty� | j| jkr|d | _td �t d �Y n0 Y n0 W d � n1 s�0 Y |\}}|r�|S |�d S r ) r� r �popleft� IndexErrorr r r^ � StopIterationr r )r r� rW r� r. r r r r� R s& 0zIMapIterator.nextc C s� | j �� | j|krn| j�|� | jd7 _| j| jv rb| j�| j�}| j�|� | jd7 _q,| j �� n || j|< | j| jkr�| j| j = d | _ W d � n1 s�0 Y d S �Nr ) r� r r r� r �pop�notifyr rf r� r^ r� r r r r� j s zIMapIterator._setc C sV | j �< || _| j| jkr4| j �� | j| j= d | _W d � n1 sH0 Y d S r )r� r r r rf r� r^ )r r� r r r r� { s zIMapIterator._set_length)N) r r r! r r r� �__next__r� r� r r r r r� B s r� c @ s e Zd Zdd� ZdS )r� c C sj | j �P | j�|� | jd7 _| j �� | j| jkrH| j| j= d | _W d � n1 s\0 Y d S r ) r� r r� r r r rf r� r^ r� r r r r� � s zIMapUnorderedIterator._setN)r r r! r� r r r r r� � s r� c @ sV e Zd ZdZedd� �Zddd�Zdd � Zd d � Zed d � �Z edd� �Z dd� Z dS )r Fc O s ddl m} ||i |��S )Nr rX )ZdummyrY )rZ r rN rY r r r rY � s zThreadPool.ProcessNr c C s t �| |||� d S r )r r )r r� rG rH r r r r � s zThreadPool.__init__c C s, t �� | _t �� | _| jj| _| jj| _d S r )rb rc rv rw r: r} r; r� r r r r ra � s zThreadPool._setup_queuesc C s | j jgS r )re r> r r r r rs � s zThreadPool._get_sentinelsc C s g S r r r� r r r r� � s z ThreadPool._get_worker_sentinelsc C sD z| j dd� qW n tjy& Y n0 t|�D ]}| �d � q0d S )NF)�block)r; rb ZEmptyr� r: )rE r� r� rL r r r r� � s z ThreadPool._help_stuff_finishc C s t �|� d S r )r� r� )r r� r� r� r r r r� � s z ThreadPool._wait_for_updates)NNr ) r r r! rx r� rY r ra rs r� r� r� r r r r r � s )Nr NF))�__all__r� r rj rb rt r� r$ r� r� r# r r r Z connectionr r r r r �countr� r r rB r r" r) r, rR rC �dictrS �objectr r� Z AsyncResultr� r� r� r r r r r �<module> sN � - =++E
Save File
Cancel