← Back
Editing: pool.cpython-37.pyc
B � f�h � @ s0 d dgZ ddlZddlZddlZddlZddlZddlZddlZddlm Z ddlm Z m Z dZ dZ dZe�� Zdd � Zd d � ZG d d � d e�ZG dd� d�Zdd� ZG dd� de�Zd$dd�Zdd� ZG dd � d e�ZG dd � 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� c C s t t| � �S )N)�list�map)�args� r �9/opt/alt/python37/lib64/python3.7/multiprocessing/pool.py�mapstar+ s r c C s t t�| d | d ��S )Nr r )r � itertools�starmap)r r r r � starmapstar. s r c @ s e Zd Zdd� Zdd� ZdS )�RemoteTracebackc C s || _ d S )N)�tb)�selfr r r r �__init__6 s zRemoteTraceback.__init__c C s | j S )N)r )r r r r �__str__8 s zRemoteTraceback.__str__N)�__name__� __module__� __qualname__r r r r r r r 5 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 )N)� rebuild_excr r )r r r r � __reduce__A 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! D 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 )N)�reprr �value�superr$ r )r r r& )� __class__r r r P s zMaybeEncodingError.__init__c C s d| j | jf S )Nz(Error sending result: '%s'. Reason: '%s')r& r )r r r r r U s zMaybeEncodingError.__str__c C s d| j j| f S )Nz<%s: %s>)r( r )r r r r �__repr__Y s zMaybeEncodingError.__repr__)r r r �__doc__r r r) � __classcell__r r )r( r r$ L s r$ r Fc C s� |d k r(t |t�r|dks(td�|���|j}| j}t| d�rR| j�� |j �� |d k rb||� d}�xP|d ks�|�r�||k �r�y |� } W n$ t t fk r� t � d� P Y nX | d kr�t � d� P | \} } } } }yd| | |�f}W nH tk �r. } z(|�r| tk �rt||j�}d|f}W d d }~X Y nX y|| | |f� W nR tk �r� } z2t||d �}t � d | � || | d|ff� W d d }~X Y nX d } } } } } }|d7 }qjW 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�hasattrr, �close�_reader�EOFError�OSErrorr �debug� Exception�_helper_reraises_exceptionr � __traceback__r$ )�inqueue�outqueue� initializer�initargsZmaxtasksZwrap_exceptionr1 r2 Z completed�task�job�i�funcr �kwds�result�e�wrappedr r r �worker] sJ $ rH c C s | �dS )z@Pickle-able helper function for use by _guarded_task_generation.Nr )Zexr r r r: � s r: c @ s, e Zd ZdZdZdd� Zd>dd�Zd d � Zd d � Zd d� Z dd� Z di fdd�Z d?dd�Z d@dd�Z dAdd�Zdd� ZdBd d �ZdCd d �Zdi ddfd d!�ZdDd"d#�ZdEd$d%�Zed&d'� �Zed(d)� �Zed*d+� �Zed,d-� �Zd.d/� Zd0d1� Zd2d3� Z d4d5� Z ed6d7� �Z e d8d9� �Z d:d;� Z!d<d=� Z"dS )Fr zS Class which supports an async version of applying functions to arguments. Tc O s | j j||�S )N)�_ctx�Process)r r rD r r r rJ � s z Pool.ProcessNr c C sl |pt � | _| �� t�� | _i | _t| _|| _ || _ || _ |d krPt � � pNd}|dk r`td��|d k rxt|�sxtd��|| _g | _| �� tjtj| fd�| _d| j_t| j_| j�� tjtj| j| j | j | j| jfd�| _ d| j _t| j _| j �� tjtj | j | j | jfd�| _!d| j!_t| j!_| j!�� t"j#| | j$| j| j%| j | j| j| j | j!| jfdd�| _&d S )Nr z&Number of processes must be at least 1z initializer must be a callable)�targetr T� )r Z exitpriority)'r rI � _setup_queues�queue� SimpleQueue� _taskqueue�_cache�RUN�_state�_maxtasksperchild� _initializer� _initargs�os� cpu_count� ValueError�callable� TypeError� _processes�_pool�_repopulate_pool� threadingZThreadr �_handle_workers�_worker_handler�daemon�start� _handle_tasks� _quick_put� _outqueue� _task_handler�_handle_results� _quick_get�_result_handlerr ZFinalize�_terminate_pool�_inqueue� _terminate)r � processesr> r? Zmaxtasksperchild�contextr r r r � sT z Pool.__init__c C sZ d}xPt tt| j���D ]:}| j| }|jdk rt�d| � |�� d}| j|= qW |S )z�Cleanup after any worker processes which have exited due to reaching their specified lifetime. Returns True if any workers were cleaned up. FNzcleaning up worker %dT)�reversed�range�lenr] �exitcoder r8 r )r ZcleanedrB rH r r r �_join_exited_workers� s zPool._join_exited_workersc C s| xvt | jt| j� �D ]^}| jt| j| j| j| j | j | j fd�}| j� |� |j �dd�|_ d|_|�� t�d� qW dS )z�Bring the number of pool processes up to the specified number, for use after reaping workers which have exited. )rK r rJ Z PoolWorkerTz added workerN)rq r\ rr r] rJ rH rl rf rU rV rT �_wrap_exception�append�name�replacerb rc r r8 )r rB �wr r r r^ � s zPool._repopulate_poolc C s | � � r| �� dS )zEClean up any exited workers and start replacements for them. N)rt r^ )r r r r �_maintain_pool� s zPool._maintain_poolc C s4 | j �� | _| j �� | _| jjj| _| jjj| _ d S )N) rI rO rl rf r, �sendre r5 �recvri )r r r r rM � s zPool._setup_queuesc C s | � |||��� S )zT Equivalent of `func(*args, **kwds)`. Pool must be running. )� apply_asyncr2 )r rC r rD 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 r2 )r rC �iterable� chunksizer r r r 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 r2 )r rC r� r� r r r r s z Pool.starmapc C s | � ||t|||�S )z= Asynchronous version of `starmap()` method. )r r )r rC r� r� �callback�error_callbackr r r � starmap_async s zPool.starmap_asyncc c sn y0d}x&t |�D ]\}}||||fi fV qW W n8 tk rh } z||d t|fi fV W dd}~X Y nX dS )z�Provides a generator of tasks for imap and imap_unordered with appropriate handling for iterables which throw exceptions during iteration.���r N)� enumerater9 r: )r Z result_jobrC r� rB �xrF r r r �_guarded_task_generation s z Pool._guarded_task_generationr c C s� | j tkrtd��|dkrFt| j�}| j�| �|j||�|j f� |S |dk r\td� |���t � |||�}t| j�}| j�| �|jt |�|j f� dd� |D �S dS )zP Equivalent of `map()` -- can be MUCH slower than `Pool.map()`. zPool not runningr z Chunksize must be 1+, not {0:n}c s s | ]}|D ] }|V q qd S )Nr )�.0�chunk�itemr r r � <genexpr>E s z Pool.imap.<locals>.<genexpr>N)rS rR rY � IMapIteratorrQ rP r1 r� �_job� _set_lengthr0 r � _get_tasksr )r rC r� r� rE � task_batchesr r r �imap) s( z Pool.imapc C s� | j tkrtd��|dkrFt| j�}| j�| �|j||�|j f� |S |dk r\td� |���t � |||�}t| j�}| j�| �|jt |�|j f� dd� |D �S dS )zL Like `imap()` method but ordering of results is arbitrary. zPool not runningr z Chunksize must be 1+, not {0!r}c s s | ]}|D ] }|V q qd S )Nr )r� r� r� r r r r� b s z&Pool.imap_unordered.<locals>.<genexpr>N)rS rR rY �IMapUnorderedIteratorrQ rP r1 r� r� r� r0 r r� r )r rC r� r� rE r� r r r �imap_unorderedG s& zPool.imap_unorderedc C sB | j tkrtd��t| j||�}| j�|jd|||fgdf� |S )z; Asynchronous version of `apply()` method. zPool not runningr N)rS rR rY � ApplyResultrQ rP r1 r� )r rC r rD r� r� rE r r r r} d s zPool.apply_asyncc C s | � ||t|||�S )z9 Asynchronous version of `map()` method. )r r )r rC r� r� r� r� r r r � map_asynco s zPool.map_asyncc C s� | j tkrtd��t|d�s$t|�}|dkrTtt|�t| j�d �\}}|rT|d7 }t|�dkrdd}t� |||�}t | j |t|�||d�} | j � | �| j||�df� | S )zY Helper function to implement map, starmap and their async counterparts. zPool not running�__len__N� r r )r� )rS rR rY r3 r �divmodrr r] r r� � MapResultrQ rP r1 r� r� ) r rC r� Zmapperr� r� r� Zextrar� rE r r r r w s& zPool._map_asyncc C sT t �� }x0|jtks$| jr8|jtkr8| �� t�d� q W | j � d � t � d� d S )Ng�������?zworker handler exiting) r_ �current_threadrS rR rQ � TERMINATErz �time�sleeprP r1 r r8 )�pool�threadr r r r` � s zPool._handle_workersc C sj t �� }x�t| jd �D ]�\}}d }z�x�|D ]�}|jr@t�d� P y ||� W q* tk r� } zB|d d� \} } y|| �| d| f� W n t k r� Y nX W d d } ~ X Y q*X q*W |r�t�d� |r�|d nd} || d � wP W d d } }} X qW t�d� y:t�d� |� d � t�d � x|D ]} |d � �q&W W n t k �rZ t�d � Y nX t�d � d S ) Nz'task handler found thread._state != RUNr 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) r_ r� �iterr2 rS r r8 r9 �_set�KeyErrorr1 r7 ) � taskqueuer1 r= r� �cacher� ZtaskseqZ set_lengthr@ rF rA �idx�pr r r rd � sB zPool._handle_tasksc C s� t �� }x�y |� }W n" ttfk r6 t�d� d S X |jr\|jtksPtd��t�d� P |d krpt�d� P |\}}}y|| � ||� W n t k r� Y nX d } }}q W x�|�rP|jtk�rPy |� }W n" ttfk r� t�d� d S X |d k�r t�d� q�|\}}}y|| � ||� W n t k �r@ Y nX d } }}q�W t | d��r�t�d� y,x&t d�D ]}| j �� �s�P |� �qtW W n ttfk �r� Y nX 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 sentinelr5 z"ensuring that outqueue is not full� z7result handler exiting: len(cache)=%s, thread._state=%s)r_ r� r7 r6 r r8 rS r� r/ r� r� r3 rq r5 �pollrr )r= r2 r� r� r@ rA rB �objr r r rh � s\ zPool._handle_resultsc c s4 t |�}x&tt�||��}|s"d S | |fV q W d S )N)r� �tupler �islice)rC �it�sizer� r r r r� s zPool._get_tasksc C s t d��d S )Nz:pool objects cannot be passed between processes or pickled)�NotImplementedError)r r r r r" s zPool.__reduce__c C s& t �d� | jtkr"t| _t| j_d S )Nz closing pool)r r8 rS rR �CLOSEra )r r r r r4 s z Pool.closec C s$ t �d� t| _t| j_| �� d S )Nzterminating pool)r r8 r� rS ra rm )r r r r � terminate s zPool.terminatec C sn t �d� | jtkr td��n| jttfkr4td��| j�� | j �� | j �� x| j D ] }|�� qZW d S )Nz joining poolzPool is still runningzIn unknown state) r r8 rS rR rY r� r� ra r rg rj r] )r r� r r r r & s z Pool.joinc C sD t �d� | j�� x*|�� r>| j�� r>| j�� t� d� qW d S )Nz7removing tasks from inqueue until task handler finishedr ) r r8 Z_rlock�acquire�is_aliver5 r� r| r� r� )r< � task_handlerr� r r r �_help_stuff_finish2 s zPool._help_stuff_finishc C sJ t �d� t|_t|_t �d� | �||t|�� |�� sNt|�dkrNtd��t|_|�d � t �d� t � � |k r||� � |r�t |d d�r�t �d� x |D ]} | j d kr�| �� q�W t �d� t � � |k r�|� � t �d � t � � |k r�|� � |�rFt |d d��rFt �d � x0|D ](} | �� �rt �d | j � | � � �qW d S ) Nzfinalizing poolz&helping task handler/workers to finishr z.Cannot have cache with result_hander not alivezjoining worker handlerr� zterminating workerszjoining task handlerzjoining result handlerzjoining pool workerszcleaning up worker %d)r r8 r� rS r� rr r� r/ r1 r_ r� r r3 rs r� �pid) �clsr� r< r= r� Zworker_handlerr� Zresult_handlerr� r� r r r rk ; s< zPool._terminate_poolc C s | S )Nr )r r r r � __enter__k s zPool.__enter__c C s | � � d S )N)r� )r �exc_typeZexc_valZexc_tbr r r �__exit__n s z Pool.__exit__)NNr NN)N)N)NNN)r )r )NNN)NNN)#r r r r* ru rJ r rt r^ rz rM r~ r r r� r� r� r� r} r� r � staticmethodr` rd rh r� r" r4 r� r r� � classmethodrk r� r� r r r r r � sF 8 . < 0c @ s@ e Zd Zdd� Zdd� Zdd� Zddd �Zdd d �Zd d � ZdS )r� c C s4 t �� | _tt�| _|| _|| _|| _| || j< d S )N) r_ ZEvent�_event�next� job_counterr� rQ � _callback�_error_callback)r r� r� r� r r r r w s zApplyResult.__init__c C s | j �� S )N)r� Zis_set)r r r r �ready s zApplyResult.readyc C s | � � std�| ���| jS )Nz{0!r} not ready)r� rY r0 �_success)r r r r � successful� s zApplyResult.successfulNc C s | j �|� d S )N)r� �wait)r �timeoutr r r r� � s zApplyResult.waitc C s, | � |� | �� st�| jr"| jS | j�d S )N)r� r� r r� �_value)r r� r r r r2 � s zApplyResult.getc C sT |\| _ | _| jr$| j r$| �| j� | jr<| j s<| �| j� | j�� | j| j= d S )N)r� r� r� r� r� �setrQ r� )r rB r� r r r r� � s zApplyResult._set)N)N) r r r r r� r� r� r2 r� r r r r r� u s r� c @ s e Zd Zdd� Zdd� ZdS )r� c C sf t j| |||d� d| _d g| | _|| _|dkrLd| _| j�� || j= n|| t || � | _d S )N)r� Tr ) r� r r� r� � _chunksize� _number_leftr� r� r� �bool)r r� r� �lengthr� r� r r r r � s zMapResult.__init__c C s� | j d8 _ |\}}|rp| jrp|| j|| j |d | j �< | j dkr�| jrZ| �| j� | j| j= | j�� nF|s�| jr�d| _|| _| j dkr�| j r�| � | j� | j| j= | j�� d S )Nr r F) r� r� r� r� r� rQ r� r� r� r� )r rB Zsuccess_result�successrE 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 sJ t �t �� �| _tt�| _|| _t� � | _ d| _ d | _ i | _ | || j< d S )Nr )r_ Z ConditionZLock�_condr� r� r� rQ � collections�deque�_items�_index�_length� _unsorted)r r� r r r r � s zIMapIterator.__init__c C s | S )Nr )r r r r �__iter__� s zIMapIterator.__iter__Nc C s� | j �� y| j�� }W nn tk r� | j| jkr8td �| j �|� y| j�� }W n, tk r~ | j| jkrttd �td �Y nX Y nX W d Q R X |\}}|r�|S |�d S )N) r� r� �popleft� IndexErrorr� r� � StopIterationr� r )r r� r� r� r& r r r r� � s" zIMapIterator.nextc C s� | j �� | j|krr| j�|� | jd7 _x8| j| jkrd| j�| j�}| j�|� | jd7 _q.W | j �� n || j|< | j| jkr�| j| j = W d Q R X d S )Nr ) r� r� r� rv r� �pop�notifyr� rQ r� )r rB r� r r r r� � s zIMapIterator._setc C s< | j �, || _| j| jkr.| j �� | j| j= W d Q R X d S )N)r� r� r� r� rQ 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� � s r� c @ s e Zd Zdd� ZdS )r� c C sP | j �@ | j�|� | jd7 _| j �� | j| jkrB| j| j= W d Q R X d S )Nr )r� r� rv r� r� r� rQ r� )r rB r� r r r r� s zIMapUnorderedIterator._setN)r r r r� r r r r r� s r� c @ s: e Zd ZdZedd� �Zd dd�Zdd � Zed d � �ZdS ) r Fc O s ddl m} || |�S )Nr )rJ )ZdummyrJ )r rD rJ r r r rJ s zThreadPool.ProcessNr c C s t �| |||� d S )N)r r )r rn r> r? r r r r ! s zThreadPool.__init__c C s, t �� | _t �� | _| jj| _| jj| _d S )N)rN rO rl rf r1 re r2 ri )r r r r rM $ s zThreadPool._setup_queuesc C sN yx| j dd� qW W n tjk r, Y nX xt|�D ]}| �d � q8W d S )NF)�block)r2 rN ZEmptyrq r1 )r< r� r� rB r r r r� * s z ThreadPool._help_stuff_finish)NNr ) r r r ru r� rJ r rM r� r r r r r s )Nr NF)"�__all__r_ rN r r� rW r� r r r r r rR r� r� �countr� r r r9 r r r! r$ rH r: �objectr r� Z AsyncResultr� r� r� r r r r r �<module> s@ , f')@
Save File
Cancel