spdl.pipeline.ThreadOutputQueue

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

A thread-backed sink with direct foreground reads and no statistics.

Pass this class, or a subclass, as use_thread_output_queue to build() or build_pipeline(). An omitted or None selector uses ThreadOutputQueueConfig; its initial configuration selects the stats-enabled thread sink. False explicitly keeps the async sink, and True selects the configured thread-output class. An explicit class takes priority over the configuration. Pass a class, not a prebuilt instance; the builder constructs the sink.

Internal stage queues still use queue_class. This plain sink does not collect statistics, require logging, or persist data. For optional background measurements and custom reporting, use StatsThreadOutputQueue.

Subclasses must accept the constructor keywords below and preserve bounded queue and lifecycle semantics. Override stage_hook to customize initialization/finalization. Classes passed through subprocess configuration must be importable and picklable, like other pipeline configuration classes.

Parameters:

• info – Stage identity assigned by the builder.

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

• interval – Reporting interval supplied by the builder. This plain base ignores it; statistics-enabled subclasses may use it.

Example:

from spdl.pipeline import PipelineBuilder, ThreadOutputQueue

pipeline = (
    PipelineBuilder()
    .add_source(range(100))
    .add_sink(8)
    .build(num_threads=2, use_thread_output_queue=ThreadOutputQueue)
)
with pipeline.auto_stop():
    for item in pipeline:
        print(item)

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.
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.
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()	Context manager, which handles init/final logic for the stage.
task_done()	Acknowledge one previously enqueued item for join accounting.

Attributes

maxsize

The configured item-capacity limit; zero means unbounded.

empty() → bool

Return whether the underlying thread buffer is empty.

full() → bool

Return whether the underlying thread buffer is full.

async get() → object

Wait for an item through an executor without blocking the event loop.

Returns:

The next queued value.

get_nowait() → object

Return a queued value without waiting, raising queue.Empty if unavailable.

Returns:

The next queued value.

async join() → None

Wait for every queued item to be acknowledged with task_done.

property maxsize: int

The configured item-capacity limit; zero means unbounded.

async put(item: object) → None

Enqueue an item, delegating full or unbounded queues to an executor.

Parameters:

item – The value to enqueue. A bounded queue with available capacity accepts it directly without an executor round trip.

put_nowait(item: object) → None

Enqueue without waiting, raising queue.Full when capacity is exhausted.

Parameters:

item – The value to enqueue.

qsize() → int

Return the number of queued items.

task_done() → None

Acknowledge one previously enqueued item for join accounting.