U /e*@sUdZddlmZddlmZddlmZddlmZddl m Z m Z m Z m Z mZmZddlZddlmZdd lmZdd lmZd Zd ed <dZded<ddZefddddddddZddddddZGd d!d!eZdS)"z4Utilities for generating and analyzing cluster dumps) annotations) defaultdict)Mapping)Path)IOAny AwaitableCallable CollectionLiteralN)scheduler_story) worker_story) to_threadmsgpackzLiteral['msgpack' | 'yaml']DEFAULT_CLUSTER_DUMP_FORMAT)Zrun_specCollection[str]DEFAULT_CLUSTER_DUMP_EXCLUDEcCs@t|ttfrdd|DSt|tr8dd|DS|SdS)NcSsg|] }t|qS_tuple_to_list).0elrrsz"_tuple_to_list..cSsi|]\}}|t|qSrr)rkvrrr sz"_tuple_to_list..) isinstancelisttupledictitems)noderrrrs  rzCallable[[], Awaitable[Any]]strzLiteral[('msgpack', 'yaml')]zdict[str, Any]None) get_stateurlformatstorage_optionsreturnc s|dkr*d}d}||s"||7}tj}nR|dkrlddld}d}||sT||7}d d d d fd d }ntd|dddl}|j||fddi|$}|IdH} t|| |IdHW5QRXdS)z;Await a cluster dump, then serialize and write it to a pathrwb .msgpack.gzyamlrNw.yamlr rr$)statefr)cst||dSN)dumpr)r/r0r,rrwriter2szwrite_state..writerzUnsupported format z*. Possible values are 'msgpack' or 'yaml'. compressioninfer)endswithrpackr, ValueErrorfsspecopenr) r%r&r'r(modesuffixr4r:r0r/rr3r write_states(    r>rr r&kwargsr)c Ks|drd}tj}n.|dr4ddl}d}|j}ntd|d|d d ddl}|j||f|}||W5QRSQRXdS) aLoads a cluster dump from a disk artefact Parameters ---------- url : str Name of the disk artefact. This should have either a ``.msgpack.gz`` or ``yaml`` suffix, depending on the dump format. **kwargs : Extra arguments passed to :func:`fsspec.open`. Returns ------- state : dict The cluster state at the time of the dump. r+rbr.rNrzurl (z)) must have a .msgpack.gz or .yaml suffixr5r6) r7runpackr,Z safe_loadr9 setdefaultr:r;)r&r@r<readerr,r:r0rrrload_cluster_dumpIs   rFc@seZdZdZddddZeddddd d Zd d Zd dZddZ ddddddZ d2dddddZ d3dddddZ ddddd Z dddd!d"Zdd#d$d%Zdd&dd'd(d)Zd4d,d-d-d.d/d0d1ZdS)5 DumpArtefacta Utility class for inspecting the state of a cluster dump .. code-block:: python dump = DumpArtefact.from_url("dump.msgpack.gz") memory_tasks = dump.scheduler_tasks("memory") executing_tasks = dump.worker_tasks("executing") r r/cCs ||_dSr1r2selfr/rrr__init__xszDumpArtefact.__init__r#rr?cKstt|f|S)aLoads a cluster dump from a disk artefact Parameters ---------- url : str Name of the disk artefact. This should have either a ``.msgpack.gz`` or ``yaml`` suffix, depending on the dump format. **kwargs : Extra arguments passed to :func:`fsspec.open`. Returns ------- state : dict The cluster state at the time of the dump. )rGrF)clsr&r@rrrfrom_url{szDumpArtefact.from_urlcCs |j|Sr1rI)rKkeyrrr __getitem__szDumpArtefact.__getitem__cCs t|jSr1)iterr2rKrrr__iter__szDumpArtefact.__iter__cCs t|jSr1)lenr2rRrrr__len__szDumpArtefact.__len__z str | Nonezdict[str, dict]z list[dict])r/contextr)cs*rfdd|DSt|SdS)Ncsg|]}|dkr|qSrHr)rrrHrrrs z/DumpArtefact._extract_tasks..)valuesr)rKr/rVrrHr_extract_tasksszDumpArtefact._extract_tasksNr)r/r)cCs|||jddS)a- Parameters ---------- state : optional, str If provided, only tasks in the given state are returned. Otherwise, all tasks are returned. Returns ------- tasks : list The list of scheduler tasks in ``state``. schedulertasks)rXr2rJrrrscheduler_tasks_in_states z%DumpArtefact.scheduler_tasks_in_statecCsDg}|jdD],}t|trd|kr||||dq|S)a) Parameters ---------- state : optional, str If provided, only tasks in the given state are returned. Otherwise, all tasks are returned. Returns ------- tasks : list The list of worker tasks in ``state`` workersrZ)r2rWrr extendrX)rKr/rZ worker_dumprrrworker_tasks_in_states z"DumpArtefact.worker_tasks_in_state)key_or_stimulus_idr)cGsLtt}|jdd}t|}t||D]}||dt|q(t|S)z Returns ------- stories : dict A list of stories for the keys/stimulus ID's in ``*key_or_stimulus_id``. rYtransition_logr)rrr2set_scheduler_storyappendrr )rKr`storieslogkeysstoryrrrr s zDumpArtefact.scheduler_storycGsht|}tt}|jdD]@}t|trd|krt||dD]}||dt |qBqt|S)z Returns ------- stories : dict A dict of stories for the keys/stimulus ID's in ``*key_or_stimulus_id`.` r\rfr) rbrrr2rWrr _worker_storyrdr)rKr`rgrer^rhrrrr szDumpArtefact.worker_story)r)cs*|jdd}|jdfdd|DS)z Returns ------- missing : list A list of workers connected to the scheduler, but which did not respond to requests for a state dump. rYr\cs&g|]}|kst|ts|qSr)rr )rr-Zresponsive_workersrrrsz0DumpArtefact.missing_workers..rI)rKZscheduler_workersrrjrmissing_workerss   zDumpArtefact.missing_workerszset[str])r/ expand_keysr)cCsLd|ks ti}i}|D]"\}}||kr6|||<q|||<q||d<|S)zXCompacts ``state`` keys into a general key, unless the key is in ``expand_keys``general)AssertionErrorr!)rKr/rlresultrmrrrrr_compact_states   zDumpArtefact._compact_stateconfigrfZlogsrZevents extensionsrfZ task_groupsrZrar\zstr | Path | Nonerr$)root_dirworker_expand_keysscheduler_expand_keysr)c CsXddl}|rt|nt}|j}t|}t|}|jd}|D]}z |d}Wntk rlYqDYnX|||} ||} | j ddd| D]@\} } t | | d} t | d}|j| ||d W5QRXqqDd }||j||}||} | j ddd| D]B\} } t | | d} t | d}|j| ||d W5QRXqdS) a Splits the Dump Artefact into a tree of yaml files with ``root_dir`` as it's base. The root level of the tree contains a directory for the scheduler and directories for each individual worker. Each directory contains yaml files describing the state of the scheduler or worker when the artefact was created. In general, keys associated with the state are compacted into a ``general.yaml`` file, unless they are in ``scheduler_expand_keys`` and ``worker_expand_keys``. Parameters ---------- root_dir : str or Path The root directory into which the tree is written. Defaults to the current working directory if ``None``. worker_expand_keys : iterable of str An iterable of artefact worker keys that will be expanded into separate yaml files. Keys that are not in this iterable are compacted into a `general.yaml` file. scheduler_expand_keys : iterable of str An iterable of artefact scheduler keys that will be expanded into separate yaml files. Keys that are not in this iterable are compacted into a ``general.yaml`` file. rNr\idT)parentsexist_okr.r-)ZDumperrY) r,rcwdZ CSafeDumperrbr2rWKeyErrorrpmkdirr!r#r;)rKrvrwrxr,Zdumperr\infoZ worker_idZ worker_stateZlog_dirnameZ_logsfilenamefdrVZscheduler_staterrrto_yamlss4*       zDumpArtefact.to_yamls)N)N)Nrqrs)__name__ __module__ __qualname____doc__rL classmethodrNrPrSrUrXr[r_r r rkrprrrrrrGms$ rG)r __future__r collectionsrcollections.abcrpathlibrtypingrrrr r r rZdistributed._storiesr rcr riZdistributed.compatibilityrr__annotations__rrr>rFrGrrrrs            ,$