ó ßüÚ\c@`sÃddlmZmZmZddlZddlmZddlmZm Z ddl m Z m Z ddl mZddlmZejeƒZd efd „ƒYZd efd „ƒYZdS( i(tprint_functiontdivisiontabsolute_importN(tgeni(t futures_oftwait(tsyncttokey(t pack_data(t _deserializetReplayExceptionSchedulercB`s eZdZd„Zd„ZRS(s A plugin for the scheduler to recreate exceptions locally This adds the following routes to the scheduler * cause_of_failure cC`s0||_|j|jjd<||jjdlib/python2.7/site-packages/distributed/recreate_exceptions.pyt__init__s cO`sÂ|jdgƒ}x©|D]¡}t|tƒr=t|ƒ}nt|ƒ}|jjj|ƒ}|dk r|j dk r|j }ig|j D]}|j ^qd6|j d6|j d6SqWdS(sX Return details of first failed task required by set of keys Parameters ---------- keys: list of keys known to the scheduler Returns ------- Dictionary with: cause: the key that failed task: the definition of that key deps: keys that the task depends on tkeystdepstcausettaskN( tpopt isinstancetlistttupleRR ttaskstgettNonetexception_blamet dependenciestkeytrun_spec(RtargstkwargsRRttsRtdts((s>lib/python2.7/site-packages/distributed/recreate_exceptions.pyR s     (t__name__t __module__t__doc__RR (((s>lib/python2.7/site-packages/distributed/recreate_exceptions.pyR s tReplayExceptionClientcB`s\eZdZd„Zed„ƒZejd„ƒZd„Z ejd„ƒZ d„Z RS(sP A plugin for the client allowing replay of remote exceptions locally Adds the following methods (and their async variants)to the given client: - ``recreate_error_locally``: main user method - ``get_futures_error``: gets the task, its details and dependencies, responsible for failure of the given future. cC`sY||_||jjd<|j|j_|j|j_|j|j_|j|j_dS(NR (tclientRtrecreate_error_locallyt_recreate_error_locallyt_get_futures_errortget_futures_error(RR)((s>lib/python2.7/site-packages/distributed/recreate_exceptions.pyREs  cC`s |jjS(N(R)R (R((s>lib/python2.7/site-packages/distributed/recreate_exceptions.pyR Nsc c`sıgt|ƒD]}|jdkr |^q }|sCtdƒ‚n|jjdg|D]}|j^qVƒV}|d|d}}t|tƒrÆt|\}}} t j ||| |fƒ‚n3td|ƒ\}}} t j ||| |fƒ‚dS(NterrorsNo errored futures passedRRR( Rtstatust ValueErrorR R RRtdictR RtReturn( RtfuturetftfuturestoutRRtfunctionR!R"((s>lib/python2.7/site-packages/distributed/recreate_exceptions.pyR,Rs.,cC`s|jj|j|ƒS(sô Ask the scheduler details of the sub-task of the given failed future When a future evaluates to a status of "error", i.e., an exception was raised in a task within its graph, we an get information from the scheduler. This function gets the details of the specific task that raised the exception and led to the error, but does not fetch data from the cluster or execute the function. Parameters ---------- future : future that failed, having ``status=="error"``, typically after an attempt to ``gather()`` shows a stack-stace. Returns ------- Tuple: - the function that raised an exception - argument list (a tuple), may include values and keys - keyword arguments (a dictionary), may include values and keys - list of keys that the function requires to be fetched to run See Also -------- ReplayExceptionClient.recreate_error_locally (R)RR,(RR3((s>lib/python2.7/site-packages/distributed/recreate_exceptions.pyR-asc c`st|ƒV|j|ƒV}|\}}}}|jji|ƒ}|jj|ƒV}t||ƒ}t||ƒ}tj|||fƒ‚dS(N(RR,R)t_graph_to_futurest_gatherRRR2( RR3R6R7R!R"RR5tdata((s>lib/python2.7/site-packages/distributed/recreate_exceptions.pyR+s cC`s5t|jj|j|ƒ\}}}|||dS(s“ 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. N(RR)tloopR+(RR3tfuncR!R"((s>lib/python2.7/site-packages/distributed/recreate_exceptions.pyR*Šs+!( R%R&R'RtpropertyR Rt coroutineR,R-R+R*(((s>lib/python2.7/site-packages/distributed/recreate_exceptions.pyR(:s    (t __future__RRRtloggingttornadoRR)RRtutilsRRt utils_commRtworkerR t getLoggerR%tloggertobjectR R((((s>lib/python2.7/site-packages/distributed/recreate_exceptions.pyts -