1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
//! Multi-threaded scaling benchmarks
//!
//! Tests monocoque's ability to scale horizontally across multiple CPU cores.
//! Each thread runs its own DEALER socket, measuring aggregate throughput.
//!
//! ## Architecture
//!
//! - Lock-free design: Each socket has its own `io_uring` context
//! - No shared mutable state in hot paths
//! - Independent TCP connections per thread
//!
//! ## Expected Results
//!
//! - Linear scaling up to # of CPU cores
//! - 8 threads × 130k msg/sec = 1M+ aggregate throughput
//! - No contention or lock overhead
use bytes::Bytes;
use criterion::{BenchmarkId, Criterion, Throughput, black_box, criterion_group, criterion_main};
// Identifies which runtime backend this build benchmarks, so compio, tokio, and smol
// results land under distinct criterion ids instead of overwriting each other.
const BENCH_BACKEND: &str = if cfg!(feature = "runtime-tokio") {
"tokio"
} else if cfg!(feature = "runtime-smol") {
"smol"
} else {
"compio"
};
use monocoque::rt::TcpListener;
use monocoque::zmq::{DealerSocket, RouterSocket, SocketOptions};
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
const MESSAGE_SIZE: usize = 64;
const MESSAGES_PER_THREAD: usize = 1_000; // Reduced to avoid deadlock
const THREAD_COUNTS: &[usize] = &[1, 2, 4, 8];
const BATCH_SIZE: usize = 100; // Process in batches to avoid deadlock
/// Benchmark multi-threaded DEALER clients against single ROUTER server
///
/// This tests horizontal scalability and lock-free architecture.
#[allow(dead_code)]
fn monocoque_multithreaded_dealers(c: &mut Criterion) {
let mut group = c.benchmark_group(format!("multithreaded/monocoque-{BENCH_BACKEND}/dealers"));
group.measurement_time(Duration::from_secs(20));
group.sample_size(10); // Minimum required by criterion
let payload = Bytes::from(vec![0u8; MESSAGE_SIZE]);
for &num_threads in THREAD_COUNTS {
let total_messages = num_threads * MESSAGES_PER_THREAD;
group.throughput(Throughput::Elements(total_messages as u64));
group.bench_with_input(
BenchmarkId::new("threads", num_threads),
&num_threads,
|b, &num_threads| {
b.iter(|| {
// Use a single runtime for the server
let rt = monocoque::rt::LocalRuntime::new().unwrap();
rt.block_on(async {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let server_addr = listener.local_addr().unwrap();
let expected_total = num_threads * MESSAGES_PER_THREAD;
let received_count = Arc::new(AtomicUsize::new(0));
// Router server task (handles all connections)
let router_task = monocoque::rt::spawn({
let received_count = Arc::clone(&received_count);
async move {
// Accept connections and spawn handler for each
let mut handlers = Vec::new();
for _ in 0..num_threads {
let (stream, _) = listener.accept().await.unwrap();
let received_count = Arc::clone(&received_count);
let handler = monocoque::rt::spawn(async move {
let mut router = RouterSocket::from_tcp_with_options(
stream,
SocketOptions::default()
.with_buffer_sizes(16384, 16384),
)
.await
.unwrap();
while received_count.load(Ordering::Relaxed)
< expected_total
{
if let Ok(Some(msg)) = router.recv().await {
received_count.fetch_add(1, Ordering::Relaxed);
router.send(msg).await.ok();
} else {
break;
}
}
});
handlers.push(handler);
}
// Wait for all handlers
for handler in handlers {
monocoque::rt::join(handler).await;
}
}
});
// Small delay to ensure server is listening
monocoque::rt::sleep(Duration::from_millis(50)).await;
// Spawn N dealer threads, each with its own runtime
let mut dealer_handles = Vec::new();
for _i in 0..num_threads {
let payload = payload.clone();
let handle = std::thread::spawn(move || {
// Each thread gets its own compio runtime
let rt = monocoque::rt::LocalRuntime::new().unwrap();
rt.block_on(async {
let stream = monocoque::rt::TcpStream::connect(server_addr)
.await
.unwrap();
let mut dealer = DealerSocket::from_tcp_with_options(
stream,
SocketOptions::default().with_buffer_sizes(16384, 16384),
)
.await
.unwrap();
// Use batched streaming to avoid deadlock
for _ in 0..(MESSAGES_PER_THREAD / BATCH_SIZE) {
// Send batch
for _ in 0..BATCH_SIZE {
dealer
.send(vec![black_box(payload.clone())])
.await
.unwrap();
}
// Receive batch
for _ in 0..BATCH_SIZE {
if dealer.recv().await.ok().flatten().is_none() {
break;
}
}
}
});
});
dealer_handles.push(handle);
}
// Wait for all dealer threads
for handle in dealer_handles {
handle.join().unwrap();
}
// Wait for router to finish
monocoque::rt::join(router_task).await;
});
});
},
);
}
group.finish();
}
/// Benchmark multi-threaded independent DEALER/ROUTER pairs
///
/// This tests scalability when each thread has completely isolated communication.
fn monocoque_multithreaded_independent_pairs(c: &mut Criterion) {
let mut group = c.benchmark_group(format!(
"multithreaded/monocoque-{BENCH_BACKEND}/independent_pairs"
));
group.measurement_time(Duration::from_secs(20));
group.sample_size(10);
let payload = Bytes::from(vec![0u8; MESSAGE_SIZE]);
for &num_threads in THREAD_COUNTS {
let total_messages = num_threads * MESSAGES_PER_THREAD;
group.throughput(Throughput::Elements(total_messages as u64));
group.bench_with_input(
BenchmarkId::new("pairs", num_threads),
&num_threads,
|b, &num_threads| {
b.iter_custom(|iters| {
// Each pair sets up ONE persistent DEALER<->ROUTER connection
// outside the timed section, then times `iters` rounds of the
// batched message workload over that same connection. Reusing
// the connection is what makes this measure message throughput
// rather than TCP setup, and it avoids churning a fresh
// connection (and its TIME_WAIT) per iteration - which, once
// the workload is fast, exhausts the ephemeral port range and
// fails a later bind with AddrInUse.
let mut handles = Vec::new();
for _i in 0..num_threads {
let payload = payload.clone();
let handle = std::thread::spawn(move || {
let rt = monocoque::rt::LocalRuntime::new().unwrap();
rt.block_on(async {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let server_addr = listener.local_addr().unwrap();
// ROUTER echoes every message across all `iters`
// rounds over the single accepted connection.
let total = iters as usize * MESSAGES_PER_THREAD;
let router_task = monocoque::rt::spawn(async move {
let (stream, _) = listener.accept().await.unwrap();
let mut router = RouterSocket::from_tcp_with_options(
stream,
SocketOptions::default().with_write_coalescing(true),
)
.await
.unwrap();
let mut buf: Vec<Bytes> = Vec::with_capacity(4);
let mut echoed = 0usize;
for _ in 0..total {
if router.recv_into(&mut buf).await.unwrap_or(false) {
router.send(buf.clone()).await.ok();
echoed += 1;
if echoed.is_multiple_of(BATCH_SIZE) {
router.flush().await.ok();
}
}
}
router.flush().await.ok();
});
let stream = monocoque::rt::TcpStream::connect(server_addr)
.await
.unwrap();
let mut dealer = DealerSocket::from_tcp_with_options(
stream,
SocketOptions::default().with_write_coalescing(true),
)
.await
.unwrap();
// Timed: `iters` rounds over the one connection.
let mut buf: Vec<Bytes> = Vec::with_capacity(4);
let start = std::time::Instant::now();
for _ in 0..iters {
for _ in 0..(MESSAGES_PER_THREAD / BATCH_SIZE) {
for _ in 0..BATCH_SIZE {
dealer
.send(vec![black_box(payload.clone())])
.await
.unwrap();
}
dealer.flush().await.unwrap();
for _ in 0..BATCH_SIZE {
if !dealer.recv_into(&mut buf).await.unwrap_or(false) {
break;
}
}
}
}
let elapsed = start.elapsed();
drop(dealer);
monocoque::rt::join(router_task).await;
elapsed
})
});
handles.push(handle);
}
// Pairs run concurrently; the batch wall-clock is the slowest.
let mut max_elapsed = Duration::ZERO;
for handle in handles {
let e = handle.join().unwrap();
if e > max_elapsed {
max_elapsed = e;
}
}
max_elapsed
});
},
);
}
group.finish();
}
/// Benchmark CPU core utilization efficiency
///
/// Measures how efficiently threads utilize CPU cores (msg/sec per core).
#[allow(dead_code)]
fn monocoque_core_efficiency(c: &mut Criterion) {
let mut group = c.benchmark_group(format!(
"multithreaded/monocoque-{BENCH_BACKEND}/core_efficiency"
));
group.measurement_time(Duration::from_secs(20));
group.sample_size(10);
let payload = Bytes::from(vec![0u8; MESSAGE_SIZE]);
let num_cores = num_cpus::get();
// Test at 50%, 100%, and 150% of available cores
let test_counts = vec![num_cores / 2, num_cores, num_cores + num_cores / 2];
for num_threads in test_counts {
if num_threads == 0 {
continue;
}
let total_messages = num_threads * MESSAGES_PER_THREAD;
group.throughput(Throughput::Elements(total_messages as u64));
group.bench_with_input(
BenchmarkId::new("cores", format!("{num_threads}/{num_cores}")),
&num_threads,
|b, &num_threads| {
b.iter(|| {
let mut handles = Vec::new();
for _i in 0..num_threads {
let payload = payload.clone();
let handle = std::thread::spawn(move || {
let rt = monocoque::rt::LocalRuntime::new().unwrap();
rt.block_on(async {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let server_addr = listener.local_addr().unwrap();
let router_task = monocoque::rt::spawn(async move {
let (stream, _) = listener.accept().await.unwrap();
let mut router = RouterSocket::from_tcp_with_options(
stream,
SocketOptions::default().with_write_coalescing(true),
)
.await
.unwrap();
let mut buf: Vec<Bytes> = Vec::with_capacity(4);
let mut echoed = 0usize;
for _ in 0..MESSAGES_PER_THREAD {
if router.recv_into(&mut buf).await.unwrap_or(false) {
router.send(buf.clone()).await.ok();
echoed += 1;
if echoed.is_multiple_of(BATCH_SIZE) {
router.flush().await.ok();
}
}
}
});
let stream = monocoque::rt::TcpStream::connect(server_addr)
.await
.unwrap();
let mut dealer = DealerSocket::from_tcp_with_options(
stream,
SocketOptions::default().with_write_coalescing(true),
)
.await
.unwrap();
let mut buf: Vec<Bytes> = Vec::with_capacity(4);
for _ in 0..(MESSAGES_PER_THREAD / BATCH_SIZE) {
for _ in 0..BATCH_SIZE {
dealer
.send(vec![black_box(payload.clone())])
.await
.unwrap();
}
dealer.flush().await.unwrap();
for _ in 0..BATCH_SIZE {
if !dealer.recv_into(&mut buf).await.unwrap_or(false) {
break;
}
}
}
monocoque::rt::join(router_task).await;
});
});
handles.push(handle);
}
for handle in handles {
handle.join().unwrap();
}
});
},
);
}
group.finish();
}
criterion_group!(
benches,
monocoque_multithreaded_independent_pairs, // Simplest case
// monocoque_multithreaded_dealers, // Disabled: complex coordination
// monocoque_core_efficiency, // Disabled: complex coordination
);
criterion_main!(benches);