ž
ÿ 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..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 Î s uD _queue_management_worker..shutdown_worker..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..shutdown_workerc S s g | ] } | j ‘ q S( ( u sentinel( u .0u p( ( u? /opt/alt/python33/lib64/python3.3/concurrent/futures/process.pyu
ß s u, _queue_management_worker..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 .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
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 , s4
% o h