← Back
Editing: pool.cpython-36.pyc
3 \�e � @ s8 d dgZ ddlZddlZddlZddlZddlZddlZddlZddlm Z ddlm Z m Z dZ dZ dZej� Zdd � Zd d � ZG d d � d e�ZG dd� d�Zdd� ZG dd� de�Zdf ddfdd�Zdd� ZG dd � d e�ZG dd� de�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 �,/usr/lib64/python3.6/multiprocessing/pool.py�mapstar+ s r c C s t tj| 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 jt|�||�}dj|�}|| _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| 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$ Fc ' C s� |d ks t |�tkr |dks t�|j}| j}t| d�rJ| jj� |jj� |d k rZ||� d}�xL|d ksx|ot||k �r�y |� } W n$ t t fk r� t j d� P Y nX | d kr�t j d� P | \} } } } }yd| | |�f}W nF t k �r" } z(|o�| tk �r t||j�}d|f}W Y d d }~X nX y|| | |f� W nR t k �r� } z4t||d �}t j d| � || | d|ff� W Y d d }~X nX d } } } } } }|d7 }qbW t j d | � d S ) Nr �_writerz)worker got EOFError or OSError -- exitingz worker got sentinel -- exitingTFr z0Possible encoding error while sending result: %sz worker exiting after %d tasks)r �int�AssertionError�put�get�hasattrr, �close�_reader�EOFError�OSErrorr �debug� Exception�_helper_reraises_exceptionr � __traceback__r$ )�inqueue�outqueue� initializer�initargsZmaxtasksZwrap_exceptionr/ r0 Z completed�task�job�i�funcr �kwds�result�e�wrappedr r r �worker] sF $ rF c C s | �dS )z@Pickle-able helper function for use by _guarded_task_generation.Nr )Zexr r r r8 � s r8 c @ s6 e Zd ZdZdZdd� Zddf ddfdd�Zdd � Zd d � Zd d � Z dd� Z f i fdd�Z d=dd�Z d>dd�Z d?dd�Zdd� Zd@dd �ZdAd d �Zf i ddfd d �ZdBd!d"�ZdCd#d$�Zed%d&� �Zed'd(� �Zed)d*� �Zed+d,� �Zd-d.� Zd/d0� Zd1d2� Z d3d4� Z ed5d6� �Z e d7d8� �Z d9d:� Z!d;d<� Z"dS )Dr zS Class which supports an async version of applying functions to arguments. Tc O s | j j||�S )N)�_ctx�Process)r r rB r r r rH � s z Pool.ProcessNc C sn |pt � | _| j� tj� | _i | _t| _|| _ || _ || _ |d krPt j � pNd}|dk r`td��|d k rzt|� rztd��|| _g | _| j� tjtj| fd�| _d| j_t| j_| jj� tjtj| j| j | j | j| jfd�| _ d| j _t| j _| j j� tjtj | j | j | jfd�| _!d| j!_t| j!_| 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 rG � _setup_queues�queue�Queue� _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 rtjd| � |j� 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 r6 r )r Zcleanedr@ rF 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�}| jj |� |j jdd�|_ d|_|j� tjd� qW dS )z�Bring the number of pool processes up to the specified number, for use after reaping workers which have exited. )rI r rH Z PoolWorkerTz added workerN)ro rZ rp r[ rH rF rj rd rS rT rR �_wrap_exception�append�name�replacer` ra r r6 )r r@ �wr r r r\ � s zPool._repopulate_poolc C s | j � r| j� dS )zEClean up any exited workers and start replacements for them. N)rr r\ )r r r r �_maintain_pool� s zPool._maintain_poolc C s4 | j j� | _| j j� | _| jjj| _| jjj| _ d S )N) rG Z SimpleQueuerj rd r, �sendrc r3 �recvrg )r r r r rK � s zPool._setup_queuesc C s | j tkst�| j|||�j� S )z6 Equivalent of `func(*args, **kwds)`. )rQ rP r. � apply_asyncr0 )r rA r rB r r r �apply� s z Pool.applyc C s | j ||t|�j� S )zx Apply `func` to each element in `iterable`, collecting the results in a list that is returned. )� _map_asyncr r0 )r rA �iterable� chunksizer r r r s zPool.mapc C s | j ||t|�j� 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 r0 )r rA r~ r r r r r s z Pool.starmapc C s | j ||t|||�S )z= Asynchronous version of `starmap()` method. )r} r )r rA 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 Y dd}~X 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���)� enumerater7 r8 )r Z result_jobrA r~ r@ �xrD r r r �_guarded_task_generation s z Pool._guarded_task_generationr c C s� | j tkrtd��|dkrFt| j�}| jj| j|j||�|j f� |S |dksRt �t j |||�}t| j�}| jj| j|jt |�|j f� dd� |D �S dS )zP Equivalent of `map()` -- can be MUCH slower than `Pool.map()`. zPool not runningr c s s | ]}|D ] }|V q qd S )Nr )�.0�chunk�itemr r r � <genexpr>@ s z Pool.imap.<locals>.<genexpr>N)rQ rP rW � IMapIteratorrO rN r/ r� �_job� _set_lengthr. r � _get_tasksr )r rA r~ r rC � task_batchesr r r �imap' s" z Pool.imapc C s� | j tkrtd��|dkrFt| j�}| jj| j|j||�|j f� |S |dksRt �t j |||�}t| j�}| jj| j|jt |�|j f� dd� |D �S dS )zL Like `imap()` method but ordering of results is arbitrary. zPool not runningr c s s | ]}|D ] }|V q qd S )Nr )r� r� r� r r r r� [ s z&Pool.imap_unordered.<locals>.<genexpr>N)rQ rP rW �IMapUnorderedIteratorrO rN r/ r� r� r� r. r r� r )r rA r~ r rC r� r r r �imap_unorderedB s" zPool.imap_unorderedc C sB | j tkrtd��t| j||�}| jj|jd|||fgdf� |S )z; Asynchronous version of `apply()` method. zPool not runningr N)rQ rP rW � ApplyResultrO rN r/ r� )r rA r rB r� r� rC r r r r{ ] s zPool.apply_asyncc C s | j ||t|||�S )z9 Asynchronous version of `map()` method. )r} r )r rA r~ r r� r� r r r � map_asynch s zPool.map_asyncc C s� | j tkrtd��t|d�s$t|�}|dkrTtt|�t| j�d �\}}|rT|d7 }t|�dkrdd}tj |||�}t | j |t|�||d�} | j j | j| j||�df� | S )zY Helper function to implement map, starmap and their async counterparts. zPool not running�__len__N� r r )r� )rQ rP rW r1 r �divmodrp r[ r r� � MapResultrO rN r/ r� r� ) r rA r~ Zmapperr r� r� Zextrar� rC r r r r} p s& zPool._map_asyncc C sT t j� }x0|jtks$| jr8|jtkr8| j� tjd� q W | j j d � t j d� d S )Ng�������?zworker handler exiting) r] �current_threadrQ rP rO � TERMINATErx �time�sleeprN r/ r r6 )�pool�threadr r r r^ � s zPool._handle_workersc C sj t j� }x�t| jd �D ]�\}}d }z�x�|D ]�}|jr@tjd� P y ||� W q* tk r� } zD|d d� \} } y|| j| d| f� W n t k r� Y nX W Y d d } ~ X q*X q*W |r�tjd� |r�|d nd } || d � wP W d d } }} X qW tjd� y:tjd� |j d � tjd� x|D ]} |d � �q&W W n t k �rZ tjd � Y nX tjd � d S ) Nz'task handler found thread._state != RUNr Fzdoing set_length()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 exitingr� ) r] r� �iterr0 rQ r r6 r7 �_set�KeyErrorr/ r5 ) � taskqueuer/ r; r� �cacher� ZtaskseqZ set_lengthr> rD r? �idx�pr r r rb � sB zPool._handle_tasksc C s� t j� }x�y |� }W n" ttfk r6 tjd� d S X |jrX|jtksLt�tjd� P |d krltjd� P |\}}}y|| j ||� W n t k r� Y nX d } }}q W x�|o�|jtk�rJy |� }W n" ttfk r� tjd� d S X |d k�rtjd� q�|\}}}y|| j ||� W n t k �r: Y nX d } }}q�W t | d��r�tjd� y,x&t d�D ]}| j j� �s�P |� �qnW W n ttfk �r� Y nX tjdt|�|j� d S ) Nz.result handler got EOFError/OSError -- exitingz,result handler found thread._state=TERMINATEzresult handler got sentinelz&result handler ignoring extra sentinelr3 z"ensuring that outqueue is not full� z7result handler exiting: len(cache)=%s, thread._state=%s)r] r� r5 r4 r r6 rQ r� r. r� r� r1 ro r3 �pollrp )r; r0 r� r� r> r? r@ �objr r r rf � s\ zPool._handle_resultsc c s4 t |�}x&ttj||��}|s"d S | |fV q W d S )N)r� �tupler �islice)rA �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 jd� | jtkr"t| _t| j_d S )Nz closing pool)r r6 rQ rP �CLOSEr_ )r r r r r2 s z Pool.closec C s$ t jd� t| _t| j_| j� d S )Nzterminating pool)r r6 r� rQ r_ rk )r r r r � terminate s zPool.terminatec C sV t jd� | jttfks t�| jj� | jj� | j j� x| j D ] }|j� qBW d S )Nz joining pool) r r6 rQ r� r� r. r_ r re rh r[ )r r� r r r r s z Pool.joinc C sD t jd� | jj� x*|j� r>| jj� r>| jj� tj d� qW d S )Nz7removing tasks from inqueue until task handler finishedr ) r r6 Z_rlock�acquire�is_aliver3 r� rz r� r� )r: � task_handlerr� r r r �_help_stuff_finish( s zPool._help_stuff_finishc C sF t jd� t|_t|_t jd� | j||t|�� |j� sJt|�dksJt�t|_|jd � t jd� t j � |k rx|j � |r�t |d d�r�t jd� x |D ]} | j d kr�| j� q�W t jd� t j � |k r�|j � t jd� t j � |k r�|j � |�rBt |d d��rBt jd � x0|D ](} | j� �rt jd | j � | j � �qW d S ) Nzfinalizing poolz&helping task handler/workers to finishr zjoining worker handlerr� zterminating workerszjoining task handlerzjoining result handlerzjoining pool workerszcleaning up worker %d)r r6 r� rQ r� rp r� r. r/ r] r� r r1 rq r� �pid) �clsr� r: r; r� Zworker_handlerr� Zresult_handlerr� r� r r r ri 1 s8 zPool._terminate_poolc C s | S )Nr )r r r r � __enter___ s zPool.__enter__c C s | j � d S )N)r� )r �exc_typeZexc_valZexc_tbr r r �__exit__b s z Pool.__exit__)N)N)NNN)r )r )NNN)NNN)#r r r r* rs rH r rr r\ rx rK r| r r r� r� r� r� r{ r� r} � staticmethodr^ rb rf r� r" r2 r� r r� � classmethodri r� r� r r r r r � sF 8 . < .c @ s@ e Zd Zdd� Zdd� Zdd� Zddd �Zdd d �Zd d � ZdS )r� c C s4 t j� | _tt�| _|| _|| _|| _| || j< d S )N) r] ZEvent�_event�next� job_counterr� rO � _callback�_error_callback)r r� r� r� r r r r k s zApplyResult.__init__c C s | j j� S )N)r� Zis_set)r r r r �readys s zApplyResult.readyc C s | j � s t�| jS )N)r� r. �_success)r r r r � successfulv s zApplyResult.successfulNc C s | j j|� d S )N)r� �wait)r �timeoutr r r r� z s zApplyResult.waitc C s, | j |� | j� st�| jr"| jS | j�d S )N)r� r� r r� �_value)r r� r r r r0 } s zApplyResult.getc C sV |\| _ | _| jr$| j r$| j| j� | jr>| j r>| j| j� | jj� | j| j= d S )N)r� r� r� r� r� �setrO r� )r r@ r� r r r r� � s zApplyResult._set)N)N) r r r r r� r� r� r0 r� r r r r r� i s r� c @ s e Zd Zdd� Zdd� ZdS )r� c C sf t j| |||d� d| _d g| | _|| _|dkrLd| _| jj� || 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= | jj� nH| r�| jr�d| _|| _| j dkr�| j r�| j | j� | j| j= | jj� d S )Nr r F) r� r� r� r� r� rO r� r� r� r� )r r@ Zsuccess_result�successrC 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 jt j� �| _tt�| _|| _tj � | _ d| _ d | _ i | _ | || j< d S )Nr )r] Z ConditionZLock�_condr� r� r� rO � 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| jj� }W nh tk r~ | j| jkr6t�| j j|� y| jj� }W n( tk rx | j| jkrpt�t�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| jj|� | jd7 _x8| j| jkrd| jj| j�}| jj|� | jd7 _q.W | j j� n || j|< | j| jkr�| j| j = W d Q R X d S )Nr ) r� r� r� rt r� �pop�notifyr� rO r� )r r@ r� r r r r� � s zIMapIterator._setc C s< | j �, || _| j| jkr.| j j� | j| j= W d Q R X d S )N)r� r� r� r� rO 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 �@ | jj|� | jd7 _| j j� | j| jkrB| j| j= W d Q R X d S )Nr )r� r� rt r� r� r� rO r� )r r@ r� r r r r� s zIMapUnorderedIterator._setN)r r r r� r r r r r� � s r� c @ s@ e Zd ZdZedd� �Zddf fdd�Zdd� Zed d � �ZdS ) r Fc O s ddl m} || |�S )Nr )rH )ZdummyrH )r rB rH r r r rH s zThreadPool.ProcessNc C s t j| |||� d S )N)r r )r rl r< r= r r r r s zThreadPool.__init__c C s, t j� | _t j� | _| jj| _| jj| _d S )N)rL rM rj rd r/ rc r0 rg )r r r r rK s zThreadPool._setup_queuesc C s< | j �, | jj� | jjd g| � | j j� W d Q R X d S )N)Z not_emptyrL �clear�extendZ notify_all)r: r� r� r r r r� s z ThreadPool._help_stuff_finish) r r r rs r� rH r rK r� r r r r r s )"�__all__r] rL r r� rU r� r r r r r rP r� r� �countr� r r r7 r r r! r$ rF r8 �objectr r� Z AsyncResultr� r� r� r r r r r �<module> s@ * \&)@
Save File
Cancel