o 1 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)getrqri/home/app/PaddleOCR-VL-test/.venv_paddleocr/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.0kvZ data_keysrr Ps"z)start_state_from_dask..rcSsi|] \}}|r||qSr)copyrrrrrVscSsh|]\}}|s|qSrrrrrr Xsz(start_state_from_dask..TkeyreversecSsi|] \}}|r||qSrrrrrrrZ) dependencies dependentswaiting waiting_datarreadyrunningfinishedreleased)r rrdictritems isinstanceraddr updaterr removesorted)dskrsortkeyrrZdsk2r&r(r'abr)Z 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#Z task_infodumpsloadsget_idpack_exceptiontaskdataresultidfailederrr execute_taskks    rFTcCsJ||dvr|d|rJ|d|=|d||r#|d|=dSdS)zQRemove data from temporary storage See Also -------- finish_task r)r-rN)r1)r#r9deleterrr release_data~s  rHFc 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)rGr,r+) r4r3appendDEBUGZ chest.corerIprintsummapvaluesr1) r5r#r9resultsr6rGrHdepsrIrrr finish_tasks6         rScs(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..)r0listtuple)indrWrrVrrTs rTcCsdS)zDefault get_idNrrrrrdefault_get_idsr\cCsrr)rEr<rrrdefault_pack_exceptionsr]cCs|j|ur |||r) __traceback__with_traceback)exctbrrrreraises  rbcCs|S)z>> identity(3) 3 r)xrrridentitysrdc " 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)rr6Nrerun_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|qSrrrrQr9rrrCr%z0get_async..fire_task..)argscallbackN)popr1r rFput)r#frA apply_asyncr5r<r>r=r?Z pretask_cbsqueuer9rr fire_task:s  zget_async..fire_taskr+crfrgrrhrirrr\s zget_async..rT)rr0rYrr r.rr rJr r:rr ValueErrorlenrr rSrT)"rpZ num_workersr5rBrr>rer?Zraise_exception callbacksr<r=kwargsZ result_flatrP_Z posttask_cbsZ started_cbsZ succeededcbZkeyorderZ start_staterrr#Zres_inforDr`rarAr@resZ worker_idrnfinishrror get_asyncs|9            Ur{rcCs2|duri}||i|}|dur||dSdS)z*A naive synchronous version of apply_asyncNr)funcrjkwdsrkryrrr apply_syncws  r~)NN)Tr)rNN)%__doc__oswarningsrqrrZdaskrZdask._task_specrr ImportErrorwarn __version__Zdask.callbacksrr Z dask.corer r r Z dask.orderr namerr:rFrHrKrSrTr\r]rbrdr{r~rrrrsV      A  $