scope_profiler.concurrency
Thread, asyncio-task and greenlet identity for the region profiler.
A region records a (start, end) pair and nothing else, which is enough
only while one call stack owns the process. With several threads - or several
coroutines interleaved on one thread - the recorded intervals of two lanes
overlap without nesting, and the whole downstream stack (flame chart,
exclusive time, .prof export) has no call graph to reconstruct. See
:class:~scope_profiler.call_stack.NestingError.
This module supplies the missing axis: every recorded call is stamped with the lane it ran on, and the lanes themselves are described in two small tables that travel with the run.
Threads. :class:ThreadRecord carries a dense per-rank index, the OS
identifiers, the thread’s name, when it started and ended, and the CPU time it
burned. The index is what the event column stores; everything else is looked
up once, per thread, not per call.
Tasks. An asyncio task is a lane of its own: it runs on a thread, but so
do the other tasks that interleave with it, and a with block held across
an await covers wall time the task did not spend running. Instrumenting
the task - rather than the loop - gives both the identity and the split:
:class:TaskRecord accumulates running and suspended nanoseconds, and a
region reads the suspended counter at entry and exit to report the await time
of that one call. Greenlets are the same shape (cooperative lanes on one
thread) and share :class:TaskRecord, tagged kind="greenlet".
Nothing here is installed unless track_threads is set: the profiler’s
default single-lane path is untouched, hot loop included.
Attributes
| Name | Description |
|---|---|
CPU_SAMPLE_MASK | No description. |
NO_TASK | No description. |
UNKNOWN | No description. |
Classes
| Name | Description |
|---|---|
ConcurrencyTracker | Per-run registry of threads, asyncio tasks and greenlets. |
TaskInfo | One row of a run's task table: an asyncio task or a greenlet. |
TaskRecord | One asyncio task or greenlet, and how its wall time split. |
ThreadInfo | One row of a run's thread table, as post-processing sees it. |
ThreadRecord | One thread of the profiled process, as the lane tables describe it. |
Functions
| Name | Description |
|---|---|
lane_ids | One id per call identifying the stack it belongs to. |
lane_tables_from_columns | Turn one rank's stored lane columns into :class:ThreadInfo/:class:TaskInfo. |
CPU_SAMPLE_MASKattributemodule attribute#
CPU_SAMPLE_MASK = 255NO_TASKattributemodule attribute#
NO_TASK = -1UNKNOWNattributemodule attribute#
UNKNOWN = -1ConcurrencyTrackerclass#
class ConcurrencyTracker(track_async: bool = False)Per-run registry of threads, asyncio tasks and greenlets.
One tracker belongs to one :class:~scope_profiler.profile_config.ProfilingConfig,
so two simultaneously active managers never share lane numbering. Creating
it is free; :meth:install is what puts the (few, cheap) hooks in place,
and :meth:uninstall takes them out again.
track_asyncattributeinstance attribute#
track_async = bool(track_async)tasksproperty#
tasks: list[TaskRecord]Every instrumented task and greenlet, in creation order.
threadsproperty#
threads: list[ThreadRecord]Every thread that recorded a call, in registration order.
current_threadmethod#
def current_thread() -> ThreadRecordThis thread’s record, registering it on first use.
On the region hot path. The common case is one failed-then-cached
attribute lookup on a :class:threading.local, which is why the
registry itself is never consulted here.
installmethod#
def install() -> NonePut the thread-start, asyncio and greenlet hooks in place.
Idempotent, so a second setup() on the same configuration does not
stack wrappers.
instrument_loopmethod#
def instrument_loop(loop) -> NoneRoute loop’s task creation through the step timer.
Public because a loop implementation that never reaches
BaseEventLoop.run_forever (uvloop, say) can be handed here
directly. Chains onto whatever task factory the application already
installed instead of replacing it.
Instrumented once per loop, and keyed on the loop rather than on whichever factory is currently in place: an application that installs its own factory after the loop has started displaces ours, and re-wrapping to get back on top would make the two call each other forever, since its captured “previous” is us. Such a loop keeps recording regions and thread ids; the tasks created after the swap simply have no lane of their own.
snapshotmethod#
def snapshot(origin_ns: int = 0) -> dictThe lane tables, as columns, ready to be written or returned.
Timestamps are shifted by origin_ns (the run’s start), matching
what the event columns store, so a thread’s lifetime and the calls on
it share one timeline. :data:UNKNOWN entries are left alone.
Returns
dict
{"threads": {...}, "tasks": {...}}, each a dict of equal-length
numpy arrays. Empty arrays when nothing was tracked.
uninstallmethod#
def uninstall() -> NoneRemove every hook and restore what was there before.
TaskInfoclass#
class TaskInfo(index: int, rank: int, kind: str, name: str, coro_name: str, thread_index: int, created_ns: int, done_ns: int, steps: int, running_ns: int, suspended_ns: int)One row of a run’s task table: an asyncio task or a greenlet.
running_time and awaiting_time split the lane’s wall time into the
part it held its thread and the part it did not, which is the number
await-heavy code is usually looking for.
awaiting_timeattributeinstance attribute#
awaiting_time = float(suspended_ns) / 1000000000.0coro_nameattributeinstance attribute#
coro_name = str(coro_name)created_timeattributeinstance attribute#
created_time = float(created_ns) / 1000000000.0done_timeattributeinstance attribute#
done_time = None if int(done_ns) == UNKNOWN else float(done_ns) / 1000000000.0indexattributeinstance attribute#
index = int(index)kindattributeinstance attribute#
kind = str(kind)nameattributeinstance attribute#
name = str(name)rankattributeinstance attribute#
rank = int(rank)running_timeattributeinstance attribute#
running_time = float(running_ns) / 1000000000.0stepsattributeinstance attribute#
steps = int(steps)thread_indexattributeinstance attribute#
thread_index = int(thread_index)TaskRecordclass#
class TaskRecord(index: int, kind: str, name: str, coro_name: str, thread_index: int)One asyncio task or greenlet, and how its wall time split.
running_ns is the time the lane actually held its thread, summed over
every step; suspended_ns is the time between steps - awaiting, or
switched away from. A region entered inside the lane reads
suspended_ns on the way in and out, and the difference is that one
call’s await time.
coro_nameattributeinstance attribute#
coro_name = coro_namecreated_nsattributeinstance attribute#
created_ns = perf_counter_ns()done_nsattributeinstance attribute#
done_ns = UNKNOWNindexattributeinstance attribute#
index = indexkindattributeinstance attribute#
kind = kindlast_step_end_nsattributeinstance attribute#
last_step_end_ns = 0nameattributeinstance attribute#
name = namerunning_nsattributeinstance attribute#
running_ns = 0step_started_nsattributeinstance attribute#
step_started_ns = 0stepsattributeinstance attribute#
steps = 0suspended_nsattributeinstance attribute#
suspended_ns = 0thread_indexattributeinstance attribute#
thread_index = thread_indexThreadInfoclass#
class ThreadInfo(index: int, rank: int, name: str, ident: int, native_id: int, daemon: bool, start_ns: int, end_ns: int, cpu_ns: int)One row of a run’s thread table, as post-processing sees it.
The read-back counterpart of :class:ThreadRecord: same fields, but
durations in seconds and timestamps relative to the start of the run,
matching every other time reported by
:class:~scope_profiler.results.ProfilingResults.
cpu_timeattributeinstance attribute#
cpu_time = float(cpu_ns) / 1000000000.0daemonattributeinstance attribute#
daemon = bool(daemon)end_timeattributeinstance attribute#
end_time = None if int(end_ns) == UNKNOWN else float(end_ns) / 1000000000.0identattributeinstance attribute#
ident = int(ident)indexattributeinstance attribute#
index = int(index)nameattributeinstance attribute#
name = str(name)native_idattributeinstance attribute#
native_id = int(native_id)rankattributeinstance attribute#
rank = int(rank)start_timeattributeinstance attribute#
start_time = float(start_ns) / 1000000000.0ThreadRecordclass#
class ThreadRecord(index: int, thread)One thread of the profiled process, as the lane tables describe it.
index is dense within a rank and is what the per-call thread_ids
column stores. cpu_ns is exact for a thread that has ended and a
sampled lower bound for one still running; see :data:CPU_SAMPLE_MASK.
cpu_nsattributeinstance attribute#
cpu_ns = thread_time_ns()daemonattributeinstance attribute#
daemon = bool(thread.daemon)end_nsattributeinstance attribute#
end_ns = UNKNOWNidentattributeinstance attribute#
ident = int(thread.ident or 0)indexattributeinstance attribute#
index = indexnameattributeinstance attribute#
name = str(thread.name)native_idattributeinstance attribute#
native_id = int(getattr(thread, 'native_id', 0) or 0)start_nsattributeinstance attribute#
start_ns = perf_counter_ns()taskattributeinstance attribute#
task: TaskRecord | None = Nonelane_idsfunction#
def lane_ids(thread_ids: np.ndarray, task_ids: np.ndarray | None) -> np.ndarrayOne id per call identifying the stack it belongs to.
A call belongs to its task when it ran inside one and to its bare thread
otherwise. The two id spaces are disjoint by construction: thread lanes
are mapped to -2 - thread, task lanes stay non-negative, and -1 is
left for a call whose thread is unknown - a region recorded by the
Fortran API, say, folded into an otherwise thread-aware run. This is the
key :func:~scope_profiler.call_stack.build_call_arrays groups by, since
only calls sharing a lane can nest.
lane_tables_from_columnsfunction#
def lane_tables_from_columns(rank: int, tables: dict) -> tuple[list, list]Turn one rank’s stored lane columns into :class:ThreadInfo/:class:TaskInfo.
The one place the column layout written to HDF5 and returned in memory is
turned back into objects, so the file and finalize(return_results=True)
cannot disagree about what a run’s lanes were.