transport_bench/
transport_bench.rs

1// RLX models — distributed inference.
2// Copyright (C) 2026 Eugene Hauptmann, Nataliya Kosmyna.
3//
4// This program is free software: you can redistribute it and/or modify
5// it under the terms of the GNU General Public License as published by
6// the Free Software Foundation, version 3.
7//
8// SPDX-License-Identifier: GPL-3.0-only
9
10//! Transport microbenchmark — loopback TCP latency / bandwidth / collectives.
11//!
12//! Spins up a 2-rank mesh on `127.0.0.1` (a thread per rank, but real
13//! kernel TCP sockets) and measures the cost the distribution layer adds:
14//! point-to-point round-trip latency, streaming bandwidth across payload
15//! sizes, and `all_reduce` latency for a hidden-state-sized vector.
16//!
17//! Run: `cargo run -p rlx-distributed --example transport_bench --release`
18//!
19//! Loopback is ~memcpy speed, so these numbers isolate framing/serialization
20//! overhead; a real Ethernet/Thunderbolt link adds its own latency floor.
21
22use rlx_distributed::{NetTransport, ProcessGroup, ReduceKind};
23use std::net::{Ipv4Addr, SocketAddr, TcpListener};
24use std::sync::Arc;
25use std::thread;
26use std::time::Instant;
27
28fn main() {
29    let world = 2u32;
30    let listeners: Vec<TcpListener> = (0..world)
31        .map(|_| TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap())
32        .collect();
33    let addrs: Vec<SocketAddr> = listeners.iter().map(|l| l.local_addr().unwrap()).collect();
34
35    println!("transport: loopback TCP, world_size = {world}");
36    let handles: Vec<_> = listeners
37        .into_iter()
38        .enumerate()
39        .map(|(rank, listener)| {
40            let addrs = addrs.clone();
41            thread::spawn(move || {
42                let t = NetTransport::from_listener(rank as u32, world, listener, addrs, 64 << 20)
43                    .unwrap();
44                bench(&ProcessGroup::new(Arc::new(t)));
45            })
46        })
47        .collect();
48    for h in handles {
49        h.join().unwrap();
50    }
51}
52
53fn bench(g: &ProcessGroup) {
54    let rank = g.rank();
55    g.barrier().unwrap();
56
57    // ---- point-to-point round-trip latency (ping-pong) ----
58    let iters = 5000;
59    if rank == 0 {
60        let p = [1.0f32];
61        for _ in 0..200 {
62            g.send_f32(1, 0, &p).unwrap();
63            g.recv_f32(1, 0).unwrap();
64        }
65        let t0 = Instant::now();
66        for _ in 0..iters {
67            g.send_f32(1, 0, &p).unwrap();
68            g.recv_f32(1, 0).unwrap();
69        }
70        let us = t0.elapsed().as_secs_f64() * 1e6 / iters as f64;
71        println!(
72            "[latency]   ping-pong: {us:6.2} µs RTT   ({:.2} µs one-way)   [{iters} iters]",
73            us / 2.0
74        );
75    } else {
76        for _ in 0..(200 + iters) {
77            let x = g.recv_f32(0, 0).unwrap();
78            g.send_f32(0, 0, &x).unwrap();
79        }
80    }
81    g.barrier().unwrap();
82
83    // ---- streaming bandwidth across payload sizes ----
84    for &elems in &[256usize, 16_384, 262_144, 4_194_304] {
85        let bytes = elems * 4;
86        let reps = (512 * 1024 * 1024 / bytes).clamp(8, 200_000);
87        if rank == 0 {
88            let buf = vec![0.5f32; elems];
89            g.send_f32(1, 1, &[0.0]).unwrap(); // sync start
90            let t0 = Instant::now();
91            for _ in 0..reps {
92                g.send_f32(1, 2, &buf).unwrap();
93            }
94            g.recv_f32(1, 3).unwrap(); // ack: all received
95            let secs = t0.elapsed().as_secs_f64();
96            let gbps = (reps * bytes) as f64 / secs / 1e9;
97            println!(
98                "[bandwidth] {:>9} B × {:>6} = {:>5.1} GB  ->  {gbps:6.2} GB/s",
99                bytes,
100                reps,
101                (reps * bytes) as f64 / 1e9
102            );
103        } else {
104            g.recv_f32(0, 1).unwrap();
105            for _ in 0..reps {
106                g.recv_f32(0, 2).unwrap();
107            }
108            g.send_f32(0, 3, &[1.0]).unwrap();
109        }
110        g.barrier().unwrap();
111    }
112
113    // ---- collective latency for a hidden-state-sized vector ----
114    // e.g. batch1 × seq1 × d_model=4096 — the kind of all_reduce a
115    // tensor-parallel layer issues, or a broadcast the pipeline issues.
116    for &elems in &[4096usize, 4096 * 8] {
117        let mut data = vec![1.0f32; elems];
118        let iters = 2000;
119        for _ in 0..50 {
120            g.all_reduce(&mut data, ReduceKind::Sum).unwrap();
121        }
122        g.barrier().unwrap();
123        let t0 = Instant::now();
124        for _ in 0..iters {
125            g.all_reduce(&mut data, ReduceKind::Sum).unwrap();
126        }
127        let us = t0.elapsed().as_secs_f64() * 1e6 / iters as f64;
128        if rank == 0 {
129            println!("[collective] all_reduce {:>6} f32: {us:6.2} µs/op", elems);
130        }
131        g.barrier().unwrap();
132    }
133
134    g.barrier().unwrap();
135}
transport_bench/transport_bench.rs

transport_bench/
transport_bench.rs
