ó ÕÜ-]c@s!dZddlZddlmZddlZddlZddlZddlZddl Z dZ ej ƒZ e ad„ZdZdefd„ƒYZd efd „ƒYZd efd „ƒYZd „Zd„Zd„Ze adad„Zdejfd„ƒYZejeƒdS(s+ 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" iÿÿÿÿN(t_bases"Brian Quinlan (brian@sweetapp.com)cCsstatrttjƒƒnd}x!|D]\}}|jdƒq+Wx$|D]\}}|jtj ƒqOWdS(N(( tTruet _shutdownt_threads_queuestlisttitemstputtNonetjointsystmaxint(Rtttq((s9lib/python2.7/site-packages/concurrent/futures/process.pyt _python_exitIs it _WorkItemcBseZd„ZRS(cCs(||_||_||_||_dS(N(tfuturetfntargstkwargs(tselfRRRR((s9lib/python2.7/site-packages/concurrent/futures/process.pyt__init__Ys   (t__name__t __module__R(((s9lib/python2.7/site-packages/concurrent/futures/process.pyRXst _ResultItemcBseZddd„ZRS(cCs||_||_||_dS(N(twork_idt exceptiontresult(RRRR((s9lib/python2.7/site-packages/concurrent/futures/process.pyR`s  N(RRRR(((s9lib/python2.7/site-packages/concurrent/futures/process.pyR_st _CallItemcBseZd„ZRS(cCs(||_||_||_||_dS(N(RRRR(RRRRR((s9lib/python2.7/site-packages/concurrent/futures/process.pyRfs   (RRR(((s9lib/python2.7/site-packages/concurrent/futures/process.pyRescCs®x§tr©|jdtƒ}|dkr8|jdƒdSy|j|j|jŽ}Wn3tjƒd}|jt |j d|ƒƒqX|jt |j d|ƒƒqWdS(sø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. tblockNiRR( RtgetRRRRRR texc_infoRR(t call_queuet result_queuet call_itemtrte((s9lib/python2.7/site-packages/concurrent/futures/process.pyt_process_workerls   cCs x™tr›|jƒrdSy|jdtƒ}Wntjk rFdSX||}|jjƒrŽ|jt ||j |j |j ƒdtƒq||=qqWdS(sMFills 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. NR( RtfullRtFalsetqueuetEmptyRtset_running_or_notify_cancelRRRRR(tpending_work_itemstwork_idsRRt work_item((s9lib/python2.7/site-packages/concurrent/futures/process.pyt_add_call_item_to_queue‰s       c s&dg‰‡‡fd†}xtr!t||ˆƒ|jdtƒ}|dk r¡||j}||j=|jrˆ|jj|jƒn|jj|j ƒ~n|ƒ} t sÅ| dksÅ| j r|sx!ˆdt |ƒkrî|ƒqÎWx|D]} | j ƒqöWˆjƒdSn~ qWdS(s‘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. ics!ˆjdƒˆdcd7RERlRFtregister(((s9lib/python2.7/site-packages/concurrent/futures/process.pyt,s.          % B a