Benchmark thread output queue¶
Compare handoff with False, ThreadOutputQueue and StatsThreadOutputQueue.
Runs a source-to-sink pipeline with 500 integer items, buffer capacity 8 and two configured executor workers. Foreground CPU busy-work is swept from 0 to 30 ms in 3 ms steps. After 20 warmup items, timed and untimed iterator reads alternate, producing 240 latency observations per variant and work setting.
The busy loop creates CPU/GIL contention rather than real GPU work. Results measure iterator handoff, not sustained-load, training or GPU-overlap performance. The False control uses the configured async sink (a synchronous mirror for the built-in StatsQueue); both thread variants are explicitly selected classes. StatsThreadOutputQueue collects with its default final reporting callback; periodic reporting is disabled. Existing internal queues and task hooks retain normal defaults in every variant.
Usage:
buck2 run //spdl/examples:benchmark_thread_output_queue
The complete measured comparison is documented on
spdl.pipeline.ThreadOutputQueue and
spdl.pipeline.StatsThreadOutputQueue.
Source¶
Source
Click here to see the source.
1# Copyright (c) Meta Platforms, Inc. and affiliates.
2# All rights reserved.
3#
4# This source code is licensed under the BSD-style license found in the
5# LICENSE file in the root directory of this source tree.
6
7
8"""Compare handoff with False, ThreadOutputQueue and StatsThreadOutputQueue.
9
10Runs a source-to-sink pipeline with 500 integer items, buffer capacity 8 and two
11configured executor workers. Foreground CPU busy-work is swept from 0 to 30 ms
12in 3 ms steps. After 20 warmup items, timed and untimed iterator reads alternate,
13producing 240 latency observations per variant and work setting.
14
15The busy loop creates CPU/GIL contention rather than real GPU work. Results
16measure iterator handoff, not sustained-load, training or GPU-overlap performance.
17The False control uses the configured async sink (a synchronous mirror for the
18built-in StatsQueue); both thread variants are explicitly selected classes.
19StatsThreadOutputQueue collects with its default final reporting callback;
20periodic reporting is disabled. Existing internal queues and task hooks retain
21normal defaults in every variant.
22
23Usage::
24
25 buck2 run //spdl/examples:benchmark_thread_output_queue
26
27The complete measured comparison is documented on
28:py:class:`spdl.pipeline.ThreadOutputQueue` and
29:py:class:`spdl.pipeline.StatsThreadOutputQueue`.
30"""
31
32from __future__ import annotations
33
34import statistics
35import time
36from dataclasses import dataclass, field
37
38from spdl.pipeline import PipelineBuilder, StatsThreadOutputQueue, ThreadOutputQueue
39
40__all__ = ["BenchResult", "run_benchmark", "main"]
41
42
43@dataclass
44class BenchResult:
45 """Collected latency measurements for a single benchmark run."""
46
47 name: str
48 latencies_us: list[float] = field(default_factory=list)
49
50 @property
51 def p50(self) -> float:
52 s = sorted(self.latencies_us)
53 return s[len(s) // 2]
54
55 @property
56 def p95(self) -> float:
57 s = sorted(self.latencies_us)
58 return s[int(len(s) * 0.95)]
59
60 @property
61 def p99(self) -> float:
62 s = sorted(self.latencies_us)
63 return s[int(len(s) * 0.99)]
64
65 @property
66 def mean(self) -> float:
67 return statistics.mean(self.latencies_us)
68
69 def summary(self) -> str:
70 """Return a one-line summary string with mean/p50/p95/p99."""
71 return (
72 f"{self.name:<50s} "
73 f"mean={self.mean:>9.1f}us "
74 f"p50={self.p50:>9.1f}us "
75 f"p95={self.p95:>9.1f}us "
76 f"p99={self.p99:>9.1f}us "
77 f"({len(self.latencies_us)} meas)"
78 )
79
80
81def _busy_wait_us(us: float) -> None:
82 """Busy-wait for *us* microseconds (spin-loop)."""
83 deadline = time.perf_counter() + us / 1e6
84 while time.perf_counter() < deadline:
85 pass
86
87
88def run_benchmark(
89 name: str,
90 n_items: int,
91 buffer_size: int,
92 use_thread_output_queue: bool | type[ThreadOutputQueue],
93 work_us: float,
94 warmup: int = 20,
95) -> BenchResult:
96 """Run a source-to-sink pipeline with foreground CPU busy-work.
97
98 After warmup, every other iterator advance is timed. The public build option
99 selects the configured async sink or an explicit thread-output class.
100
101 Args:
102 name: Human-readable label for the result.
103 n_items: Total number of integer items the source produces.
104 buffer_size: Sink buffer size in items; zero means unbounded.
105 use_thread_output_queue: ``False``, ``True``, or a thread-output queue class.
106 work_us: Duration of foreground CPU busy-work in microseconds.
107 warmup: Number of items to consume before recording latencies.
108
109 Returns:
110 Latencies of timed iterator advances, in microseconds. The workload
111 alternates timed and untimed reads after warmup; it is CPU-only and does
112 not establish sustained-load or GPU-overlap performance.
113 """
114 if n_items <= warmup + 1 or warmup < 0:
115 raise ValueError("n_items must exceed warmup by at least two")
116 if buffer_size < 0 or work_us < 0:
117 raise ValueError("buffer_size and work_us must be nonnegative")
118 items = list(range(n_items))
119 pipeline = (
120 PipelineBuilder()
121 .add_source(iter(items))
122 .add_sink(buffer_size=buffer_size)
123 .build(num_threads=2, use_thread_output_queue=use_thread_output_queue)
124 )
125
126 result = BenchResult(name=name)
127 consumed = 0
128 try:
129 for _ in pipeline.get_iterator(timeout=60):
130 consumed += 1
131 _busy_wait_us(work_us)
132 if consumed <= warmup:
133 continue
134
135 t0 = time.perf_counter()
136 try:
137 next(pipeline.get_iterator(timeout=60))
138 except StopIteration:
139 break
140 result.latencies_us.append((time.perf_counter() - t0) * 1e6)
141 consumed += 1
142 _busy_wait_us(work_us)
143 finally:
144 pipeline.stop(timeout=60)
145 return result
146
147
148def main() -> None:
149 """Run the three-way sweep and print every mean/p50/p95/p99 result."""
150 n_items = 500
151 buffer_size = 8
152
153 print("SPDL Pipeline Thread Output Queue Handoff Benchmark")
154 print("=" * 90)
155
156 results: list[BenchResult] = []
157 for work_ms in range(0, 31, 3):
158 label = f"{work_ms}ms foreground work"
159 print(f"\n--- {label} ({n_items} items, buffer_size={buffer_size}) ---")
160 for selection, tag in [
161 (False, "False"),
162 (ThreadOutputQueue, "ThreadOutputQueue"),
163 (StatsThreadOutputQueue, "StatsThreadOutputQueue"),
164 ]:
165 result = run_benchmark(
166 f"{tag}, {label}",
167 n_items,
168 buffer_size,
169 use_thread_output_queue=selection,
170 work_us=work_ms * 1000.0,
171 )
172 results.append(result)
173 print(f" {result.summary()}")
174
175 print(f"\n{'=' * 90}")
176 print("SUMMARY — foreground iterator handoff latency (lower is better)")
177 print(f"{'=' * 90}")
178 for result in results:
179 print(f" {result.summary()}")
180
181
182if __name__ == "__main__":
183 main()
API Reference¶
Functions
- run_benchmark(name: str, n_items: int, buffer_size: int, use_thread_output_queue: bool | type[ThreadOutputQueue], work_us: float, warmup: int = 20) BenchResult[source]¶
Run a source-to-sink pipeline with foreground CPU busy-work.
After warmup, every other iterator advance is timed. The public build option selects the configured async sink or an explicit thread-output class.
- Parameters:
name – Human-readable label for the result.
n_items – Total number of integer items the source produces.
buffer_size – Sink buffer size in items; zero means unbounded.
use_thread_output_queue –
False,True, or a thread-output queue class.work_us – Duration of foreground CPU busy-work in microseconds.
warmup – Number of items to consume before recording latencies.
- Returns:
Latencies of timed iterator advances, in microseconds. The workload alternates timed and untimed reads after warmup; it is CPU-only and does not establish sustained-load or GPU-overlap performance.
Classes