0FienddlmZddlmZddlZddlmZddlm Z ddl m Z ddl m Z ddlmZmZe jdZe jd ZGd d ejZdS) ) annotationsN)merge)gen)parse_timedelta)time) TimeoutErrorsyncc#pK|dV}|j}|dkr||dS|dkr*||dS |\}}}||#t $r }||Yd}~dSd}~wwxYw)z[ Coroutine that waits on Dask future, then transmits its outcome to cf_future. F)raiseitfinished cancelledN)_resultstatus set_resultcancelset_running_or_notify_cancelwith_traceback BaseException set_exception)future cf_futureresultrtypexctbs 6lib/python3.11/site-packages/distributed/cfexecutor.py_cascade_futurers >>%>00 0 0F ]F V$$$$$ ;  ..00000 )!LCb$$R(( ( ) ) )  # #C ( ( ( ( ( ( ( ( ( )s0B B5B00B5c#>K|D]} |V#t$rYwxYwdSN) Exception)futuresfuts r_wait_on_futuresr#%sO IIII    D s   cLeZdZdZegdZdZdZdZdZ d dZ d S) ClientExecutorzY A concurrent.futures Executor that executes tasks on a dask.distributed Client. )pureworkers resourcesallow_other_workersretriesc t|}||jks'tdt||jz z||_t j|_d|_||_ dS)Nz+unsupported arguments to ClientExecutor: %sF) set_allowed_kwargs TypeErrorsorted_clientweakrefWeakSet_futures _shutdown_kwargs)selfclientkwargssks r__init__zClientExecutor.__init__7sr [[T)))=d22334  ))  ctj}fd}|||jjt ||S)zK Wrap a distributed Future in a concurrent.futures Future. cv|r!jdkrdSdSdS)Nr )r rr)rrs r cf_callbackz0ClientExecutor._wrap_future..cf_callbackJsD""$$ +)E)E   )E)Er;)cfFutureadd_done_callbackr0loop add_callbackr)r6rrr>s ` r _wrap_futurezClientExecutor._wrap_futureCsbIKK       ##K000 && JJJr;c|jrtd|jj|g|Rit |j|}|j|||S)a/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. z*cannot schedule new futures after shutdown) r4 RuntimeErrorr0submitrr5r3addrD)r6fnargsr8rs rrGzClientExecutor.submitSsx > MKLL L$$RN$NNN% f2M2MNN &!!!  (((r;cB|dd ttzd|vr|d=|rtdt |zjj|g|Rijtfd}|S)a&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 ------ concurrent.futures.TimeoutError: If the entire result iterator could not be generated before the given timeout. Exception: If ``fn(*args)`` raises for any values. timeoutN chunksizez!unexpected arguments to map(): %sc3K D]v}j|B |tz VF#t$r t jwxYw|Vw t }j|j |dS#t }j|j |wxYwr) r3rHrrrr?listupdater0r)r remainingend_timefsr6rLs rresult_iteratorz+ClientExecutor.map..result_iterators  / ..FM%%f---*2"(--4660A"B"BBBBB+222"$/12%mmoo----.!HH  $$Y/// ##I.....!HH  $$Y/// ##I....s# C&A  C A$$CAD ) poprrr.r/r0mapr5iter)r6rI iterablesr8rTrRrSrLs` @@@rrVzClientExecutor.mapcs2**Y--  %g..G'H & {#  R?&..PQQ Q T\ b =9 = = = = ="XX / / / / / / / /    r;Tc|js[d|_t|j}|r"t|jjt |dS|j|dSdS)aClean-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. TN)r4rOr3r r0rBr#r)r6waitrSs rshutdownzClientExecutor.shutdownsp~ (!DNdm$$B (T\&(8"===== ##B'''''  ( (r;N)T) __name__ __module__ __qualname____doc__ frozensetr-r:rDrGrVr[r;rr%r%.s iJJJO    ))) :!:!:!x((((((r;r%) __future__rconcurrent.futuresr!r?r1tlzrtornador dask.utilsrdistributed.metricsrdistributed.utilsrr coroutinerr#Executorr%rar;rrks """"""&&&&&&$$$$$$00000000)))*C(C(C(C(C(R[C(C(C(C(C(r;