raw_http/
raw_http.rs

1//! A do-nothing HTTP/1.1 responder, to measure the runtime's ceiling.
2//!
3//! It finds the end of each request head (`\r\n\r\n`, searched naively) and answers every
4//! request with a fixed `200 OK` and keep-alive. Pipelined requests in one read are answered in
5//! one batch. `--mode threads` runs the same responder as a thread-per-connection std server,
6//! the baseline.
7//!
8//! ```text
9//! cargo run --release -p simoom-rt --example raw_http -- --port 8080 --threads 4 --secs 30
10//!     [--mode rt|threads] [--backend auto|epoll|io_uring|iocp] [--cpus 0,2,4,6]
11//! ```
12//!
13//! At exit it prints the requests served and, in `rt` mode, the system calls per request.
14
15#![forbid(unsafe_code)]
16
17use simoom_rt::{Backend, ConnInfo, Connection, Ctx, Runtime};
18use std::cell::Cell;
19use std::io::{Read, Write};
20use std::sync::atomic::{AtomicU64, Ordering};
21use std::sync::{mpsc, Arc};
22use std::time::Duration;
23
24const RESPONSE: &[u8] = b"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 13\r\nConnection: keep-alive\r\n\r\nHello, World!";
25
26thread_local! {
27    static SERVED: Cell<u64> = const { Cell::new(0) };
28}
29
30/// Counts complete request heads in `data`; returns (requests, bytes up to the last one's end).
31fn heads(data: &[u8]) -> (usize, usize) {
32    let mut n = 0;
33    let mut end = 0;
34    let mut i = 0;
35    while i + 4 <= data.len() {
36        if &data[i..i + 4] == b"\r\n\r\n" {
37            n += 1;
38            i += 4;
39            end = i;
40        } else {
41            i += 1;
42        }
43    }
44    (n, end)
45}
46
47struct Raw;
48impl Connection for Raw {
49    fn on_data(&mut self, cx: &mut Ctx<'_>, data: &[u8]) -> usize {
50        let (n, end) = heads(data);
51        for _ in 0..n {
52            cx.send_static(RESPONSE);
53        }
54        SERVED.with(|c| c.set(c.get() + n as u64));
55        // `Connection: close` (a connection per request, S5): close once the reply is written,
56        // as an HTTP server does.
57        if n > 0 && data[..end].windows(17).any(|w| w.eq_ignore_ascii_case(b"connection: close")) {
58            cx.close();
59        }
60        end
61    }
62}
63
64struct Args {
65    port: u16,
66    threads: usize,
67    secs: u64,
68    mode: String,
69    backend: Backend,
70    cpus: Option<Vec<usize>>,
71    eager: bool,
72}
73
74fn args() -> Args {
75    let mut a = Args { port: 8080, threads: 0, secs: 30, mode: "rt".into(), backend: Backend::Auto, cpus: None, eager: false };
76    let v: Vec<String> = std::env::args().skip(1).collect();
77    let mut i = 0;
78    while i < v.len() {
79        let val = v.get(i + 1).cloned().unwrap_or_default();
80        match v[i].as_str() {
81            "--port" => a.port = val.parse().expect("--port"),
82            "--threads" => a.threads = val.parse().expect("--threads"),
83            "--secs" => a.secs = val.parse().expect("--secs"),
84            "--mode" => a.mode = val,
85            "--backend" => {
86                a.backend = match val.as_str() {
87                    "epoll" => Backend::Epoll,
88                    "io_uring" | "uring" => Backend::IoUring,
89                    "iocp" => Backend::Iocp,
90                    "kqueue" => Backend::Kqueue,
91                    _ => Backend::Auto,
92                }
93            }
94            "--eager" => {
95                a.eager = true;
96                i += 1;
97                continue;
98            }
99            "--cpus" => a.cpus = Some(val.split(',').map(|s| s.trim().parse().expect("--cpus")).collect()),
100            other => panic!("unknown argument {other}"),
101        }
102        i += 2;
103    }
104    a
105}
106
107fn main() {
108    let a = args();
109    let addr = format!("127.0.0.1:{}", a.port);
110    if a.mode == "threads" {
111        threads_mode(&addr, a.secs);
112        return;
113    }
114    let cpus = a.cpus.clone();
115    let rt = Runtime::builder()
116        .configure(|c| {
117            c.threads = a.threads;
118            c.driver.backend = a.backend;
119            c.driver.iocp_eager_recv = a.eager;
120            if let Some(cpus) = cpus {
121                c.pin_threads = true;
122                c.cpus = Some(cpus);
123            }
124        })
125        .listen(addr.as_str(), |_: &ConnInfo| Some(Box::new(Raw) as Box<dyn Connection>))
126        .start()
127        .expect("start");
128    eprintln!(
129        "raw_http: {} loops on {:?}, back end {:?}{}",
130        rt.loops().len(),
131        rt.local_addrs(),
132        rt.backend(),
133        rt.fallback_reason().map(|r| format!(" ({r})")).unwrap_or_default()
134    );
135    std::thread::sleep(Duration::from_secs(a.secs));
136    let (tx, rx) = mpsc::channel();
137    for l in rt.loops() {
138        let tx = tx.clone();
139        l.run(move |_| tx.send(SERVED.with(|c| c.get())).unwrap());
140    }
141    drop(tx);
142    let served: u64 = rx.iter().sum();
143    let t = rt.stats().total();
144    let d = t.driver;
145    let per = |x: u64| if served == 0 { 0.0 } else { x as f64 / served as f64 };
146    println!("served {served} requests on {} connections", t.accepted);
147    println!(
148        "per request: {:.3} syscalls (waits {:.3}, recv {:.3}, send {:.3}, other {:.3}); {:.3} io_uring SQEs; {:.3} OS events; {:.3} sleeps; {:.3} spin polls",
149        per(d.syscalls()),
150        per(d.waits),
151        per(d.recv_calls),
152        per(d.send_calls),
153        per(d.accept_calls + d.other_calls),
154        per(d.sqes),
155        per(d.os_events),
156        per(d.sleeps),
157        per(d.spin_polls)
158    );
159    rt.stop(Duration::from_millis(100));
160    rt.join();
161}
162
163/// The baseline: a std thread per connection, blocking reads and writes.
164fn threads_mode(addr: &str, secs: u64) {
165    let l = std::net::TcpListener::bind(addr).expect("bind");
166    eprintln!("raw_http: thread per connection on {}", l.local_addr().unwrap());
167    let served = Arc::new(AtomicU64::new(0));
168    let s2 = served.clone();
169    std::thread::spawn(move || {
170        for s in l.incoming() {
171            let Ok(mut s) = s else { continue };
172            let _ = s.set_nodelay(true);
173            let served = s2.clone();
174            std::thread::spawn(move || {
175                let mut buf = vec![0u8; 4096];
176                let mut have = 0;
177                let mut out = Vec::with_capacity(4096);
178                let mut local = 0u64;
179                loop {
180                    let n = match s.read(&mut buf[have..]) {
181                        Ok(0) | Err(_) => break,
182                        Ok(n) => n,
183                    };
184                    have += n;
185                    let (reqs, end) = heads(&buf[..have]);
186                    if reqs > 0 {
187                        out.clear();
188                        for _ in 0..reqs {
189                            out.extend_from_slice(RESPONSE);
190                        }
191                        if s.write_all(&out).is_err() {
192                            break;
193                        }
194                        buf.copy_within(end..have, 0);
195                        have -= end;
196                        local += reqs as u64;
197                        if local >= 1024 {
198                            served.fetch_add(local, Ordering::Relaxed);
199                            local = 0;
200                        }
201                    }
202                    if have == buf.len() {
203                        break;
204                    }
205                }
206                served.fetch_add(local, Ordering::Relaxed);
207            });
208        }
209    });
210    std::thread::sleep(Duration::from_secs(secs));
211    println!("served {} requests (thread per connection)", served.load(Ordering::Relaxed));
212    let _ = std::io::stdout().flush();
213    std::process::exit(0);
214}
raw_http/raw_http.rs

raw_http/
raw_http.rs
