caproto.server.stats.MemoryTracingHelper

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

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
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
            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)