U /e@s~ddlmZddlZddlmZddlmZmZddlm Z ddl m Z ddl m Z eeZGdd d ZGd d d ZdS) ) annotationsN) stringify) futures_ofwait)sync) pack_data) _deserializec@s<eZdZdZddZddZdddd Zd d d d Zd S)ReplayTaskSchedulerzA plugin for the scheduler to recreate tasks locally This adds the following routes to the scheduler * get_runspec * get_error_cause cCs&||_|j|jjd<|j|jjd<dS)N get_runspecget_error_cause) schedulerr handlersr )selfr r>/tmp/pip-unpacked-wheel-g426oqom/distributed/recreate_tasks.py__init__szReplayTaskScheduler.__init__cCst|trt|}t|}|SN) isinstancelisttupler)rkeyrrr _process_keys z ReplayTaskScheduler._process_keyrkeyscOsD|D]:}||}|jj|}|dk r|jdk r|jjSqdSr)rr tasksgetZexception_blamer)rrargskwargsrtsrrrr #s  z#ReplayTaskScheduler.get_error_causeNrcOs0||}|jj|}|jdd|jDdS)NcSsg|] }|jqSrr).0Zdtsrrr -sz3ReplayTaskScheduler.get_runspec..)taskdeps)rr rrZrun_specZ dependencies)rrrrrrrrr *s zReplayTaskScheduler.get_runspec)__name__ __module__ __qualname____doc__rrr r rrrrr s r c@sTeZdZdZddZeddZddZdd Zd d Z d d Z ddZ ddZ dS)ReplayTaskClienta A plugin for the client allowing replay of remote tasks locally Adds the following methods to the given client: - ``recreate_error_locally``: main user method for replaying failed tasks - ``recreate_task_locally``: main user method for replaying any task cCsR||_||jjd<|j|j_|j|j_|j|j_|j|j_|j|j_|j|j_dS)Nz replay-tasks)client extensions_get_raw_components_from_future_prepare_raw_components_get_components_from_future_get_errored_futurerecreate_task_locallyrecreate_error_locally)rr)rrrr:s     zReplayTaskClient.__init__cCs|jjSr)r)r )rrrrr GszReplayTaskClient.schedulerc st|tr|}nt|IdH|j}|jj|dIdH}|d|d}}t|trptf|\}}}||||fSt|d\}}}||||fSdS)z~ For a given future return the func, args and kwargs and future deps that would be executed remotely. Nrr#r")r")rstrrrr r dictr) rfuturerspecr#r"functionrrrrrr+Ks   z0ReplayTaskClient._get_raw_components_from_futurecsJ|\}}}}|ji|}|j|IdH}t||}t||}|||fS)zF Take raw components and resolve future dependencies. N)r)Z_graph_to_futuresZ_gatherr)rraw_componentsr5rrr#futuresdatarrrr,^s    z(ReplayTaskClient._prepare_raw_componentscs ||IdH}||IdHS)z For a given future return the func, args and kwargs that would be executed remotely. Any args/kwargs that are themselves futures will be resolved to the return value of those futures. N)r+r,)rr3r6rrrr-isz,ReplayTaskClient._get_components_from_futurecCs"t|jj|j|\}}}|||S)aK For any calculation, whether it succeeded or failed, perform the task locally for debugging. This operation should be performed after a future (result of ``gather``, ``compute``, etc) comes back with a status other than "pending". Cases where you might want to debug a successfully completed future could include a calculation that returns an unexpected results. A common debugging process might include running the task locally in debug mode, with `pdb.runcall`. Examples -------- >>> import pdb # doctest: +SKIP >>> future = c.submit(div, 1, 1) # doctest: +SKIP >>> future.status # doctest: +SKIP 'finished' >>> pdb.runcall(c.recreate_task_locally, future) # doctest: +SKIP Parameters ---------- future : future The same thing as was given to ``gather``. Returns ------- Any; will return the result of the task future. )rr)loopr-)rr3funcrrrrrr/rs  z&ReplayTaskClient.recreate_task_locallycsDt|IdHddt|D}|s,td|jj|dIdH}|S)zf For a given future collection, return the first future that raised an error. NcSsg|]}|jdkr|jqS)error)statusr)r frrrr!s z8ReplayTaskClient._get_errored_future..zNo errored futures passedr)rr ValueErrorr r )rr3r7Z cause_keyrrrr.s z$ReplayTaskClient._get_errored_futurecCst|jj|j|}||S)a For a failed calculation, perform the blamed task locally for debugging. This operation should be performed after a future (result of ``gather``, ``compute``, etc) comes back with a status of "error", if the stack- trace is not informative enough to diagnose the problem. The specific task (part of the graph pointing to the future) responsible for the error will be fetched from the scheduler, together with the values of its inputs. The function will then be executed, so that ``pdb`` can be used for debugging. Examples -------- >>> future = c.submit(div, 1, 0) # doctest: +SKIP >>> future.status # doctest: +SKIP 'error' >>> c.recreate_error_locally(future) # doctest: +SKIP ZeroDivisionError: division by zero If you're in IPython you might take this opportunity to use pdb >>> %pdb # doctest: +SKIP Automatic pdb calling has been turned ON >>> c.recreate_error_locally(future) # doctest: +SKIP ZeroDivisionError: division by zero 1 def div(x, y): ----> 2 return x / y ipdb> Parameters ---------- future : future or collection that failed The same thing as was given to ``gather``, but came back with an exception/stack-trace. Can also be a (persisted) dask collection containing any errored futures. Returns ------- Nothing; the function runs and should raise an exception, allowing the debugger to run. )rr)r9r.r/)rr3Zerrored_future_keyrrrr0s+z'ReplayTaskClient.recreate_error_locallyN) r$r%r&r'rpropertyr r+r,r-r/r.r0rrrrr(0s     " r() __future__rloggingZ dask.utilsrZdistributed.clientrrZdistributed.utilsrZdistributed.utils_commrZdistributed.workerr getLoggerr$loggerr r(rrrrs      !