U /en@sddlmZddlmZddlZddlmZddlm Z ddl m Z ddl m Z ddlmZmZe jdd Ze jd d ZGd d d ejZdS)) annotationsN)merge)gen)parse_timedelta)time) TimeoutErrorsyncc cs|jddV}|j}|dkr(||n`|dkrB||nFz|\}}}||Wn,tk r}z||W5d}~XYnXdS)z[ Coroutine that waits on Dask future, then transmits its outcome to cf_future. F)Zraiseitfinished cancelledN)Z_resultstatusZ set_resultcancelZset_running_or_notify_cancelwith_traceback BaseExceptionZ set_exception)future cf_futureresultr typexctbr:/tmp/pip-unpacked-wheel-g426oqom/distributed/cfexecutor.py_cascade_futures   rc cs.|D]$}z |VWqtk r&YqXqdSN) Exception)futuresZfutrrr_wait_on_futures%s  rc@sLeZdZdZedddddgZddZd d Zd d Zd dZ dddZ dS)ClientExecutorzY A concurrent.futures Executor that executes tasks on a dask.distributed Client. Zpureworkers resourcesZallow_other_workersretriescKsHt|}||jks(tdt||j||_t|_d|_||_ dS)Nz+unsupported arguments to ClientExecutor: %sF) set_allowed_kwargs TypeErrorsorted_clientweakrefWeakSet_futures _shutdown_kwargs)selfclientkwargsskrrr__init__7s   zClientExecutor.__init__cs4t}fdd}|||jjt||S)zK Wrap a distributed Future in a concurrent.futures Future. cs|rjdkrdS)Nr )r r r )rrrr cf_callbackJsz0ClientExecutor._wrap_future..cf_callback)cfZFutureZadd_done_callbackr$loopZ add_callbackr)r*rrr0rr/r _wrap_futureCs   zClientExecutor._wrap_futurecOs@|jrtd|jj|f|t|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) r( RuntimeErrorr$submitrr)r'addr3)r*fnargsr,rrrrr5Ss  zClientExecutor.submitcs~|dddk r&ttd|kr4|d=|rHtdt|jj|f|jtfdd}|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(): %sc 3sz`D]V}j|dk rRz|tVWq\tk rNt jYq\Xq|VqW5t}j|j|XdSr) listr'updater$r r6rrrr1) remainingrZend_timefsr*r9rrresult_iterators  z+ClientExecutor.map..result_iterator) poprrr"r#r$mapr)iter)r*r7 iterablesr,r@rr>rrBcs  zClientExecutor.mapTcCs<|js8d|_t|j}|r,t|jjt|n |j|dS)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)r(r;r'rr$r2rr )r*waitr?rrrshutdowns  zClientExecutor.shutdownN)T) __name__ __module__ __qualname____doc__ frozensetr!r.r3r5rBrFrrrrr.s  s