spdl.pipeline.StatsThreadOutputQueue

class StatsThreadOutputQueue(info: StageInfo, *, buffer_size: int = 1, interval: float = -1)

A thread sink with background statistics and overridable reporting hooks.

Select this class explicitly with use_thread_output_queue in build() or build_pipeline(). Foreground operations capture numeric observations; background batches update statistics. No storage backend is required. The default callbacks use Python logging; subclasses can instead keep packets in memory, display them, or send them to an application-owned destination. See ThreadOutputQueue for class selection and the subclass constructor contract.

Parameters:

• info – Stage identity assigned by the builder.

• buffer_size – Maximum queued items; zero permits unbounded buffering.

• interval – Positive seconds between interval callbacks. Nonpositive values disable interval reports, not collection or the final callback.

Example without logging or a storage backend:

from collections import deque
from spdl.pipeline import (
    AsyncQueue,
    PipelineBuilder,
    QueuePerfStats,
    StatsThreadOutputQueue,
)

packets = deque(maxlen=100)

class MyStatsSink(StatsThreadOutputQueue):
    async def interval_stats_callback(self, stats: QueuePerfStats) -> None:
        packets.append(stats)

    async def final_stats_callback(self, stats: QueuePerfStats) -> None:
        packets.append(stats)

pipeline = (
    PipelineBuilder()
    .add_source(range(100))
    .add_sink(8)
    .build(
        num_threads=2,
        queue_class=AsyncQueue,
        task_hook_factory=lambda _: [],
        use_thread_output_queue=MyStatsSink,
        report_stats_interval=5,
    )
)
with pipeline.auto_stop():
    for item in pipeline:
        print(item)

The example disables statistics on internal queues and task hooks; those choices are independent of the final sink’s callbacks. A nonpositive report_stats_interval disables interval reports, not collection or the final callback. Changing logger levels does not disable measurement. Select ThreadOutputQueue to avoid sink tracking entirely.

Callbacks run on the background event loop. Keep their work bounded and avoid blocking that loop. They must terminate; finalization waits for the final callback. Classes passed through subprocess configuration must be importable and picklable, like other pipeline configuration classes.

Reporting covers the producer-stage window. Items buffered when that stage finishes remain readable, but those later reads do not extend the final measurement window. num_items counts queue removals during that window, including cleanup drains, not necessarily batches consumed by a training loop. Overflow of bounded numeric capture marks statistics unavailable and suppresses invalid reports with a diagnostic instead of blocking data or silently biasing percentiles. Counter-allocation exhaustion also disables reporting. Unexpected aggregation errors propagate through the collector and stage finalization rather than silently discarding stage metrics.

This customization does not change stage execution or serialization. Measure the intended workload before choosing an implementation.

Benchmark comparison

The complete three-way comparison below measures foreground iterator handoff latency on a Linux CPU host with CPython 3.12.16. Lower is better. It compares use_thread_output_queue=False, ThreadOutputQueue and StatsThreadOutputQueue using the same source-to-sink workload: 500 integer items, buffer capacity 8, two configured executor workers and 20 warmup items. Timed and untimed reads alternate, giving 240 observations per cell. Each cell lists mean / p50 / p95 / p99, all in microseconds.

Foreground handoff latency (mean / p50 / p95 / p99, us)

Foreground work (ms)	False	ThreadOutputQueue	StatsThreadOutputQueue
0	142.6 / 142.2 / 290.7 / 340.8	87.5 / 95.7 / 190.1 / 251.2	106.3 / 102.7 / 220.2 / 286.5
3	152.2 / 170.3 / 352.5 / 472.1	119.9 / 128.5 / 303.1 / 368.3	133.9 / 128.1 / 328.9 / 478.0
6	53.4 / 11.9 / 242.2 / 357.9	91.3 / 102.4 / 251.6 / 375.8	89.7 / 13.8 / 314.6 / 380.3
9	51.5 / 19.1 / 288.1 / 340.5	62.8 / 10.2 / 208.5 / 308.6	42.1 / 11.3 / 261.8 / 402.8
12	23.4 / 21.2 / 42.6 / 87.9	14.8 / 9.6 / 22.3 / 175.3	22.7 / 9.9 / 177.2 / 248.5
15	22.6 / 20.6 / 34.6 / 46.4	12.7 / 11.8 / 21.0 / 28.7	16.5 / 14.9 / 33.8 / 39.9
18	23.4 / 20.5 / 39.1 / 87.6	14.5 / 12.9 / 22.6 / 58.9	19.6 / 17.6 / 34.3 / 59.9
21	24.3 / 21.9 / 40.8 / 57.0	14.5 / 14.0 / 22.1 / 33.1	19.1 / 17.5 / 34.4 / 49.6
24	25.7 / 22.9 / 49.5 / 63.1	17.2 / 15.4 / 32.7 / 41.6	17.7 / 17.1 / 28.2 / 40.0
27	25.7 / 22.2 / 55.2 / 73.1	15.2 / 14.3 / 26.5 / 39.9	19.7 / 17.6 / 39.1 / 64.6
30	36.9 / 31.3 / 64.7 / 91.1	32.3 / 29.2 / 64.4 / 86.6	33.0 / 30.5 / 58.3 / 69.7

Results are one full sweep, in the listed variant order at each work setting, not interleaved repeated trials. The foreground work is a CPU busy loop that contends for the GIL, not GPU work. The timed operation is advancing a newly created pipeline iterator, not a bare queue call. The False control uses the built-in StatsQueue/sync-mirror sink; both thread classes are selected explicitly. Internal queues/task hooks use ordinary defaults in all variants. The statistics class retains collection and its default final reporting callback; periodic reporting is disabled. P99 has few tail samples, and these lightweight results do not establish sustained-load, GPU-overlap or training performance.

Reproduce the sweep with spdl.examples.benchmark_thread_output_queue:

python -m spdl.examples.benchmark_thread_output_queue

Added in version 0.9.0.

Methods

empty()	Return whether the underlying thread buffer is empty.
final_stats_callback(stats)	Handle the final producer-stage packet after pending samples are flushed.
full()	Return whether the underlying thread buffer is full.
get()	Wait for an item through an executor without blocking the event loop.
get_nowait()	Return a queued value without waiting, raising queue.Empty if unavailable.
interval_stats_callback(stats)	Handle one interval packet on the background loop.
join()	Wait for every queued item to be acknowledged with task_done.
put(item)	Enqueue an item, delegating full or unbounded queues to an executor.
put_nowait(item)	Enqueue without waiting, raising queue.Full when capacity is exhausted.
qsize()	Return the number of queued items.
stage_hook()	Own collection and reporting through stage completion and cancellation.
task_done()	Acknowledge one previously enqueued item for join accounting.

Attributes

maxsize

The configured item-capacity limit; zero means unbounded.

async final_stats_callback(stats: QueuePerfStats) → None

Handle the final producer-stage packet after pending samples are flushed.

Parameters:

stats – Completed cumulative stage-window statistics. The default logs them; override this method to retain or consume them.

async interval_stats_callback(stats: QueuePerfStats) → None

Handle one interval packet on the background loop.

Parameters:

stats – Completed interval statistics. The default logs them; override this method to choose a different destination without storage dependencies in the queue implementation.

stage_hook() → AsyncIterator[None]

Own collection and reporting through stage completion and cancellation.