σ ίόΪ\c@`sΔddlmZmZmZddljZddlZddlZddl m Z ddl m Z ddl mZddlmZe jd„ƒZe jd „ƒZd ejfd „ƒYZdS( i(tprint_functiontdivisiontabsolute_importN(tmerge(tgeni(ttime(tsynccc`s“|jdtƒV}|j}|dkr8|j|ƒnW|dkr[|jƒ|jƒn4ytj|ŒWn tk rŽ}|j |ƒnXdS(s[ Coroutine that waits on Dask future, then transmits its outcome to cf_future. traiseittfinishedt cancelledN( t_resulttFalsetstatust set_resulttcanceltset_running_or_notify_canceltsixtreraiset BaseExceptiont set_exception(tfuturet cf_futuretresultR texc((s5lib/python2.7/site-packages/distributed/cfexecutor.pyt_cascade_futures     cc`s2x+|D]#}y |VWqtk r)qXqWdS(N(t Exception(tfuturestfut((s5lib/python2.7/site-packages/distributed/cfexecutor.pyt_wait_on_futures%s    tClientExecutorcB`sYeZdZedddddgƒZd„Zd„Zd„Zd „Ze d „Z RS( sY A concurrent.futures Executor that executes tasks on a dask.distributed Client. tpuretworkerst resourcestallow_other_workerstretriescK`sit|ƒ}||jks;tdt||jƒƒ‚n||_tjƒ|_t|_ ||_ dS(Ns+unsupported arguments to ClientExecutor: %s( tsett_allowed_kwargst TypeErrortsortedt_clienttweakreftWeakSett_futuresR t _shutdownt_kwargs(tselftclienttkwargstsk((s5lib/python2.7/site-packages/distributed/cfexecutor.pyt__init__7s   c`sEtjƒ}‡fd†}|j|ƒ|jjjtˆ|ƒ|S(sK Wrap a distributed Future in a concurrent.futures Future. c`s,|jƒr(ˆjdkr(ˆjƒndS(NR (R R R(R(R(s5lib/python2.7/site-packages/distributed/cfexecutor.pyt cf_callbackJs(tcftFuturetadd_done_callbackR'tloopt add_callbackR(R-RRR2((Rs5lib/python2.7/site-packages/distributed/cfexecutor.pyt _wrap_futureCs   cO`sY|jrtdƒ‚n|jj||t|j|ƒŽ}|jj|ƒ|j|ƒS(s/Submits a callable to be executed with the given arguments. Schedules the callable to be executed as ``fn(*args, **kwargs)`` and returns a Future instance representing the execution of the callable. Returns ------- A Future representing the given call. s*cannot schedule new futures after shutdown( R+t RuntimeErrorR'tsubmitRR,R*taddR8(R-tfntargsR/R((s5lib/python2.7/site-packages/distributed/cfexecutor.pyR:Ss $c`s|jddƒ‰ˆdk r.ˆtƒ‰nd|krD|d=n|rctdt|ƒƒ‚nˆjj||ˆjŽ‰‡‡‡‡fd†}|ƒS(sχReturns an iterator equivalent to ``map(fn, *iterables)``. Parameters ---------- fn: A callable that will take as many arguments as there are passed iterables. iterables: One iterable for each parameter to *fn*. timeout: The maximum number of seconds to wait. If None, then there is no limit on the wait time. chunksize: ignored. Returns ------- An iterator equivalent to: ``map(fn, *iterables)`` but the calls may be evaluated out-of-order. Raises ------ TimeoutError: If the entire result iterator could not be generated before the given timeout. Exception: If ``fn(*args)`` raises for any values. ttimeoutt chunksizes!unexpected arguments to map(): %sc3`s½zxxqˆD]i}ˆjj|ƒˆdk rhy|jˆtƒƒVWqstjk rdtj‚qsXq |jƒVq WWdtˆƒ}x|D]}ˆjj|ƒqŽWˆj j |ƒXdS(N( R*R;tNoneRRRt TimeoutErrorR3tlistR'R(Rt remaining(tend_timetfsR-R>(s5lib/python2.7/site-packages/distributed/cfexecutor.pytresult_iterator†s    N(tpopR@RR%R&R'tmapR,(R-R<t iterablesR/RF((RDRER-R>s5lib/python2.7/site-packages/distributed/cfexecutor.pyRHcs   cC`sW|jsSt|_t|jƒ}|r@t|jjt|ƒqS|jj|ƒndS(sšClean-up the resources associated with the Executor. It is safe to call this method several times. Otherwise, no other methods can be called after this one. Parameters ---------- wait: If True then shutdown will not return until all running futures have finished executing. If False then all running futures are cancelled immediately. N( R+tTrueRBR*RR'R6RR(R-twaitRE((s5lib/python2.7/site-packages/distributed/cfexecutor.pytshutdown™s  ( t__name__t __module__t__doc__t frozensetR$R1R8R:RHRJRL(((s5lib/python2.7/site-packages/distributed/cfexecutor.pyR.s   6(t __future__RRRtconcurrent.futuresRR3R(RttoolzRttornadoRtmetricsRtutilsRt coroutineRRtExecutorR(((s5lib/python2.7/site-packages/distributed/cfexecutor.pyts