ԋg\TddlmZddlZddlZddlZddlmZddlmZm Z m Z m Z ddl m Z ddlmZddlmZmZmZddlZddlmZdd lmZdd lmZdd lmZerdd lmZm Z dd lm!Z"ddl#m$Z$ddl%m&Z&m'Z'dZ(GddeZ)e ddZ*GddZ+GddZ,GddZ-y)) annotationsN) defaultdict)HashableIterableIteratorMapping)contextmanager)islice) TYPE_CHECKINGAny TypedDict)Key) sum_mappings)ffill)time) SchedulerWorker) scheduler) SourceCode) TaskGroupTaskStateState)executep2pceZdZUded<y) SpanMetadataz list[dict] collectionsN)__name__ __module__ __qualname____annotations__1lib/python3.12/site-packages/distributed/spans.pyrr sr"rc'*K|s tdtjjd}|r|dnd}|r|dnd}t d|D}tj ||z||zd5|d d d d y #1swYy xYww) a!Tag group of tasks to be part of a certain group, called a span. This context manager can be nested, thus creating sub-spans. If you close and re-open a span context manager with the same tag, you'll end up with two separate spans. Every cluster defines a global "default" span when no span has been defined by the client; the default span is automatically closed and reopened when all tasks associated to it have been completed; in other words the cluster is idle save for tasks that are explicitly annotated by a span. Note that, in some edge cases, you may end up with overlapping default spans, e.g. if a worker crashes and all unique tasks that were in memory on it need to be recomputed. You may capture the ID of a span on the client to match it with the :class:`~distributed.spans.Span` objects the scheduler: >>> client = Client() >>> with span("my workflow") as span_id: ... client.submit(lambda: "Hello world!").result() >>> client.cluster.scheduler.extensions["spans"].spans[span_id] Span Notes ----- You may retrieve the current span with ``dask.get_annotations().get("span")``. You can do so in the client code as well as from inside a task. z"Must specify at least one span tagspannamer!idsc3VK|]!}ttj#ywN)struuiduuid4).0_s r# zspan..Js04aDJJL!4s')r&r')r%N) ValueErrordaskget_annotationsgettupleannotate)tags annotation prev_tagsprev_idsr's r#r%r%$s: =>>%%'++F3J&0 6"bI%/z% BH 040 0C Y%5hnM N"g  O N NsA4B6B> BB BceZdZUded<ded<ded<ded<d ed <d ed <d ed<ded<ded<ded<ded<ded<eeZeZded< d2dZd3dZ e d4dZ d5dZ e d6dZ d7d Zd8d!Ze d9d"Ze d9d#Ze d:d$Ze d;d%Ze dd)Ze d?d*Ze d@d+ZedAd,ZdBd-ZdCd.Ze dDd/Ze d9d0Zy1)ESpantuple[str, ...]r&r*idzweakref.ref[Span] | None_parent list[Span]childrenzset[TaskGroup]groupsfloatenqueuedz"dict[tuple[SourceCode, ...], None]_codeSpanMetadata | None _metadataz(defaultdict[tuple[Hashable, ...], float]_cumulative_worker_metricslist[tuple[float, int]]_total_nthreads_historyint_total_nthreads_offsetr __weakref__zset[int]_metadata_seenc>||_||_|tj|nd|_t |_g|_t|_ i|_ d|_ tt|_t|dkDsJ||_t|dz |_y)Nr)r&r?weakrefrefr@rrErBsetrCrFrHrrLrIlenrKrM)selfr&id_parenttotal_nthreads_historys r#__init__z Span.__init__s .4.@w{{6*d   e  +6c*:')*Q...'=$&)*@&AA&E#r"c<d|jd|jdS)Nz Span)r&r?rVs r#__repr__z Span.__repr__sDII;eDGG9A66r"cH|jr|j}|sJ|Syr))r@rVouts r#rXz Span.parents# <<,,.CJ3Jr"ct|}||jvry|jj||jt j ||_y|jdj |dy)z,Add metadata to the span, e.g. code snippetsNr)r?rOaddrHcopydeepcopyextend)rVmetadatarWs r# add_metadatazSpan.add_metadatasfl $%% %  $ >> !!]]84DN NN= ) 0 0-1H Ir"c|jdk(ryg}|}|r*|j|j|j}|r*|jt t |dS)zRebuild the dask graph annotation which contains the full id history Note that this may not match the original annotation in case of TaskGroup collision. defaultNr0)r&appendr?rXr6reversed)rVr'nodes r#r9zSpan.annotationsX 99 $  JJtww ;;D % *>??r"c#jK||jD]}|jEd{y7w)zgTop-down recursion of all spans belonging to this branch off span tree, including self N)rBtraverse_spans)rVchilds r#rpzSpan.traverse_spanss0 ]]E++- - -# -s '313c#bK|jD]}|jEd{y7w)z4All TaskGroups belonging to this branch of span treeN)rprC)rVr%s r#traverse_groupszSpan.traverse_groupss)'')D{{ " "* "s #/-/cztd|jDd}|rt||j}|S)aEarliest time when a task belonging to this span tree started computing; 0 if no task has *finished* computing yet. Notes ----- This is not updated until at least one task has *finished* computing. It could move backwards as tasks complete. See Also -------- enqueued stop distributed.scheduler.TaskGroup.start c3TK|] }|jdk7s|j"yw)N)startr-tgs r#r/zSpan.start..s! J 6""((c/RXX 6s((rvrj)minrsmaxrEr`s r#rwz Span.starts<  J 4 4 6 J  c4==)C r"c|jr!td|jD}n t}t||jS)atWhen this span tree finished computing, or current timestamp if it didn't finish yet. Notes ----- This differs from ``TaskGroup.stop`` when there aren't unfinished tasks; is also will never be zero. See Also -------- enqueued start done distributed.scheduler.TaskGroup.stop c34K|]}|jywr))stoprxs r#r/zSpan.stop..s?(>"bgg(>)doner{rsrrEr`s r#r~z Span.stops>" 99?(<(<(>??C&C3 &&r"c|jSr))rHr]s r#rgz Span.metadatas ~~r"cBtd|jDS)zThe number of tasks currently in each state in this span tree; e.g. ``{"memory": 10, "processing": 3, "released": 4, ...}``. See Also -------- distributed.scheduler.TaskGroup.states c34K|]}|jywr))statesrxs r#r/zSpan.states..sG0F"BII0Frrrsr]s r#rz Span.states sG0D0D0FGGGr"cBtd|jDS)aReturn True if all tasks in this span tree are completed; False otherwise. Notes ----- This property may transition from True to False, e.g. when a new sub-span is added or when a worker that contained the only replica of a task in memory crashes and the task need to be recomputed. See Also -------- distributed.scheduler.TaskGroup.done c34K|]}|jywr))rrxs r#r/zSpan.done..$s<%;r277%;r)allrsr]s r#rz Span.dones./sN7MB,,7Mrrr]s r#rzSpan.all_durations&sNt7K7K7MNNNr"cBtd|jDS)zThe total amount of time spent on all tasks in this span tree See Also -------- all_durations distributed.scheduler.TaskGroup.duration c34K|]}|jywr))durationrxs r#r/z Span.duration..:s@)?22;;)?rsumrsr]s r#rz Span.duration1s@)=)=)?@@@r"cBtd|jDS)zThe total number of bytes that this span tree has produced See Also -------- distributed.scheduler.TaskGroup.nbytes_total c34K|]}|jywr)) nbytes_totalrxs r#r/z$Span.nbytes_total..DsD-Cr2??-Crrr]s r#rzSpan.nbytes_total<sDT-A-A-CDDDr"chttjd|jDS)zCode snippets, sent by the client on compute(), persist(), and submit(). Only populated if ``distributed.diagnostics.computations.nframes`` is non-zero. c3BK|]}|jD]}|ywr))rF)r-rqscs r#r/zSpan.code..NsV*?%++B"+"*?s)listdictfromkeysrpr]s r#codez Span.codeFs, MMV$*=*=*?V V  r"ctd|jD}td|jD}t d|j |z }||d<|S)aReplica of ``Worker.digests_total`` and ``Scheduler.cumulative_worker_metrics``, but only for the metrics that can be attributed to the current span tree. The span id has been removed from the key. At the moment of writing, all keys are ``("execute", , , )`` or ``("p2p", , , )`` but more may be added in the future with a different format; please test e.g. for ``k[0] == "execute"``. c34K|]}|jywr))rIr-rqs r#r/z1Span.cumulative_worker_metrics..^s :OE , ,:Orc3JK|]\}}|ddk(s|ddk(s|yw)rrr1secondsNr!)r-kvs r#r/z1Span.cumulative_worker_metrics..as. %$!Q1):qu ?QA+s ###rv)rzN/Azidle or other spansr)rrpritemsr{active_cpu_seconds)rVra known_secondsunknown_secondss r#cumulative_worker_metricszSpan.cumulative_worker_metricsQsl :>:M:M:O   ))+  c4#:#:]#JKBQ >? r"c|s tdtddd|dj}td|D|_|j j |td|D|_|S) zpMerge multiple spans into a synthetic one. The input spans must not be related with each other. zNothing to merge)(merged)rNrr&rWrXrYc34K|]}|jywr))rMrs r#r/zSpan.merge..xs) 6;UE ( (erc34K|]}|jywr))rErs r#r/zSpan.merge..|s=ue5>>ur)r2r=rKrzrMrBrfrE)rras r#mergez Span.mergeksz /0 0#(8#C#C   &)) 6;) & " E"=u==  r"c#K|jr |jnd}t|j|jdD](\}}|r||k\ryt |j ||f*yw)zEYield (timestamp, number of threads across the cluster), forward-fillrN)rr~r rKrMr{rE)rVr~tns r#_nthreads_timeserieszSpan._nthreads_timeseriessa IItyy1  ( ($*E*Et DAqT dmmQ'* *  sA(A*c#HK|jdk7r!|jdf|jdfyg}|jD]!}||jdf|jdfgz }#|j dd }|D]"\}}|s |dk(sJ|df||z }|r|df$yw) zIf this span is the output of :meth:`merge`, yield (timestamp, True if at least one input span is active), forward-fill. rTFNrQr1c |dSNrr!)els r#z)Span._active_timeseries..s2a5r")keyr)r?rEr~rBsort)rVeventsrqn_activerdeltas r#_active_timeserieszSpan._active_timeseriess 77j --% %))U" " ]]E *UZZ,<= =F#  ( )HAuz!zg  Hh s BB" B"c .t|j\}}t|j\}}th||}t |||d}t |||d}t||dd||D cgc]\}} } } | r|| | fc} } } }Scc} } } }w)a Returns ------- List of tuples: - begin timestamp - end timestamp - Scheduler.total_nthreads during this interval When the Span is the output of :meth:`merge`, the intervals may not be contiguous. See Also -------- enqueued stop active_cpu_seconds distributed.scheduler.SchedulerState.total_nthreads r)leftFrQN)ziprrsortedr) rV nthreads_tnthreads_count is_active_tis_active_flagt_interpnthreads_count_interpis_active_flag_interpt0t1ractives r#nthreads_intervalszSpan.nthreads_intervalss*&)$*C*C*E%F" N&)4+B+B+D&E# ^5J556 %h NQR S %h ^RW X&)(12,(=?T& &!B6 QK&   s6B c:td|jDS)a Return number of CPU seconds that were made available on the cluster while this Span was running; in other words ``(Span.stop - Span.enqueued) * Scheduler.total_nthreads``. This accounts for workers joining and leaving the cluster while this Span was active. If this Span is the output of :meth:`merge`, do not count gaps between input spans. See Also -------- enqueued stop nthreads_intervals distributed.scheduler.SchedulerState.total_nthreads c34K|]\}}}||z |zywr)r!)r-rrnthreadss r#r/z*Span.active_cpu_seconds..s#X@W,rWr*rX Span | NonerYrJ)returnr*)rr)rgrrNone)rz!dict[str, tuple[str, ...]] | None)rIterator[Span])rzIterator[TaskGroup])rrD)rrG)rzdict[TaskStateState, int])rbool)rzdict[str, float])rrL)rzlist[tuple[SourceCode, ...]]r!dict[tuple[Hashable, ...], float])rr=rr=)rzIterator[tuple[float, int]])rzIterator[tuple[float, bool]])rzlist[tuple[float, float, int]]) rrrr r6 __slots__rTrOrZr^propertyrXrhr9rprsrwr~rgrrrrrrr staticmethodrrrrrr!r"r#r=r=Os  G %% O .-"" HH54o&I"uNH$FFF F !8 F.7 J @ @.# 0''0HH = =OOAAEE  2&+4  BYYr"r=ceZdZUdZded<ded<ded<ded <d ed <dd Z dd ZddZddZddZ ddZ ddZ ddZ y)SpansSchedulerExtensionz%Scheduler extension for spans supportrrzdict[str, Span]spansrA root_spansz(defaultdict[tuple[str, ...], list[Span]]spans_search_by_namezdefaultdict[str, list[Span]]spans_search_by_tagc~||_i|_g|_tt|_tt|_yr))rrrrrrr)rVrs r#rZz SpansSchedulerExtension.__init__s1" $/$5!#.t#4 r"ci}d}|D]4}|jt|_|j}|jr|j|j}nv|jj d} | r|j | d| d}n|s|j}|}|j|_|jj||rd|j|<|r|j||jx} r | x|jd<||j<|jjdd7|S)aAcknowledge the existence of runnable tasks on the scheduler. These may either be new tasks, tasks that were previously unrunnable, or tasks that were already fed into this method already. Attach newly observed tasks to either the desired span or to ("default", ). Update TaskGroup.span_id and wipe TaskState.annotations["span"]. Returns ------- Updated 'span' annotations: {key: {"name": (..., ...), "ids": (..., ...)}} Nr%r&r')rrgroupspan_idrr5 _ensure_span_ensure_default_spanr?rCrcrFrhr9rpop) rVtssr span_metadatara default_spantsryr%anns r# observe_tasksz%SpansSchedulerExtension.observe_taskss" B~~%!%Bzzzz"**-nn((0,,S[#e*ED''+'@'@'B 'D!WW  ##' 4 !!-0oo%s%7::v&RVV""640?B r"c|jd}|r|djs|dS|jdtt j fS)aReturn the currently active default span, or create one if the previous one terminated. In other words, do not reuse the previous default span if all tasks that were not explicitly annotated with :func:`spans` on the client side are finished. rjr1)rrrr*r+r,)rVdefaultss r#rz,SpansSchedulerExtension._ensure_default_span1sN ,,Z8 HRL--B<   DJJL0A/CDDr"c| |j|dS#t$rYnwxYwt|t|k(sJt|dkDsJd}tdt|D]}|j |d||d|}t ||d||j j}||j|j<|j|j||D] }|j|j|"|r|jj||S|jj||S)z-Create Span if it doesn't exist and return itr1rNrQr)rKeyErrorrUrangerr=rrYr?rrlrrBr)rVr&r'rXir%tags r#rz$SpansSchedulerExtension._ensure_span<s7 ::c"g& &   4yCH$$$4y1}}q#d)$A&&tBQxRa9F%B#'>>#H#H   # 477 !!$'..t4C  $ $S ) 0 0 6  OO " "4 (  OO " "4 ( s   c'ZKtt}|D]=}|j|D])}|t|jj |+?t }t|jD]0\}}|j||D]}|j|vs|2yw)zYield all spans that contain any of the given tags. When a tag is shared both by a span and its (grand)children, only return the parent. N) rrrrUr&rlrTrrupdaterX)rVr8by_levelrspseenr.levels r# find_by_tagsz$SpansSchedulerExtension.find_by_tags[s t$C..s3RWW&--b14ux~~/0HAu KK 99D(H1s BB+" B+c:tj|jS)z5Return a synthetic Span which is the sum of all spans)r=rrr]s r# merge_allz!SpansSchedulerExtension.merge_alllszz4??++r"c@tj|j|S)z`Return a synthetic Span which is the sum of all spans containing the given tags )r=rr)rVr8s r# merge_by_tagsz%SpansSchedulerExtension.merge_by_tagsps!zz,4,,d344r"c|jD]D\^}}}}t|tsJ|j|}|j|g|xx|z cc<Fy)aTriggered by :meth:`SpansWorkerExtension.heartbeat`. Populate :meth:`Span.cumulative_worker_metrics` with data from the worker. See Also -------- SpansWorkerExtension.heartbeat Span.cumulative_worker_metrics N)r isinstancer*rrI)rVwsdatacontextrotherrr%s r# heartbeatz!SpansSchedulerExtension.heartbeatvs\.2ZZ\ ) &Wggs+ ++::g&D  + +W,=u,= >! C >.:r"N)rr)rz$Iterable[scheduler_module.TaskState]rztuple[SourceCode, ...]rrrzdict[Key, dict])rr=)r&r>r'r>rr=)r8r*rr)r8r*rr=)rzscheduler_module.WorkerStaterrrr) rrr__doc__r rZrrrrrrrr!r"r#rrs/ CB6555 15%5$ 5  5n E>",5 D.D6WD Dr"rcFeZdZUdZded<ded<d dZ d dZd dZy ) SpansWorkerExtensionz"Worker extension for spans supportrworkerrdigests_total_since_heartbeatc ||_i|_yr))rr)rVrs r#rZzSpansWorkerExtension.__init__s -/*r"c|jrJ|jDcic]#\}}t|tr|dtvr||%c}}|_ycc}}wr)rrrr6CONTEXTS_WITH_SPAN_ID)rVrrrs r#collect_digestsz$SpansWorkerExtension.collect_digestssa 55556;;=. =1!U#!0E(E qD=. *. s(Ac,|j}i|_|S)aTApportion the metrics that do have a span to the Spans on the scheduler Returns ------- ``{(context, span_id, prefix, activity, unit): value}}`` See Also -------- SpansSchedulerExtension.heartbeat Span.cumulative_worker_metrics distributed.worker.Worker.get_metrics )rr`s r#rzSpansWorkerExtension.heartbeats00-/* r"N)rr)rzMapping[Hashable, float]rrr)rrrrr rZr rr!r"r#rrs/, N#DD0  -E    r"r)r8r*rz Iterator[str]). __future__rrdr+rRrrcollections.abcrrrr contextlibr itertoolsr typingr r r dask.configr3 dask.typingrdistributed.collectionsrdistributed.itertoolsrdistributed.metricsr distributedrrrscheduler_moduledistributed.clientrdistributed.schedulerrrr rr%r=rrr!r"r#rs" #AA%000'$.9-?+9''TGYGYT lDlD^%%r"