caproto.server.stats.MemoryTracingHelper

class caproto.server.stats.MemoryTracingHelper(*args, **kwargs)

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

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)

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)