← Back
Editing: pool.cpython-310.pyc
o 0�i� � @ sd 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 �;/opt/alt/python310/lib64/python3.10/multiprocessing/pool.py�mapstar/ � r c C s t t�| d | d ��S )Nr r )r � itertools�starmapr r r r � starmapstar2 s r c @ � e Zd Zdd� Zdd� ZdS )�RemoteTracebackc C s || _ d S r ��tb)�selfr r r r �__init__: � 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 @ r )�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 � 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�superr0 r )r r+ r2 �� __class__r r r T s zMaybeEncodingError.__init__c C s d| j | jf S )Nz(Error sending result: '%s'. Reason: '%s')r2 r+ r r r r r! Y s �zMaybeEncodingError.__str__c C s d| j j| f S )Nz<%s: %s>)r5 r" r r r r �__repr__] r. zMaybeEncodingError.__repr__)r" r# r$ �__doc__r r! r6 � __classcell__r r r4 r r0 P s r0 r Fc C s� |d urt |t�r |dkstd�|���|j}| j}t| d�r)| j�� |j �� |d ur1||� d}|d u s=|r�||k r�z|� } W n t t fyR t � d� Y n�w | d u r]t � d� n�| \} } } } }z d| | i |��f}W n" ty� } z|r�| tur�t||j�}d|f}W Y d }~nd }~ww z || | |f� W n) ty� } z t||d �}t � d | � || | d|ff� W Y d }~nd }~ww d } } } } } }|d7 }|d u s=|r�||k s=t � 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�hasattrr9 �close�_reader�EOFError�OSErrorr �debug� Exception�_helper_reraises_exceptionr% � __traceback__r0 )�inqueue�outqueue� initializer�initargsZmaxtasks�wrap_exceptionr> r? Z completed�task�job�i�funcr �kwds�result�e�wrappedr r r �workera sX � ��� ��� rV c C s | �)z@Pickle-able helper function for use by _guarded_task_generation.r )Zexr r r rG � � rG 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 )rZ r3 r )r rZ r rR r4 r r r � s z_PoolCache.__init__c s$ t � �|� | s| j�d � d S d S r )r3 � __delitem__rZ r> )r �itemr4 r r r[ � s �z_PoolCache.__delitem__)r" r# r$ r7 r r[ r8 r r r4 r rX � s rX c @ s� e Zd ZdZdZedd� �Z dLdd�Zej e fd d �Z d d � Z d d� Z edd� �Zedd� �Zdd� Zedd� �Zedd� �Zdd� Zdd � Zdi fd d �ZdMd d �ZdMd!d"�Z dNd#d$�Zd%d&� ZdOd(d)�ZdOd*d+�Zdi ddfd,d-�Z dNd.d/�Z dNd0d1�Z edMd2d3��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 )Pr zS Class which supports an async version of applying functions to arguments. Tc O s | j |i |��S r ��Process)�ctxr rR r r r r^ � s z Pool.ProcessNr c C s0 g | _ t| _|p t� | _| �� t�� | _| j�� | _ t | j d�| _ || _ || _ || _|d u r5t�� p4d}|dk r=td��|d urNt|t�rJ|dkrNtd��|d urZt|�sZtd��|| _z| �� W n! ty� | j D ] }|jd u rx|�� qm| j D ]}|�� q|� w | � � }t j t j | j | j| j| j!| j| j | j"| j#| j | j| j | j$|| j fd�| _%d| j%_&t'| j%_| j%�(� t j t j)| j| j*| j#| j | j fd�| _+d| j+_&t'| j+_| j+�(� t j t j,| j#| j-| j fd�| _.d| j._&t'| j._| j.�(� t/j0| | j1| j| j"| j#| j | j | j%| j+| j.| j f d d �| _2t'| _d S ) NrY r z&Number of processes must be at least 1r z/maxtasksperchild must be a positive int or Nonez initializer must be a callable��targetr T� )r Z exitpriority)3�_poolr �_stater �_ctx� _setup_queues�queue� SimpleQueue� _taskqueue�_change_notifierrX �_cache�_maxtasksperchild� _initializer� _initargs�os� cpu_count� ValueErrorr: r; �callable� TypeError� _processes�_repopulate_poolrF �exitcode� terminater* �_get_sentinels� threadingZThreadr �_handle_workersr^ �_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 � processesrK rL �maxtasksperchild�context�p� sentinelsr r r r � s� � � �� �� � �� z Pool.__init__c C sF | j |kr |d| ��t| d� t| dd �d ur!| j�d � d S d S d S )Nz&unclosed running multiprocessing pool )�sourcerj )rd �ResourceWarning�getattrrj 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=�>)r5 r# r$ rd �lenrc )r �clsr r r r6 s ��z Pool.__repr__c C s | j jg}| jjg}g |�|�S r )r| rB rj )r Ztask_queue_sentinelsZself_notifier_sentinelsr r r rx s zPool._get_sentinelsc C s dd� | D �S )Nc S s g | ] }t |d �r|j�qS )�sentinel)r@ r� )�.0rV 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 ]}| | }|jdur%t�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� rv r rE r* )�poolZcleanedrP rV 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_staticre r^ rt rc r{ r| rm rn rl r} r r r r ru 1 s �zPool._repopulate_poolc C sf t |t|� �D ](} || 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` r^ Z PoolWorkerTz added workerN) r� r� rV �name�replacer r� �appendr rE ) r_ r^ r� r� rI rJ rK rL r� rM rP �wr r r r� : s �� �z Pool._repopulate_pool_staticc C s. t �|�rt �| ||||||||| � dS dS )zEClean up any exited workers and start replacements for them. N)r r� r� ) r_ r^ r� r� rI rJ rK rL r� rM r r r �_maintain_poolM s ��zPool._maintain_poolc C s4 | j �� | _| j �� | _| jjj| _| jjj| _ d S r ) re rh r{ r| r9 �sendr� rB �recvr� r r r r rf Y s zPool._setup_queuesc C s | j tkr td��d S )NzPool not running)rd r rq 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 rQ r rR r r r �applyc s z Pool.applyc C � | � ||t|��� S )zx Apply `func` to each element in `iterable`, collecting the results in a list that is returned. )� _map_asyncr r? �r rQ �iterable� chunksizer r r r j s zPool.mapc C r� )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 q s z Pool.starmapc C � | � ||t|||�S )z= Asynchronous version of `starmap()` method. )r� r �r rQ r� r� �callback�error_callbackr r r � starmap_asyncy s �zPool.starmap_asyncc c sn � zd}t |�D ] \}}||||fi fV qW dS ty6 } z||d t|fi fV W Y d}~dS d}~ww )z�Provides a generator of tasks for imap and imap_unordered with appropriate handling for iterables which throw exceptions during iteration.���r N)� enumeraterF rG )r Z result_jobrQ r� rP �xrT r r r �_guarded_task_generation� s ��$��z Pool._guarded_task_generationr c C � | � � |dkr t| �}| j�| �|j||�|jf� |S |dk r(td�|���t � |||�}t| �}| j�| �|jt |�|jf� dd� |D �S )zP Equivalent of `map()` -- can be MUCH slower than `Pool.map()`. r z Chunksize must be 1+, not {0:n}c s � � | ] }|D ]}|V qqd S r r �r� �chunkr\ r r r � <genexpr>� � � z Pool.imap.<locals>.<genexpr>) r� � IMapIteratorri r> r� �_job� _set_lengthrq r= r � _get_tasksr �r rQ r� r� rS � task_batchesr r r �imap� s4 �������z Pool.imapc C r� )zL Like `imap()` method but ordering of results is arbitrary. r z Chunksize must be 1+, not {0!r}c s r� r r r� r r r r� � r� z&Pool.imap_unordered.<locals>.<genexpr>) r� �IMapUnorderedIteratorri r> r� r� r� rq r= 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� � ApplyResultri r> r� )r rQ r rR r� r� rS r r r r� � s zPool.apply_asyncc C r� )z9 Asynchronous version of `map()` method. )r� r r� r r r � map_async� s �zPool.map_asyncc C s� | � � t|d�s t|�}|du r%tt|�t| j�d �\}}|r%|d7 }t|�dkr-d}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� rc r r� � MapResultri r> r� r� ) r rQ r� Zmapperr� r� r� Zextrar� rS r r r r� � s, ����zPool._map_asyncc C s, t | |d� |�� s|�� |�� r d S d S )N)�timeout)r �emptyr? )r� �change_notifierr� r r r �_wait_for_updates� s �zPool._wait_for_updatesc C s� t �� }|jtks|r9|jtkr9| �||||||| | | | � g | �|��| �}| �||� |jtks|r9|jtks|�d � t � d� d S )Nzworker handler exiting) ry �current_threadrd r r r� r� r� r> r rE )r� �cache� taskqueuer_ r^ r� r� rI rJ rK rL r� rM r� r� �threadZcurrent_sentinelsr r r rz � s � � zPool._handle_workersc C st t �� }t| jd �D ]z\}}d }zm|D ]D}|jtkr!t�d� nTz||� W q tyW } z$|d d� \} } z || � | d| f� W n t yL Y nw W Y d } ~ qd } ~ ww |rmt�d� |re|d nd} || d � W d } }} q W d } }} n d } }} w t�d� zt�d� |� d � t�d � |D ]} |d � q�W n t y� t�d � Y nw 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) ry r� �iterr? rd r r rE rF �_set�KeyErrorr> rD ) r� r> rJ r� r� r� ZtaskseqZ set_lengthrN rT rO �idxr� r r r r� sN ��� � � �zPool._handle_tasksc C s� t �� } z|� }W n ttfy t�d� Y d S w |jtkr0|jtks*J d��t�d� n*|d u r:t�d� n |\}}}z || � ||� W n t yR Y nw d } }}q|r�|jtkr�z|� }W n ttfyw t�d� Y d S w |d u r�t�d� qZ|\}}}z || � ||� W n t y� Y nw d } }}|r�|jtksat | d�r�t�d� zt d �D ] }| j �� s� n|� q�W n ttfy� Y nw t�d t|�|j� d S ) Nr z.result handler got EOFError/OSError -- exitingzThread not in TERMINATEz,result handler found thread._state=TERMINATEzresult handler got sentinelz&result handler ignoring extra sentinelrB z"ensuring that outqueue is not full� z7result handler exiting: len(cache)=%s, thread._state=%s)ry r� rD rC r rE rd r r r� r� r@ r� rB �pollr� )rJ r? r� r� rN rO rP �objr r r r� = sn � � � � � � �� �zPool._handle_resultsc c s0 � t |�} tt�||��}|sd S | |fV qr )r� �tupler �islice)rQ �it�sizer� r r r r� y s � �zPool._get_tasksc C s t d��)Nz:pool objects cannot be passed between processes or pickled)�NotImplementedErrorr r r r r- � s �zPool.__reduce__c C s6 t �d� | jtkrt| _t| j_| j�d � d S d S )Nz closing pool)r rE rd r r r~ rj r> r r r r rA � s �z Pool.closec C s t �d� t| _| �� d S )Nzterminating pool)r rE r rd r� r r r r rw � s zPool.terminatec C sh t �d� | jtkrtd��| jttfvrtd��| j�� | j �� | j �� | j D ]}|�� q+d S )Nz joining poolzPool is still runningzIn unknown state) r rE rd r rq r r r~ r* r� r� rc )r r� r r r r* � s �z Pool.joinc C s\ t �d� | j�� |�� r(| j�� r,| j�� t� d� |�� r*| j�� sd S d S d S d S )Nz7removing tasks from inqueue until task handler finishedr ) r rE Z_rlock�acquire�is_aliverB r� r� �time�sleep)rI � task_handlerr� r r r �_help_stuff_finish� s "�zPool._help_stuff_finishc C sV t �d� t|_|�d � t|_t �d� | �||t|�� |�� s,t| �dkr,td��t|_|�d � |�d � t �d� t � � |urH|� � |rdt |d d�rdt �d� |D ] } | j d u rc| �� qXt �d� t � � |urs|� � t �d � t � � |ur�|� � |r�t |d d�r�t �d � |D ]} | �� r�t �d | j � | � � q�d S d S d S ) Nzfinalizing poolz&helping task handler/workers to finishr z.Cannot have cache with result_hander not alivezjoining worker handlerrw zterminating workerszjoining task handlerzjoining result handlerzjoining pool workersr� )r rE r rd r> r� r� r� r< ry r� r* r@ rv rw �pid) r� r� rI rJ r� r� Zworker_handlerr� Zresult_handlerr� r� r r r r� � sJ � � ��zPool._terminate_poolc C s | � � | S r )r� r r r r � __enter__� s zPool.__enter__c C s | � � d S r )rw )r �exc_typeZexc_valZexc_tbr r r �__exit__� r z Pool.__exit__)NNr NNr )NNN)r )-r" r# r$ r7 r} � staticmethodr^ r �warnings�warnr r� r6 rx r� r� ru r� r� rf r� r� r r r� r� r� r� r� r� r� r� � classmethodrz r� r� r� r- rA rw r* r� r� r� r� r r r r r � sx �S � � � � - ; 5 c @ 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 ) rc ry ZEvent�_event�next� job_counterr� rk � _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� r zApplyResult.readyc C s | � � s td�| ���| jS )Nz{0!r} not ready)r� rq r= �_successr r r r � successful� s zApplyResult.successfulNc C s | j �|� d S r )r� r �r r� r r r r � r. zApplyResult.waitc C s( | � |� | �� s t�| jr| jS | j�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� �setrk r� rc �r rP r� r r r r� s zApplyResult._setr ) r" r# r$ r r� r� r r? r� r� �types� GenericAlias�__class_getitem__r r r r r� � s r� c @ r )r� c C sj t j| |||d� d| _d g| | _|| _|dkr(d| _| j�� | j| j = d S || t || � | _d S )Nr� Tr ) r� r r� r� � _chunksize� _number_leftr� r� rk r� �bool)r r� r� �lengthr� r� r r r r s � zMapResult.__init__c C s� | j d8 _ |\}}|r>| jr>|| j|| j |d | j �< | j dkr<| jr-| �| j� | j| j= | j�� d | _ d S d S |sI| jrId| _|| _| j dkrf| j rW| � | j� | j| j= | j�� d | _ d S d S )Nr r F) r r� r� r r� rk r� r� r� rc r� )r rP Zsuccess_result�successrS 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 )rc ry Z ConditionZLock�_condr� r� r� rk � collections�deque�_items�_index�_length� _unsorted)r r� r r r r G s zIMapIterator.__init__c C s | S r r r r r r �__iter__R s zIMapIterator.__iter__Nc C s� | j �I z| j�� }W n9 tyD | j| jkr d | _td �| j �|� z| j�� }W n tyA | j| jkr>d | _td �t d �w Y nw W d � n1 sOw Y |\}}|r\|S |�r ) r r �popleft� IndexErrorr r rc � StopIterationr r )r r� r\ r r2 r r r r� U s0 ���� �zIMapIterator.nextc C s� | j �\ | j|kr<| j�|� | jd7 _| j| jv r6| j�| j�}| j�|� | jd7 _| j| jv s | j �� n|| j|< | j| jkrW| j| j = d | _ W d � d S W d � d S 1 sbw Y d S �Nr ) r r r r� r �pop�notifyr rk r� rc r� r r r r� m s" � � "�zIMapIterator._setc C sh | j �' || _| j| jkr"| j �� | j| j= d | _W d � d S W d � d S 1 s-w Y d S r )r r r r rk r� rc )r r r r r r� ~ s �"�zIMapIterator._set_lengthr ) r" r# r$ r r r� �__next__r� r� r r r r r� E s r� c @ s e Zd Zdd� ZdS )r� c C s| | j �1 | j�|� | jd7 _| j �� | j| jkr,| j| j= d | _W d � d S W d � d S 1 s7w Y d S r ) r r r� r r r rk r� rc 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 r] )Zdummyr^ )r_ r rR r^ r r r r^ � s zThreadPool.ProcessNr c C s t �| |||� d S r )r r )r r� rK rL r r r r � s zThreadPool.__init__c C s, t �� | _t �� | _| jj| _| jj| _d S r )rg rh r{ r| r> r� r? r� r r r r rf � s zThreadPool._setup_queuesc C s | j jgS r )rj rB r r r r rx � r zThreadPool._get_sentinelsc C s g S r r r� r r r r� � rW z ThreadPool._get_worker_sentinelsc C sB z | j dd� q tjy Y nw t|�D ]}| �d � qd S )NTF)�block)r? rg ZEmptyr� r> )rI r� r� rP 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$ r} r� r^ r rf rx r� r� r� r r r r r � s )Nr NF))�__all__r r ro rg ry r� r' r� r� r& r r r Z connectionr r r r r �countr� r r rF r r% r, r0 rV rG �dictrX �objectr r� Z AsyncResultr� r� r� r r r r r �<module> sP �- @++E
Save File
Cancel