U /eI @sdZddlmZddlZddlmZmZmZddlm Z m Z ddl m Z ddl mZmZddlmZdd lmZmZdd lmZmZmZmZmZdd lmZejd krd dZnddZd6ddZddZ ddZ!d7ddZ"de"fddZ#ddZ$ddZ%dd Z&d8d!d"Z'd#d$Z(de%de&e'de(e(df d%d&Z)Gd'd(d(e Z*e*Z+d)d*d+d,d-Z,Gd.d/d/e Z-d0d1Z.d2d3Z/d4d5Z0dS)9aQ Asynchronous Shared-Memory Scheduler for Dask Graphs. This scheduler coordinates several workers to execute tasks in a dask graph in parallel. It depends on a ``concurrent.futures.Executor`` and a corresponding Queue for worker-to-scheduler communication. It tries to execute tasks in an order which maintains a small memory footprint throughout execution. It does this by running tasks that allow us to release data resources. Task Selection Policy ===================== When we complete a task we add more data in to our set of available data; this new data makes new tasks available. We preferentially choose tasks that were just made available in a last-in-first-out fashion. We implement this as a simple stack. This results in more depth-first rather than breadth first behavior which encourages us to process batches of data to completion before starting in on new data when possible. When the addition of new data readies multiple tasks simultaneously we add tasks to the stack in sorted order so that tasks with greater keynames are run first. This can be handy to break ties in a predictable fashion. State ===== Many functions pass around a ``state`` variable that holds the current state of the computation. This variable consists of several other dictionaries and sets, explained below. Constant state -------------- 1. dependencies: {x: [a, b ,c]} a,b,c, must be run before x 2. dependents: {a: [x, y]} a must run before x or y Changing state -------------- ### Data 1. cache: available concrete data. {key: actual-data} 2. released: data that we've seen, used, and released because it is no longer needed ### Jobs 1. ready: A fifo stack of ready-to-run tasks 2. running: A set of tasks currently in execution 3. finished: A set of finished tasks 4. waiting: which tasks are still waiting on others :: {key: {keys}} Real-time equivalent of dependencies 5. waiting_data: available data to yet-to-be-run-tasks :: {key: {keys}} Real-time equivalent of dependents Examples -------- >>> import pprint # doctest: +SKIP >>> inc = lambda x: x + 1 >>> add = lambda x, y: x + y >>> dsk = {'x': 1, 'y': 2, 'z': (inc, 'x'), 'w': (add, 'z', 'y')} # doctest: +SKIP >>> pprint.pprint(start_state_from_dask(dsk)) # doctest: +SKIP {'cache': {'x': 1, 'y': 2}, 'dependencies': {'w': {'z', 'y'}, 'x': set(), 'y': set(), 'z': {'x'}}, 'dependents': defaultdict(None, {'w': set(), 'x': {'z'}, 'y': {'w'}, 'z': {'w'}}), 'finished': set(), 'ready': ['z'], 'released': set(), 'running': set(), 'waiting': {'w': {'z'}}, 'waiting_data': {'x': {'z'}, 'y': {'w'}, 'z': {'w'}}} Optimizations ============= We build this scheduler with out-of-core array operations in mind. To this end we have encoded some particular optimizations. Compute to release data ----------------------- When we choose a new task to execute we often have many options. Policies at this stage are cheap and can significantly impact performance. One could imagine policies that expose parallelism, drive towards a particular output, etc.. Our current policy is to run tasks that were most recently made available. Inlining computations --------------------- We hold on to intermediate computations either in memory or on disk. For very cheap computations that may emit new copies of the data, like ``np.transpose`` or possibly even ``x + 1`` we choose not to store these as separate pieces of data / tasks. Instead we combine them with the computations that require them. This may result in repeated computation but saves significantly on space and computation complexity. See the function ``inline_functions`` for more information. ) annotationsN)HashableMappingSequence)ExecutorFuture)partial)EmptyQueue)config)local_callbacksunpack_callbacks) _execute_taskflattenget_dependencies has_tasks reverse_dict)orderntcCs,z|jdddWStk r$YqXqdS)NTg?)blocktimeout)getr qr./tmp/pip-unpacked-wheel-dbjnr7gq/dask/local.py queue_getsrcCs|SN)rrrrrrsc s:|dkrt|j}|dkr&tdd}|dkr4t}t|D]$\}}t||sB|||<|qB| |fdd|D}fdd|D}t |}|D]$}||dD]} ||  |qqdd|D} dd |D} t | |d d } d d|D}|||| || tttd } | S)aStart state from a dask Examples -------- >>> inc = lambda x: x + 1 >>> add = lambda x, y: x + y >>> dsk = {'x': 1, 'y': 2, 'z': (inc, 'x'), 'w': (add, 'z', 'y')} # doctest: +SKIP >>> from pprint import pprint # doctest: +SKIP >>> pprint(start_state_from_dask(dsk)) # doctest: +SKIP {'cache': {'x': 1, 'y': 2}, 'dependencies': {'w': {'z', 'y'}, 'x': set(), 'y': set(), 'z': {'x'}}, 'dependents': defaultdict(None, {'w': set(), 'x': {'z'}, 'y': {'w'}, 'z': {'w'}}), 'finished': set(), 'ready': ['z'], 'released': set(), 'running': set(), 'waiting': {'w': {'z'}}, 'waiting_data': {'x': {'z'}, 'y': {'w'}, 'z': {'w'}}} Ncachecsi|]}|t|qSr)r).0k)dsk2rr sz)start_state_from_dask..cs"i|]\}}|kr||qSrcopyrr v) data_keysrrr"srcSsi|]\}}|r||qSrr#r%rrrr"scSsh|]\}}|s|qSrrr%rrr sz(start_state_from_dask..TkeyreversecSsi|]\}}|r||qSrrr%rrrr"s) dependencies dependentswaiting waiting_datarreadyrunningfinishedreleased) rrr dictsetitemsraddr$updaterremovesorted)dskrsortkeyr r&r,r.r-abr/Z ready_setr0stater)r'r!rstart_state_from_dasksD      r@c Cslz0||\}}t||}|} ||| f}d} Wn0tk r`} z|| |}d} W5d} ~ XYnX||| fS)zy Compute task and handle all administration See Also -------- _execute_task : actually execute task FTN)r BaseException) r*Z task_infodumpsloadsget_idpack_exceptiontaskdataresultidfailederrr execute_tasks    rLcCsdd|DS)z? Batch computing of multiple tasks with `execute_task` cSsg|] }t|qSr)rL)rr=rrr sz'batch_execute_tasks..r)itrrrbatch_execute_taskssrOTcCsF||dkr&|d|rt|d|=|d||rB|d|=dS)zRRemove data from temporary storage See Also -------- finish_task r/r3rN)AssertionErrorr7)r*r?deleterrr release_datas   rRc Cst|d||ddD]6}|d|}|||s|d|=|d|q|d|D]\}||dkr|d|}|||s||kr||||dqZ|rZ||krZ||||dqZ|d ||d ||S) zo Update execution state after a task finishes Mutates. This should run atomically (with a lock). r-Tr)r.r0r,r/)rQr2r1)r:r9appendr7) r;r*r?resultsr<rQrRdepsrrr finish_tasks"        rWcs,t|tr tfdd|DS|SdS)zGet nested index from collection Examples -------- >>> nested_get(1, 'abc') 'b' >>> nested_get([1, 0], 'abc') ('b', 'a') >>> nested_get([[1, 0], [0, 1]], 'abc') (('b', 'a'), ('a', 'b')) c3s|]}t|VqdSr) nested_get)ricollrr .sznested_get..N) isinstancelisttuple)indr[rrZrrX s rXcCsdS)zDefault get_idNrrrrrdefault_get_id3sracCsdSrr)rKrBrrrdefault_pack_exception8srbcCs|j|k r|||dSr) __traceback__with_traceback)exctbrrrreraise<s  rgcCs|S)z=Identity function. Returns x. >>> identity(3) 3 r)xrrridentityBsric # sF| ptdd} tt|tr.tt|}n|h}t|}tt| } t | \}}}}g}d}iz| D]"}|dr|d| |qvt }t ||jd| D]\}}}}}|r|q|dkrtdd}drd st d  f d d }ds@d s@d r|| tD]\}}}|r|\}}|rfddt|D}|}t||n ||||\} }!| d|<t|||j|D]}"|"|| |!qڐqTq"d}W5|D]$\}}}}}|r|| qXW5QRXt|dS)adAsynchronous get function This is a general version of various asynchronous schedulers for dask. It takes a ``concurrent.futures.Executor.submit`` function to form a more specific ``get`` method that walks through the dask array with parallel workers, avoiding repeat computation and minimizing memory use. Parameters ---------- submit : function A ``concurrent.futures.Executor.submit`` function num_workers : int The number of workers that task submissions can be spread over dsk : dict A dask dictionary specifying a workflow result : key or list of keys Keys corresponding to desired data cache : dict-like, optional Temporary storage of results get_id : callable, optional Function to return the worker id, takes no arguments. Examples are `threading.current_thread` and `multiprocessing.current_process`. rerun_exceptions_locally : bool, optional Whether to rerun failing tasks in local process to enable debugging (False by default) pack_exception : callable, optional Function to take an exception and ``dumps`` method, and return a serialized tuple of ``(exception, traceback)`` to send back to the scheduler. Default is to just raise the exception. raise_exception : callable, optional Function that takes an exception and a traceback, and raises an error. callbacks : tuple or list of tuples, optional Callbacks are passed in as tuples of length 5. Multiple sets of callbacks may be passed in as a list of tuples. For more information, see the dask.diagnostics documentation. dumps: callable, optional Function to serialize task data and results to communicate between worker and parent. Defaults to identity. loads: callable, optional Inverse function of `dumps`. Defaults to identity. chunksize: int, optional Size of chunks to use when dispatching work. Defaults to 1. If -1, will be computed to evenly divide ready work across workers. See Also -------- threaded.get chunksizeFr)rr<Nrerun_exceptions_locallyr.r0z Found no accessible jobs in daskc s*td}|dkr&|}|  }n0td|  }t|d}t|||}g}t|D]n}d}d|D]}||qfddt|D} |||| ffqbtt||  D]>} || || d|} | sq& t| } | j qdS) z"Fire off a task to the thread poolr0r1rcsi|]}|d|qSrrrrUr?rrr"sz1get_async..fire_tasks..rkN) lenmaxminrangepopr7rrSrOZadd_done_callbackput) rjZnreadyZntasksZ used_workersZ avail_workersargs_r*frGrYZ each_argsfut r;rBrDrC num_workersrEZ pretask_cbsqueuer?submitrr fire_taskss>      zget_async..fire_tasksr1csi|]}|d|qSrnrrorprrr"szget_async..rT)r rr r]r^r5rr4r r rSrr@ ValueErrorrrHrrrWrX)#r~r|r;rHrrDrlrEZraise_exception callbacksrBrCrjkwargsZ result_flatrTrxZ posttask_cbsZ started_cbsZ succeededfinishcbZkeyorderZ start_staterr*Zres_inforJrerfrGrFresZ worker_idryrr{r get_async^sb@      .      rc@seZdZdZddZdS)SynchronousExecutorrkc OsLt}z||||Wn,tk rF}z||W5d}~XYnX|Sr)r set_resultrA set_exception)selffnrwrrzrKrrrr~s zSynchronousExecutor.submitN)__name__ __module__ __qualname__ _max_workersr~rrrrrsrrzSequence[Hashable] | Hashable)r;keyscKs"|ddttjtj||f|S)zOA naive synchronous version of get_async Can be useful for debugging. r|N)rursynchronous_executorr~r)r;rrrrrget_sync's rc@seZdZddZddZdS)MultiprocessingPoolExecutorcCs||_t|j|_dSr)poolrq_poolr)rrrrr__init__Bsz$MultiprocessingPoolExecutor.__init__cOst|jj|f||Sr)submit_apply_asyncr apply_async)rrrwrrrrr~Fsz"MultiprocessingPoolExecutor.submitN)rrrrr~rrrrrAsrcOst}|||||j|j|Sr)rrr)rrrwrrzrrrrJsrcOsttt||f||Sr)rrr)rr|rwrrrrget_apply_asyncPsrcCst|j|fS)aSorting key function that is robust to different types Both strings and tuples are common key types in dask graphs. However In Python 3 one can not compare strings with tuples directly. This function maps many types to a form where they can be compared Examples -------- >>> sortkey('Hello') ('str', 'Hello') >>> sortkey(('x', 1)) ('tuple', ('x', 1)) )typer)itemrrrr<Vsr<)NN)T)N)1__doc__ __future__roscollections.abcrrrconcurrent.futuresrr functoolsrr}r r Zdaskr Zdask.callbacksr r Z dask.corerrrrrZ dask.orderrnamerr@rLrOrRrWrXrarbrgrirrrrrrrr<rrrrsVl       J   ! ;