o $iT-@s@dZddlZddlZddlmZmZddlZddlmZz ddlm Z m Z Wne y9e dej dYnwddlmZmZdd lmZmZmZdd lmZejd kr\d d Zndd Zd(ddZddZd)ddZdZdefddZddZddZddZ d*dd Z!d!d"Z"dede e!de"e"fd#d$Z#d+d&d'Z$dS),zu The following is adapted from Dask release 2021.03.1: https://github.com/dask/dask/blob/2021.03.1/dask/local.py N)EmptyQueue)config)DataNodeDependenciesMappingzEDask on Ray is available only on dask>=2024.11.0, you are on version .)local_callbacksunpack_callbacks)flattenget_dependencies reverse_dict)orderntcCs( z|jdddWStyYnwq)NTg?)blocktimeout)getrqrZ/home/ubuntu/veenaModal/venv/lib/python3.10/site-packages/ray/util/dask/scheduler_utils.py queue_gets rcCs|SN)rrrrrr&sc s2|dur t|j}|durtdd}|durt}t|D]\}}t|tr4|||<|q!| }| |t |}fdd|D}t |}|D]} || dD] } ||  | q[qSdd|D} dd|D} t| |d d } d d|D}|||| || tttd }|S) aStart state from a dask Examples -------- >>> 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': {'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'}}} Ncachecs"i|] \}}|vr|t|qSr)set.0kv data_keysrr Ps"z)start_state_from_dask..rcSsi|] \}}|r||qSr)copyrrrrr VscSsh|]\}}|s|qSrrrrrr Xsz(start_state_from_dask..TkeyreversecSsi|] \}}|r||qSrrrrrrr Z) dependencies dependentswaiting waiting_datarreadyrunningfinishedreleased)r rrdictritems isinstanceraddr!updaterr removesorted)dskrsortkeyrrdsk2r'r)r(abr* ready_setr+staterrrstart_state_from_dask*sH       r=c Cslz||\}}||}|} ||| f}d} Wnty0} z || |}d} WYd} ~ nd} ~ ww||| fS)zx Compute task and handle all administration See Also -------- _execute_task : actually execute task FTN) BaseException) r$ task_infodumpsloadsget_idpack_exceptiontaskdataresultidfailederrr execute_taskks    rJTcCsJ||dvr|d|rJ|d|=|d||r#|d|=dSdS)zQRemove data from temporary storage See Also -------- finish_task r*r.rN)r2)r$r<deleterrr release_data~s  rLFc Cst|d||ddD]}|d|}|||s&|d|=|d|q |d|D]J}||dvrj|d|}|||si||vritrbdd lm} td ||tt| |d  d f||||d q-|rw||vrw||||d q-|d ||d||S)zn Update execution state after a task finishes Mutates. This should run atomically (with a lock). r(Tr#r)r+r'r*r)nbytesz&Key: %s Dep: %s NBytes: %.2f Releaserg.A)rKr-r,) r5r4appendDEBUG chest.corerMprintsummapvaluesr2) r6r$r<resultsr7rKrLdepsrMrrr finish_tasks6         rXcs(t|trtfdd|DS|S)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..)r1listtuple)indr\rr[rrYs rYcCsdS)zDefault get_idNrrrrrdefault_get_idsracCsrr)rIr@rrrdefault_pack_exceptionsrbcCs|j|ur |||r) __traceback__with_traceback)exctbrrrreraises  rgcCs|S)z>> identity(3) 3 r)xrrridentitysric " stt|trtt|} n|h} t| }tt| =} t| \}}}}g}d}iz| D]}|dr?|d||q3t }t ||j d| D]\}}}}}|ra|qS|durlt dd}drxdsxt df d d }drtd |kr|drtd |ksdsdsd r&t\}}}|rو|\}}|rԇfd d t|D}|}||n||||\}}|d|<t|||j |D] } | |||qdrtd |kr|drtd |ks dsdsd sd}W|D]\}}}}}!|!r<|!| q+n|D]\}}}}}!|!rR|!| qAwWdn 1s`wYt|dS)aAsynchronous get function This is a general version of various asynchronous schedulers for dask. It takes a an apply_async function as found on Pool objects to form a more specific ``get`` method that walks through the dask array with parallel workers, avoiding repeat computation and minimizing memory use. Parameters ---------- apply_async : function Asynchronous apply function as found on Pool or ThreadPool num_workers : int The number of active tasks we should have at any one time 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. 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. 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. See Also -------- threaded.get Fr)rr7Nrerun_exceptions_locallyr)r+z Found no accessible jobs in daskcstd}d|D]}||qfddt|D}t|||ffjddS)z"Fire off a task to the thread poolr+r,ci|] }|d|qSrrrrVr<rrr Cr&z0get_async..fire_task..)argscallbackN)popr2r rJput)r$frE apply_asyncr6r@rBrArC pretask_cbsqueuer<rr fire_task:s  zget_async..fire_taskr,crkrlrrmrnrrr \s zget_async..rT)rr1r^rr r/rr rNr r=rr ValueErrorlenrr rXrY)"ru num_workersr6rFrrBrjrCraise_exception callbacksr@rAkwargs result_flatrU_ posttask_cbs started_cbs succeededcbkeyorder start_staterxr$res_inforHrerfrErDres worker_idrsfinishrrtr get_asyncs|9            UrrcCs2|duri}||i|}|dur||dSdS)z*A naive synchronous version of apply_asyncNr)funcrokwdsrprrrr apply_syncws  r)NN)Tr)rNN)%__doc__oswarningsrwrrdaskrdask._task_specrr ImportErrorwarn __version__dask.callbacksrr dask.corer r r dask.orderr namerr=rJrLrOrXrYrarbrgrirrrrrrsV      A  $