stormlog.cpu_profiler

CPU-only memory profiler and tracker.

Classes

CPUMemoryProfiler()

Lightweight CPU memory profiler mirroring the GPU API.

CPUMemorySnapshot(timestamp, rss, vms, ...)

Point-in-time CPU memory snapshot.

CPUMemoryTracker([sampling_interval, ...])

CPU tracker offering a superset of the GPU tracker interface.

CPUProfileResult(name, duration, ...)

Results from profiling a CPU function/context.
class stormlog.cpu_profiler.CPUMemorySnapshot(timestamp, rss, vms, cpu_percent)[source]

Bases: object

Point-in-time CPU memory snapshot.

Parameters:

  • timestamp (float)

  • rss (int)

  • vms (int)

  • cpu_percent (float)

timestamp: float
rss: int
vms: int
cpu_percent: float
to_dict()[source]
Return type:

Dict[str, Any]

class stormlog.cpu_profiler.CPUProfileResult(name, duration, snapshot_before, snapshot_after, peak_rss)[source]

Bases: object

Results from profiling a CPU function/context.

Parameters:
name: str
duration: float
snapshot_before: CPUMemorySnapshot
snapshot_after: CPUMemorySnapshot
peak_rss: int
memory_diff()[source]
Return type:

int

to_dict()[source]
Return type:

Dict[str, Any]

class stormlog.cpu_profiler.CPUMemoryProfiler[source]

Bases: object

Lightweight CPU memory profiler mirroring the GPU API.

start_monitoring(interval=0.1)[source]
Parameters:

interval (float)

Return type:

None

stop_monitoring()[source]
Return type:

None

profile_function(func, *args, **kwargs)[source]
Parameters:

  • func (Callable[[...], Any])

  • args (Any)

  • kwargs (Any)

Return type:

CPUProfileResult

profile_context(name='context')[source]
Parameters:

name (str)

Return type:

Any

clear_results()[source]
Return type:

None

get_summary()[source]
Return type:

Dict[str, Any]

class stormlog.cpu_profiler.CPUMemoryTracker(sampling_interval=0.5, max_events=10000, enable_alerts=True, job_id=None, rank=None, local_rank=None, world_size=None, telemetry_sink_config=None)[source]

Bases: object

CPU tracker offering a superset of the GPU tracker interface.

Parameters:

  • sampling_interval (float)

  • max_events (int)

  • enable_alerts (bool)

  • job_id (Optional[str])

  • rank (Optional[int])

  • local_rank (Optional[int])

  • world_size (Optional[int])

  • telemetry_sink_config (Optional[TelemetrySinkConfig])

get_session_summary()[source]
Return type:

SessionSummary | None

start_tracking()[source]
Return type:

None

stop_tracking()[source]
Return type:

None

enter_phase(name, *, metadata=None)[source]

Enter one structured CPU tracking phase.

Parameters:

  • name (str)

  • metadata (Dict[str, Any] | None)

Return type:

PhaseHandle

phase(name, *, metadata=None)[source]

Context manager that emits structured CPU phase telemetry.

Parameters:

  • name (str)

  • metadata (Dict[str, Any] | None)

Return type:

Any

get_events(event_type=None, last_n=None, since=None)[source]

Get tracking events with optional filtering.

Parameters:

  • event_type (str | None) – Filter by event type

  • last_n (int | None) – Get last N events

  • since (float | None) – Get events since timestamp

Returns:

List of filtered events

Return type:

List[TrackingEvent]

get_statistics()[source]
Return type:

Dict[str, Any]

get_memory_timeline(interval=1.0)[source]
Parameters:

interval (float)

Return type:

Dict[str, List[float]]

clear_events()[source]
Return type:

None

export_events(filename, format='csv')[source]
Parameters:

  • filename (str)

  • format (str)

Return type:

None

export_events_with_timestamp(directory, format)[source]
Parameters:

  • directory (str)

  • format (str)

Return type:

str