Benchmark thread output queue¶
Benchmark: SPDL pipeline handoff latency with and without thread output queue.
Compares main-thread get_item latency between the default asyncio-based
handoff (run_coroutine_threadsafe + Future.result polling) and the
queue.Queue-based handoff (direct queue.Queue.get).
The benchmark emulates a realistic training loop: each iteration the
foreground thread receives a batch, does simulated work (busy-wait to
mimic a GPU forward/backward pass), and then times how long the next
get_item call takes. When the simulated work is long enough the
producer has time to pre-fill the queue, so the handoff latency
isolates the cross-thread scheduling overhead.
Foreground work is swept from 0ms to 30ms in 3ms steps.
Usage:
buck2 run //spdl/examples:benchmark_thread_output_queue
Example results (devserver, no GPU, 500 lightweight int items):
FG work default (p50) default (p99) thread_q (p50) thread_q (p99)
------- ------------- ------------- -------------- --------------
0ms 199us 625us 116us 313us
3ms 246us 642us 223us 646us
6ms 232us 532us 190us 555us
9ms 287us 485us 153us 549us
12ms 280us 822us 14us 464us
15ms 221us 396us 8us 385us
18ms 227us 431us 9us 70us
21ms 224us 450us 11us 41us
24ms 240us 533us 12us 34us
27ms 227us 470us 12us 27us
30ms 242us 512us 12us 26us
The default asyncio path stays flat at ~220-280us regardless of overlap
time — that is the fixed run_coroutine_threadsafe scheduling tax.
The thread output queue drops to ~8-14us once the foreground work is
long enough (>= ~12ms on this machine) for the producer to pre-fill it.
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"""Benchmark: SPDL pipeline handoff latency with and without thread output queue.
9
10Compares main-thread ``get_item`` latency between the default asyncio-based
11handoff (``run_coroutine_threadsafe`` + ``Future.result`` polling) and the
12``queue.Queue``-based handoff (direct ``queue.Queue.get``).
13
14The benchmark emulates a realistic training loop: each iteration the
15foreground thread receives a batch, does simulated work (busy-wait to
16mimic a GPU forward/backward pass), and then times how long the *next*
17``get_item`` call takes. When the simulated work is long enough the
18producer has time to pre-fill the queue, so the handoff latency
19isolates the cross-thread scheduling overhead.
20
21Foreground work is swept from 0ms to 30ms in 3ms steps.
22
23Usage::
24
25 buck2 run //spdl/examples:benchmark_thread_output_queue
26
27Example results (devserver, no GPU, 500 lightweight int items)::
28
29 FG work default (p50) default (p99) thread_q (p50) thread_q (p99)
30 ------- ------------- ------------- -------------- --------------
31 0ms 199us 625us 116us 313us
32 3ms 246us 642us 223us 646us
33 6ms 232us 532us 190us 555us
34 9ms 287us 485us 153us 549us
35 12ms 280us 822us 14us 464us
36 15ms 221us 396us 8us 385us
37 18ms 227us 431us 9us 70us
38 21ms 224us 450us 11us 41us
39 24ms 240us 533us 12us 34us
40 27ms 227us 470us 12us 27us
41 30ms 242us 512us 12us 26us
42
43The default asyncio path stays flat at ~220-280us regardless of overlap
44time — that is the fixed ``run_coroutine_threadsafe`` scheduling tax.
45The thread output queue drops to ~8-14us once the foreground work is
46long enough (>= ~12ms on this machine) for the producer to pre-fill it.
47"""
48
49from __future__ import annotations
50
51import statistics
52import time
53from dataclasses import dataclass, field
54
55from spdl.pipeline import PipelineBuilder
56
57__all__ = ["BenchResult", "main"]
58
59
60@dataclass
61class BenchResult:
62 """Collected latency measurements for a single benchmark run."""
63
64 name: str
65 latencies_us: list[float] = field(default_factory=list)
66
67 @property
68 def p50(self) -> float:
69 s = sorted(self.latencies_us)
70 return s[len(s) // 2]
71
72 @property
73 def p95(self) -> float:
74 s = sorted(self.latencies_us)
75 return s[int(len(s) * 0.95)]
76
77 @property
78 def p99(self) -> float:
79 s = sorted(self.latencies_us)
80 return s[int(len(s) * 0.99)]
81
82 @property
83 def mean(self) -> float:
84 return statistics.mean(self.latencies_us)
85
86 def summary(self) -> str:
87 """Return a one-line summary string with mean/p50/p95/p99."""
88 return (
89 f"{self.name:<50s} "
90 f"mean={self.mean:>9.1f}us "
91 f"p50={self.p50:>9.1f}us "
92 f"p95={self.p95:>9.1f}us "
93 f"p99={self.p99:>9.1f}us "
94 f"({len(self.latencies_us)} meas)"
95 )
96
97
98def _busy_wait_us(us: float) -> None:
99 """Busy-wait for *us* microseconds (spin-loop)."""
100 deadline = time.perf_counter() + us / 1e6
101 while time.perf_counter() < deadline:
102 pass
103
104
105def _run_bench(
106 name: str,
107 n_items: int,
108 buffer_size: int,
109 use_thread_output_queue: bool,
110 work_us: float,
111 warmup: int = 20,
112) -> BenchResult:
113 """Run a single benchmark: source -> sink pipeline with foreground work.
114
115 Builds a trivial pipeline (source of *n_items* integers -> sink) and
116 iterates it. Between each consumed item the foreground thread
117 busy-waits for *work_us* microseconds to simulate consumer-side
118 compute (e.g. a GPU training step). After warmup, the time spent
119 inside each ``get_item`` call is recorded.
120
121 Args:
122 name: Human-readable label for the result.
123 n_items: Total number of items the source produces.
124 buffer_size: Sink buffer size (number of slots).
125 use_thread_output_queue: Whether to use the thread output queue handoff.
126 work_us: Duration of simulated foreground work in microseconds.
127 warmup: Number of items to consume before recording latencies.
128 """
129 items = list(range(n_items))
130 pipeline = (
131 PipelineBuilder()
132 .add_source(iter(items))
133 .add_sink(buffer_size=buffer_size)
134 .build(num_threads=2, use_thread_output_queue=use_thread_output_queue)
135 )
136
137 result = BenchResult(name=name)
138 consumed = 0
139 for _ in pipeline.get_iterator(timeout=60):
140 consumed += 1
141
142 # Simulate foreground work (e.g. GPU fwd+bwd).
143 # This gives the producer time to pre-fill the queue
144 # so the next get_item measures pure handoff overhead.
145 _busy_wait_us(work_us)
146
147 if consumed <= warmup:
148 continue
149
150 # Time the next get_item call — this is what we're measuring.
151 t0 = time.perf_counter()
152 try:
153 next(pipeline.get_iterator(timeout=60))
154 except StopIteration:
155 break
156 result.latencies_us.append((time.perf_counter() - t0) * 1e6)
157 consumed += 1
158
159 # Do work after the timed item too, so the *next* iteration's
160 # measurement also has overlap time.
161 _busy_wait_us(work_us)
162
163 return result
164
165
166def main() -> None:
167 """Run the full benchmark sweep and print results."""
168 n_items = 500
169 buffer_size = 8
170
171 print("SPDL Pipeline Thread Output Queue Handoff Benchmark")
172 print("=" * 90)
173
174 results: list[BenchResult] = []
175
176 work_ms_values = list(range(0, 31, 3))
177 for work_ms in work_ms_values:
178 work_us = work_ms * 1000.0
179 label = f"{work_ms}ms foreground work"
180 print(f"\n--- {label} ({n_items} items, buffer_size={buffer_size}) ---")
181
182 for use_toq, tag in [
183 (False, "default asyncio"),
184 (True, "thread output queue"),
185 ]:
186 r = _run_bench(
187 f"{tag}, {label}",
188 n_items,
189 buffer_size,
190 use_thread_output_queue=use_toq,
191 work_us=work_us,
192 )
193 results.append(r)
194 print(f" {r.summary()}")
195
196 print(f"\n{'=' * 90}")
197 print("SUMMARY — main thread get_item() latency (lower is better)")
198 print(f"{'=' * 90}")
199 for r in results:
200 if r.latencies_us:
201 print(f" {r.summary()}")
202
203
204if __name__ == "__main__":
205 main()
API Reference¶
Functions
Classes