ԋg8ddlmZddlZddlZddlZddlZddlZddlm Z m Z m Z ddl m Z ddlmZddlmZddlmZddlmZdd lmZddlZdd lmZej6d d Zefd ZedZedZGddZ erLejBdkr=e ej dj Ze ejDdj Z"nej ZejDZ"ejFZ# ejHZ$eGddZ'e ejPdf ddZ)GddZ*e*Z+GddZ,y#e%e&f$re#Z$YQwxYw)) annotationsN)CallableHashableIterator)contextmanager) ContextVar) dataclass)wraps)nan)Literal)WINDOWS_empty_namedtuplecRtt|tfd}|S)zo Return a function calling the given psutil *method_name*, or returning *default* if psutil fails. c@ S#t$r cYSwxYwN) RuntimeError)defaultmeths3lib/python3.12/site-packages/distributed/metrics.pywrapperz_psutil_caller..wrappers& 6M 9  s  )getattrpsutilr ) method_namerrrs ` @r_psutil_callerrs. 6; 'D 4[ Ndisk_io_countersnet_io_counterscjeZdZUdZded<ded<ded<ded<ded < d dd Zdd Zdd Zy ) _WindowsTimezxCombine time.time() or time.monotonic() with time.perf_counter() to get an absolute clock with fine resolution. Callable[[], float] base_timerfloatdeltaz float | Noneprevious next_resync resync_everych||_|r tdnd|_td|_||_y)Nz-inf)r"r#r%r&r')selfbase is_monotonicr's r__init__z_WindowsTime.__init__9s.)5f 4  =(rctj}||jkDr$|j||jz|_||j z }|j %||j kr|j dz}||_|S)Ng& .>)timemod perf_counterr&resyncr'r$r%)r)curs rtimez_WindowsTime.timeAsx""$ !! ! KKM"T%6%66D  tzz == $dmm#mmd*DM rc|j}tj}d} t|dzDcgc]}||f}}t j d|D}|j d\}}||kr`|D cgc]} | d|k(s | dc} dd} t| |dz k\sJ| |t| t| z z |_ ycc}wcc} w)Nc3&K|] }|d yw)rNr).0ts r z&_WindowsTime.resync..Ts+@%QAaD%sr) r"r.r/range collectionsCounter most_commonlensumr$) r)_time _perf_counter min_samples_times abs_timesfirstnfirstr8 perf_timess rr0z_WindowsTime.resyncNs,,  9>{Q9OP9OAeg}/9OEP#+++@%+@@I%113A6ME6 #(-?11!A$?DJz?kAo5 Az A5Z3z?!BBDJ Q@sC CCN)g@)r*r!r+boolr'r#)returnr#)rLNone)__name__ __module__ __qualname____doc____annotations__r,r2r0rrrr r .sT$# LTY)')7;)KP) rr ) F)r+Tc<eZdZUded<ded<ded<eeZy) MeterOutputr#startstopr$N)rNrOrPrRtuple __slots__rrrrVrVts L K Lo&IrrVc#|Kt|tt} |||_|j|jz |_|durt ||j|_yy#||_|j|jz |_|durt ||j|_wwxYww)a1Convenience context manager which calls func() before and after the wrapped code and calculates the delta. Parameters ---------- label: str label to pass to the callback func: callable function to call before and after, which must return a number. Besides time, it could return e.g. cumulative network traffic or disk usage. Default: :func:`timemod.perf_counter` floor: float or False, optional Floor the delta to the given value (default: 0). This is useful for strictly cumulative functions that can occasionally glitch and go backwards. Set to False to disable. Yields ------ :class:`MeterOutput` where the ``start`` attribute is populated straight away, while ``stop`` and ``delta`` are nan until context exit. FN)rVr rXrWr$max)funcfloorouts rmeterra|s4 dfc3 'C. 6HHsyy(  E399-CI 6HHsyy(  E399-CI sB<A-A B<-A B99B<ceZdZUdZded<dZdZedZe ddd  dd Z e dd Z dd Z e d e jdf ddZy) ContextMeteraContext-based general purpose meter. Usage ----- 1. In high level code, call :meth:`add_callback` to install a hook that defines an activity 2. In low level code, typically many stack levels below, log quantitative events (e.g. elapsed time, transferred bytes, etc.) so that they will be attributed to the high-level code calling it, either with :meth:`meter`, :meth:`meter_function`, or :meth:`digest_metric`. Examples -------- In the code that e.g. sends a Python object from A to B over the network: >>> from distributed.metrics import context_meter >>> with context_meter.add_callback(partial(print, "A->B comms:")): ... await send_over_the_network(obj) In the serialization utilities, called many stack levels below: >>> with context_meter.meter("dumps"): ... pik = pickle.dumps(obj) >>> with context_meter.meter("compress"): ... pik = lz4.compress(pik) And finally, elsewhere, deep into the TCP stack: >>> with context_meter.meter("network-write"): ... await comm.write(frames) When you call the top-level code, you'll get:: A->B comms: dumps 0.012 seconds A->B comms: compress 0.034 seconds A->B comms: network-write 0.567 seconds zBContextVar[dict[Hashable, Callable[[Hashable, float, str], None]]] _callbackscBtdt|di|_y)Nz MetricHook._callbacks)r)ridrdr)s rr,zContextMeter.__init__s $"T(< 0" rc<|tusJd|jdfS)NzFound copy of singletonr) context_meter_unpickle_singletonrgs r __reduce__zContextMeter.__reduce__s'}$?&??$''++rctSr)rirrrrjz ContextMeter._unpickle_singletonsrNF)key allow_offloadc#K|r1tjtjdfd }n}| t }|j j }|j}|||<|j j|} d|jj|y#|jj|wxYww)a5Add a callback when entering the context and remove it when exiting it. The callback must accept the same parameters as :meth:`digest_metric`. Parameters ---------- callback: Callable ``f(label, value, unit)`` to be executed key: Hashable, optional Unique key for the callback. If two nested calls to ``add_callback`` use the same key, suppress the outermost callback. allow_offload: bool, optional If set to True, this context must be executed inside a running asyncio event loop. If a call to :meth:`digest_metric` is performed from a different thread, e.g. from inside :func:`distributed.utils.offload`, ensure that the callback is executed in the event loop's thread instead. crtjk(r |||yj|||yr) threading get_identcall_soon_threadsafe)labelvalueunitcallbacklooptids rsafe_cbz*ContextMeter.add_callback..safe_cbs4&&(C/UE40--hudKrNrtrrur#rvstrrLrM) asyncioget_running_looprqrrobjectrdgetcopysetvarreset) r)rwrmrnrzcbstokrxrys ` @@r add_callbackzContextMeter.add_callbacks0 ++-D%%'C L LG ;(Coo!!#hhjCoo!!#&   GGMM# CGGMM# sB CB2C2CCc#K|jji} d|jj|y#|jj|wxYww)z8Do not trigger any callbacks set outside of this contextN)rdrrr)r)rs rclear_callbackszContextMeter.clear_callbackssDoo!!"%   GGMM# CGGMM# sA?AAAcv|jj}|jD] }||||y)zaInvoke the currently set context callbacks for an arbitrary quantitative metric. N)rdrvalues)r)rtrurvrcbs r digest_metriczContextMeter.digest_metric s2oo!!#**,B ueT "rsecondsr[c#Kdk7r9 t||5}|ddd|j|jygdfd } |j|5t|d5}|ddddddjt z }|dur t ||}||_|j||y#1swYxYw#|j|jwxYw#1swYxYw#1swYxYw#jt z }|dur t ||}||_|j||wxYww)aConvenience context manager or decorator which calls func() before and after the wrapped code, calculates the delta, and finally calls :meth:`digest_metric`. If unit=='seconds', it also subtracts any other calls to :meth:`meter` or :meth:`digest_metric` with the same unit performed within the context, so that the total is strictly additive. Parameters ---------- label: Hashable label to pass to the callback unit: str, optional unit to pass to the callback. Default: seconds func: callable see :func:`meter` floor: bool, optional see :func:`meter` Yields ------ :class:`MeterOutput` where the ``start`` attribute is populated straight away, while ``stop`` and ``delta`` are nan until context exit. In case of multiple nested calls to :meth:`meter`, then delta (for seconds only) is reduced by the inner metrics, to a minimum of ``floor``. r)r_Nc4|k(rj|yyr)append)label2value2unit2offsetsrvs rrwz$ContextMeter.meter..callback?s}v&rF)rrrr#rr|rLrM)rarr$rrAr]) r)rtrvr^r_mrwr$rs ` @rrazContextMeter.meters5B 9  94u-G.""5!''48  '  3""8,eD.F!/G,GGc'l*EE!E5)AG   ueT 2/.-""5!''48/G.F,,GGc'l*EE!E5)AG   ueT 2s{E CCC'E D C>,C21C>9D AEC  CC//E2C; 7C>>DD AEE)rwz&Callable[[Hashable, float, str], None]rmHashable | NonernrKrLIterator[None])rLrr{) rtrrvr|r^r!r_float | Literal[False]rLIterator[MeterOutput])rNrOrPrQrRr,rk staticmethodrjrrrrr.r/rarrrrcrcs DSR , $# -8- -  -  --^#$+$8$8(+ 939393" 93 & 93  9393rrcceZdZUdZded<ded<ded<ej fddZdd Ze d d dd Z d dd Z y )DelayedMetricsLedgeraAdd-on to :class:`ContextMeter` that helps in the case where: - The code to be metered is not easily expressed as a self-contained code block e.g. you want to measure latency in the asyncio event loop before and after running a task - You want to alter the metrics depending on how the code ends; e.g. you want to post them differently in case of failure. Examples -------- >>> ledger = DelayedMetricsLedger() # Metering starts here >>> async def wrapper(): ... with ledger.record(): ... return await metered_function() >>> task = asyncio.create_task(wrapper()) >>> # (later, elsewhere) >>> try: ... await task ... coarse_time = False ... except Exception: ... coarse_time = "failed" ... raise ... finally: ... # Metering stops here ... for label, value, unit in ledger.finalize(coarse_time): ... # actually log metrics r!r^r#rWz!list[tuple[Hashable, float, str]]metricsc8||_||_g|_yr)r^rWr)r)r^s rr,zDelayedMetricsLedger.__init__ts V  rc@|jj|||fyr)rr)r)rtrurvs r _callbackzDelayedMetricsLedger._callbackys UE401rNrmc#Ktj|j|5ddddy#1swYyxYww)aIngest metrics logged with :meth:`ContextMeter.digest_metric` or :meth:`ContextMeter.meter` and temporarily store them in :ivar:`metrics`. Parameters ---------- key: Hashable, optional See :meth:`ContextMeter.add_callback` rN)rirr)r)rms rrecordzDelayedMetricsLedger.record|s. ' 'C ' @ A @ @s">2 >;>c#K|j}||jz }|jD]"\}}}|dk7s|s|||f|dk(s|r||z}$|dur t||}|xsd|dfyw)a4The metered code is terminated, and we now know how to log it. Parameters ---------- coarse_time: str | False, optional False Yield all acquired metrics, plus an extra time metric, labelled "other", which is the time between creating the DelayedMetricsLedger and calling this method, minus any time logged in the metrics. label Yield all acquired non-time metrics. Yield a single metric, labelled , which is the time between creating the DelayedMetricsLedger and calling this method. floor: float | False, optional Floor either the "other" or the metric to this value (default: 0). Set to False to disable. rFotherN)r^rWrr])r) coarse_timer_rXr$rtrurvs rfinalizezDelayedMetricsLedger.finalizes,yy{tzz!"&,, E5$y UD((y  #/  u%E$WeY66sAA/ A/ #A/)r^r!r{)rmrrLr)Fr[)rzstr | Literal[False]r_rrLz%Iterator[tuple[Hashable, float, str]]) rNrOrPrQrRr.r/r,rrrrrrrrrSsm8  L ..3:3G3G 2/3  -2(+!7)!7&!7 / !7rr)r^r!r_rrLr)- __future__rr}r=sysrqr2r.collections.abcrrr contextlibr contextvarsr dataclassesr functoolsr mathr typingr rdistributed.compatibilityr namedtuplerrrrr version_info monotonic process_time thread_timeAttributeErrorOSErrorrVr/rarcrirrrrrs|" 88%"! -*K**+>C):$""45 !23//f s')  5 9 > >DW..TBGGI <