← Back
Editing: process.cpython-33.pyc
� �� f�C c @ sl d Z d Z d d l Z d d l Z d d l m Z d d l Z d d l Z d d l m Z m Z d d l m Z d d l Z d d l Z e j � Z d a d d � Z d Z Gd d � d e � Z Gd d � d e � Z Gd d � d e � Z d d � Z d d � Z d d � Z d a d a d d � Z Gd d � d e � Z! Gd d � d e j" � Z# e j$ e � d S( u+ Implements ProcessPoolExecutor. The follow diagram and text describe the data-flow through the system: |======================= In-process =====================|== Out-of-process ==| +----------+ +----------+ +--------+ +-----------+ +---------+ | | => | Work Ids | => | | => | Call Q | => | | | | +----------+ | | +-----------+ | | | | | ... | | | | ... | | | | | | 6 | | | | 5, call() | | | | | | 7 | | | | ... | | | | Process | | ... | | Local | +-----------+ | Process | | Pool | +----------+ | Worker | | #1..n | | Executor | | Thread | | | | | +----------- + | | +-----------+ | | | | <=> | Work Items | <=> | | <= | Result Q | <= | | | | +------------+ | | +-----------+ | | | | | 6: call() | | | | ... | | | | | | future | | | | 4, result | | | | | | ... | | | | 3, except | | | +----------+ +------------+ +--------+ +-----------+ +---------+ Executor.submit() called: - creates a uniquely numbered _WorkItem and adds it to the "Work Items" dict - adds the id of the _WorkItem to the "Work Ids" queue Local worker thread: - reads work ids from the "Work Ids" queue and looks up the corresponding WorkItem from the "Work Items" dict: if the work item has been cancelled then it is simply removed from the dict, otherwise it is repackaged as a _CallItem and put in the "Call Q". New _CallItems are put in the "Call Q" until "Call Q" is full. NOTE: the size of the "Call Q" is kept small because calls placed in the "Call Q" can no longer be cancelled with Future.cancel(). - reads _ResultItems from "Result Q", updates the future stored in the "Work Items" dict and deletes the dict entry Process #1..n: - reads _CallItems from "Call Q", executes the calls, and puts the resulting _ResultItems in "Request Q" u" Brian Quinlan (brian@sweetapp.com)i N( u _base( u SimpleQueueu Full( u waitc C sa d a t t j � � } x! | D] \ } } | j d � q Wx | D] \ } } | j � qC Wd S( NT( u Trueu _shutdownu listu _threads_queuesu itemsu putu Noneu join( u itemsu tu q( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu _python_exitK s u _python_exiti c B s | Ee Z d Z d d � Z d S( u _WorkItemc C s( | | _ | | _ | | _ | | _ d S( N( u futureu fnu argsu kwargs( u selfu futureu fnu argsu kwargs( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu __init__[ s u _WorkItem.__init__N( u __name__u __module__u __qualname__u __init__( u __locals__( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu _WorkItemZ s u _WorkItemc B s& | Ee Z d Z d d d d � Z d S( u _ResultItemc C s | | _ | | _ | | _ d S( N( u work_idu exceptionu result( u selfu work_idu exceptionu result( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu __init__b s u _ResultItem.__init__N( u __name__u __module__u __qualname__u Noneu __init__( u __locals__( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu _ResultItema s u _ResultItemc B s | Ee Z d Z d d � Z d S( u _CallItemc C s( | | _ | | _ | | _ | | _ d S( N( u work_idu fnu argsu kwargs( u selfu work_idu fnu argsu kwargs( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu __init__h s u _CallItem.__init__N( u __name__u __module__u __qualname__u __init__( u __locals__( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu _CallItemg s u _CallItemc C s� x� | j d d � } | d k r8 | j t j � � d Sy | j | j | j � } WnA t k r� } z! | j t | j d | �� WYd d } ~ Xq X| j t | j d | �� q d S( u� Evaluates calls from call_queue and places the results in result_queue. This worker is run in a separate process. Args: call_queue: A multiprocessing.Queue of _CallItems that will be read and evaluated by the worker. result_queue: A multiprocessing.Queue of _ResultItems that will written to by the worker. shutdown: A multiprocessing.Event that will be set as a signal to the worker that it should exit when call_queue is empty. u blockNu exceptionu resultT( u getu Trueu Noneu putu osu getpidu fnu argsu kwargsu BaseExceptionu _ResultItemu work_id( u call_queueu result_queueu call_itemu ru e( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu _process_workern s u _process_workerc C s� x� | j � r d Sy | j d d � } Wn t j k rD d SYq X| | } | j j � r� | j t | | j | j | j � d d �q | | =q q d S( uM Fills call_queue with _WorkItems from pending_work_items. This function never blocks. Args: pending_work_items: A dict mapping work ids to _WorkItems e.g. {5: <_WorkItem...>, 6: <_WorkItem...>, ...} work_ids: A queue.Queue of work ids e.g. Queue([5, 6, ...]). Work ids are consumed and the corresponding _WorkItems from pending_work_items are transformed into _CallItems and put in call_queue. call_queue: A multiprocessing.Queue that will be filled with _CallItems derived from _WorkItems. Nu blockFT( u fullu getu Falseu queueu Emptyu futureu set_running_or_notify_cancelu putu _CallItemu fnu argsu kwargsu True( u pending_work_itemsu work_idsu call_queueu work_idu work_item( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu _add_call_item_to_queue� s u _add_call_item_to_queuec s1 d � � f d d � } � � f d d � } | j } x�t | | � � d d � � j � D� } | sn t � t | g | � } | | k r� | j � } n� | � � � d k r� d � _ d � _ d � n x0 | j � D]" \ } } | j j t d � � q� W| j � x � j � D] } | j � qW| � d St | t � r�| � sSt � � j | � } | j � � s�| � d Sne | d k r�| j | j d � } | d k r�| j r�| j j | j � q�| j j | j � q�n | � � | � r$y | s | � d SWq$t k r Yq$Xn d � q9 d S( u� Manages the communication between this process and the worker processes. This function is run in a local thread. Args: executor_reference: A weakref.ref to the ProcessPoolExecutor that owns this thread. Used to determine if the ProcessPoolExecutor has been garbage collected and that this function can exit. process: A list of the multiprocessing.Process instances used as workers. pending_work_items: A dict mapping work ids to _WorkItems e.g. {5: <_WorkItem...>, 6: <_WorkItem...>, ...} work_ids_queue: A queue.Queue of work ids e.g. Queue([5, 6, ...]). call_queue: A multiprocessing.Queue that will be filled with _CallItems derived from _WorkItems for processing by the process workers. result_queue: A multiprocessing.Queue of _ResultItems generated by the process workers. c s t p � d k p � j S( N( u _shutdownu Noneu _shutdown_thread( ( u executor( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu shutting_down� s u/ _queue_management_worker.<locals>.shutting_downc su t d d � � j � D� � } x$ t d | � D] } � j d � q/ W� j � x � j � D] } | j � q] Wd S( Nc s s | ] } | j � Vq d S( N( u is_alive( u .0u p( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu <genexpr>� s uD _queue_management_worker.<locals>.shutdown_worker.<locals>.<genexpr>i ( u sumu valuesu rangeu put_nowaitu Noneu closeu join( u nb_children_aliveu iu p( u call_queueu processes( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu shutdown_worker� s u1 _queue_management_worker.<locals>.shutdown_workerc S s g | ] } | j � q S( ( u sentinel( u .0u p( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu <listcomp>� s u, _queue_management_worker.<locals>.<listcomp>u^ A process in the process pool was terminated abruptly while the future was running or pending.NT( u Noneu _readeru _add_call_item_to_queueu valuesu AssertionErroru waitu recvu Trueu _brokenu _shutdown_threadu itemsu futureu set_exceptionu BrokenProcessPoolu clearu terminateu isinstanceu intu popu joinu work_idu exceptionu set_resultu resultu Full( u executor_referenceu processesu pending_work_itemsu work_ids_queueu call_queueu result_queueu shutting_downu shutdown_workeru readeru sentinelsu readyu result_itemu work_idu work_itemu p( ( u call_queueu executoru processesu? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu _queue_management_worker� sb u _queue_management_workerc C s� t r t r t t � � q n d a y t j d � } Wn t t f k rU d SYn X| d k rf d S| d k rv d Sd | a t t � � d S( Nu SC_SEM_NSEMS_MAXi i u@ system provides too few semaphores (%d available, 256 necessary)Ti����( u _system_limits_checkedu _system_limitedu NotImplementedErroru Trueu osu sysconfu AttributeErroru ValueError( u nsems_max( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu _check_system_limits s u _check_system_limitsc B s | Ee Z d Z d Z d S( u BrokenProcessPooluy Raised when a process in a ProcessPoolExecutor terminated abruptly while a future was in the running state. N( u __name__u __module__u __qualname__u __doc__( u __locals__( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu BrokenProcessPool7 s u BrokenProcessPoolc B sz | Ee Z d Z d d d � Z d d � Z d d � Z d d � Z e j j j e _ d d d � Z e j j j e _ d S( u ProcessPoolExecutorc C s� t � | d k r% t j � | _ n | | _ t j | j t � | _ d | j _ t � | _ t j � | _ d | _ i | _ d | _ t j � | _ d | _ d | _ i | _ d S( u/ Initializes a new ProcessPoolExecutor instance. Args: max_workers: The maximum number of processes that can be used to execute the given calls. If None or not given then as many worker processes will be created as the machine has processors. i NTF( u _check_system_limitsu Noneu multiprocessingu cpu_countu _max_workersu Queueu EXTRA_QUEUED_CALLSu _call_queueu Trueu _ignore_epipeu SimpleQueueu _result_queueu queueu _work_idsu _queue_management_threadu _processesu Falseu _shutdown_threadu threadingu Locku _shutdown_locku _brokenu _queue_countu _pending_work_items( u selfu max_workers( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu __init__? s u ProcessPoolExecutor.__init__c C s� | j d d � } | j d k r� | j � t j d t d t j | | � | j | j | j | j | j f � | _ d | j _ | j j � | j t | j <n d S( Nc S s | j d � d S( N( u putu None( u _u q( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu weakref_cbg s uF ProcessPoolExecutor._start_queue_management_thread.<locals>.weakref_cbu targetu argsT( u _result_queueu _queue_management_threadu Noneu _adjust_process_countu threadingu Threadu _queue_management_workeru weakrefu refu _processesu _pending_work_itemsu _work_idsu _call_queueu Trueu daemonu startu _threads_queues( u selfu weakref_cb( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu _start_queue_management_threadd s u2 ProcessPoolExecutor._start_queue_management_threadc C sh xa t t | j � | j � D]D } t j d t d | j | j f � } | j � | | j | j <q Wd S( Nu targetu args( u rangeu lenu _processesu _max_workersu multiprocessingu Processu _process_workeru _call_queueu _result_queueu startu pid( u selfu _u p( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu _adjust_process_countx s " u) ProcessPoolExecutor._adjust_process_countc O s� | j �� | j r" t d � � n | j r: t d � � n t j � } t | | | | � } | | j | j <| j j | j � | j d 7_ | j j d � | j � | SWd QXd S( NuK A child process terminated abruptly, the process pool is not usable anymoreu* cannot schedule new futures after shutdowni ( u _shutdown_locku _brokenu BrokenProcessPoolu _shutdown_threadu RuntimeErroru _baseu Futureu _WorkItemu _pending_work_itemsu _queue_countu _work_idsu putu _result_queueu Noneu _start_queue_management_thread( u selfu fnu argsu kwargsu fu w( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu submit� s u ProcessPoolExecutor.submitc C ss | j � d | _ Wd QX| j rK | j j d � | rK | j j � qK n d | _ d | _ d | _ d | _ d S( NT( u _shutdown_locku Trueu _shutdown_threadu _queue_management_threadu _result_queueu putu Noneu joinu _call_queueu _processes( u selfu wait( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu shutdown� s u ProcessPoolExecutor.shutdownNT( u __name__u __module__u __qualname__u Noneu __init__u _start_queue_management_threadu _adjust_process_countu submitu _baseu Executoru __doc__u Trueu shutdown( u __locals__( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu ProcessPoolExecutor> s % u ProcessPoolExecutorF(% u __doc__u __author__u atexitu osu concurrent.futuresu _baseu queueu multiprocessingu multiprocessing.queuesu SimpleQueueu Fullu multiprocessing.connectionu waitu threadingu weakrefu WeakKeyDictionaryu _threads_queuesu Falseu _shutdownu _python_exitu EXTRA_QUEUED_CALLSu objectu _WorkItemu _ResultItemu _CallItemu _process_workeru _add_call_item_to_queueu _queue_management_workeru _system_limits_checkedu Noneu _system_limitedu _check_system_limitsu RuntimeErroru BrokenProcessPoolu Executoru ProcessPoolExecutoru register( ( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu <module>, s4 % o h
Save File
Cancel