caproto.server.stats.MemoryTracingHelper

class caproto.server.stats.MemoryTracingHelper(*args, filters: Optional[List[tracemalloc.Filter]] = None, **kwargs)

A helper which quickly allows for tracing memory usage and allocations on a caproto server instance.

Parameters

prefixstr

Prefix for all PVs in the group

macrosdict, optional

Dictionary of macro name to value

parentPVGroup, optional

Parent PVGroup

namestr, optional

Name for the group, defaults to the class name

statesdict, optional

A dictionary of states used for channel filtering. See https://epics.anl.gov/base/R3-15/5-docs/filters.html

filterslist of tracemalloc.Filter, optional

Filters to apply to the snapshot. Defaults to TRACEMALLOC_FILTERS.

MemoryTracingHelper pvproperties

Attribute	Suffix	Docs	Type	Notes	Alarm Group
diff_results	TraceDiffResults	Trace diff from snapshot to snapshot	str (waveform)	Read-only Length(20000)
enable_tracing	EnableTracing	Enable/disable in-depth memory analysis	ENUM (bo)	Startup
top_allocations	TraceTopAllocations	Top allocations in snapshot	str (waveform)	Read-only Length(20000)
trace_count	TraceCount	Number of top allocations to view	int (ao)

Methods

group_read(instance)	Generic read called for channels without get defined
group_write(instance, value)	Generic write called for channels without put defined

Attributes

default_values
type_map
type_map_read_only

pvproperty methods

enable_tracing.startup(self, instance, async_lib)

Source code: enable_tracing.startup

            async def scanned_startup(group, prop, async_lib):
                if use_scan_field and period is not None:
                    if prop.field_inst.scan_rate_sec is None:
                        # This is a hook to allow setting of the default scan
                        # rate through the 'period' argument of the decorator.
                        prop.field_inst._scan_rate_sec = period
                        # TODO: update .SCAN to reflect this number

                sleep = async_lib.library.sleep
                while True:
                    t0 = time.monotonic()
                    if use_scan_field:
                        iter_time = prop.field_inst.scan_rate_sec
                        if iter_time is None:
                            iter_time = 0
                    else:
                        iter_time = period

                    if iter_time > 0:
                        await call_scan_function(group, prop, async_lib)
                    else:
                        iter_time = 0.1
                        # TODO: could the scan rate - or values in general -
                        # have events tied with them so busy loops are
                        # unnecessary?
                    elapsed = time.monotonic() - t0
                    sleep_time = (max(0, iter_time - elapsed)
                                  if subtract_elapsed
                                  else iter_time)
                    await sleep(sleep_time)