Struct RuntimeHandle

Source

pub struct RuntimeHandle { /* private fields */ }

Expand description

A running runtime. Dropping it stops the runtime at once (no grace) and waits for its threads; call RuntimeHandle::stop and RuntimeHandle::join for a graceful stop.

Implementations§

Source §

impl RuntimeHandle

Source

pub fn local_addrs(&self) -> &[SocketAddr]

The bound TCP addresses, in the order of Builder::listen calls.

Examples found in repository ?

examples/raw_http.rs (line 131)

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}

Source

pub fn udp_addrs(&self) -> &[SocketAddr]

The bound UDP addresses, in the order of Builder::udp calls.

Source

pub fn loops(&self) -> Vec<LoopHandle>

The loops that run, by index, to run tasks on them. With Config::lazy_loops that is loop 0 until three connections were open at once, then each loop as it starts.

Examples found in repository ?

examples/raw_http.rs (line 130)

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}

More examples

Hide additional examples

examples/loadgen.rs (line 110)

76fn main() {
77    let mut addr: SocketAddr = "127.0.0.1:8080".parse().unwrap();
78    let (mut conns, mut threads, mut secs, mut pipeline) = (64usize, 4usize, 10u64, 1usize);
79    let mut cpus: Option<Vec<usize>> = None;
80    let v: Vec<String> = std::env::args().skip(1).collect();
81    let mut i = 0;
82    while i < v.len() {
83        let val = v.get(i + 1).cloned().unwrap_or_default();
84        match v[i].as_str() {
85            "--addr" => addr = val.parse().expect("--addr"),
86            "--conns" => conns = val.parse().expect("--conns"),
87            "--threads" => threads = val.parse().expect("--threads"),
88            "--secs" => secs = val.parse().expect("--secs"),
89            "--pipeline" => pipeline = val.parse().expect("--pipeline"),
90            "--cpus" => cpus = Some(val.split(',').map(|s| s.trim().parse().expect("--cpus")).collect()),
91            other => panic!("unknown argument {other}"),
92        }
93        i += 2;
94    }
95    let rt = Runtime::builder()
96        .configure(|c| {
97            c.threads = threads;
98            c.thread_name = "loadgen".into();
99            if let Some(cpus) = cpus {
100                c.pin_threads = true;
101                c.cpus = Some(cpus);
102            }
103        })
104        .start()
105        .expect("start");
106    let (htx, hrx) = mpsc::channel::<ConnHandle>();
107    for k in 0..conns {
108        let sock = Socket::tcp_connect(addr).expect("connect");
109        let htx = htx.clone();
110        rt.loops()[k % threads].run(move |lcx| {
111            let h = lcx.adopt(sock, Box::new(Client { pipeline })).expect("adopt");
112            htx.send(h).unwrap();
113        });
114    }
115    drop(htx);
116    let handles: Vec<ConnHandle> = hrx.iter().take(conns).collect();
117    for h in &handles {
118        h.send(Box::new(()));
119    }
120    let count = |rt: &simoom_rt::RuntimeHandle| -> (u64, u64) {
121        let (tx, rx) = mpsc::channel();
122        for l in rt.loops() {
123            let tx = tx.clone();
124            l.run(move |_| tx.send((DONE.with(|c| c.get()), ERRORS.with(|c| c.get()))).unwrap());
125        }
126        drop(tx);
127        rx.iter().fold((0, 0), |a, b| (a.0 + b.0, a.1 + b.1))
128    };
129    std::thread::sleep(Duration::from_secs(1));
130    let (r0, _) = count(&rt);
131    let s0 = rt.stats().total().driver;
132    let t0 = Instant::now();
133    std::thread::sleep(Duration::from_secs(secs));
134    let (r1, errors) = count(&rt);
135    let el = t0.elapsed().as_secs_f64();
136    let s1 = rt.stats().total().driver;
137    let rps = (r1 - r0) as f64 / el;
138    let lat_us = (conns * pipeline) as f64 / rps * 1e6;
139    println!(
140        "{conns} conns x {pipeline} pipelined, {threads} threads: {rps:.0} req/s, mean latency {lat_us:.1} us (Little), closed connections {errors}, client syscalls/req {:.2}",
141        (s1.syscalls() - s0.syscalls()) as f64 / (r1 - r0).max(1) as f64
142    );
143    rt.stop(Duration::ZERO);
144    rt.join();
145}

Source

pub fn backend(&self) -> Backend

The back end the loops use. With Config::lazy_loops it waits until loop 0 has built its driver (well under a millisecond after Builder::start).

Examples found in repository ?

examples/raw_http.rs (line 132)

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}

Source

pub fn fallback_reason(&self) -> Option<&str>

Why the configured back end was replaced (io_uring falling back to epoll), if it was.

Examples found in repository ?

examples/raw_http.rs (line 133)

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}

Source

pub fn offload(&self) -> &OffloadPool

The shared offload pool.

Source

pub fn stats(&self) -> Stats

Current counters.

Examples found in repository ?

examples/raw_http.rs (line 143)

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}

More examples

Hide additional examples

examples/loadgen.rs (line 131)

76fn main() {
77    let mut addr: SocketAddr = "127.0.0.1:8080".parse().unwrap();
78    let (mut conns, mut threads, mut secs, mut pipeline) = (64usize, 4usize, 10u64, 1usize);
79    let mut cpus: Option<Vec<usize>> = None;
80    let v: Vec<String> = std::env::args().skip(1).collect();
81    let mut i = 0;
82    while i < v.len() {
83        let val = v.get(i + 1).cloned().unwrap_or_default();
84        match v[i].as_str() {
85            "--addr" => addr = val.parse().expect("--addr"),
86            "--conns" => conns = val.parse().expect("--conns"),
87            "--threads" => threads = val.parse().expect("--threads"),
88            "--secs" => secs = val.parse().expect("--secs"),
89            "--pipeline" => pipeline = val.parse().expect("--pipeline"),
90            "--cpus" => cpus = Some(val.split(',').map(|s| s.trim().parse().expect("--cpus")).collect()),
91            other => panic!("unknown argument {other}"),
92        }
93        i += 2;
94    }
95    let rt = Runtime::builder()
96        .configure(|c| {
97            c.threads = threads;
98            c.thread_name = "loadgen".into();
99            if let Some(cpus) = cpus {
100                c.pin_threads = true;
101                c.cpus = Some(cpus);
102            }
103        })
104        .start()
105        .expect("start");
106    let (htx, hrx) = mpsc::channel::<ConnHandle>();
107    for k in 0..conns {
108        let sock = Socket::tcp_connect(addr).expect("connect");
109        let htx = htx.clone();
110        rt.loops()[k % threads].run(move |lcx| {
111            let h = lcx.adopt(sock, Box::new(Client { pipeline })).expect("adopt");
112            htx.send(h).unwrap();
113        });
114    }
115    drop(htx);
116    let handles: Vec<ConnHandle> = hrx.iter().take(conns).collect();
117    for h in &handles {
118        h.send(Box::new(()));
119    }
120    let count = |rt: &simoom_rt::RuntimeHandle| -> (u64, u64) {
121        let (tx, rx) = mpsc::channel();
122        for l in rt.loops() {
123            let tx = tx.clone();
124            l.run(move |_| tx.send((DONE.with(|c| c.get()), ERRORS.with(|c| c.get()))).unwrap());
125        }
126        drop(tx);
127        rx.iter().fold((0, 0), |a, b| (a.0 + b.0, a.1 + b.1))
128    };
129    std::thread::sleep(Duration::from_secs(1));
130    let (r0, _) = count(&rt);
131    let s0 = rt.stats().total().driver;
132    let t0 = Instant::now();
133    std::thread::sleep(Duration::from_secs(secs));
134    let (r1, errors) = count(&rt);
135    let el = t0.elapsed().as_secs_f64();
136    let s1 = rt.stats().total().driver;
137    let rps = (r1 - r0) as f64 / el;
138    let lat_us = (conns * pipeline) as f64 / rps * 1e6;
139    println!(
140        "{conns} conns x {pipeline} pipelined, {threads} threads: {rps:.0} req/s, mean latency {lat_us:.1} us (Little), closed connections {errors}, client syscalls/req {:.2}",
141        (s1.syscalls() - s0.syscalls()) as f64 / (r1 - r0).max(1) as f64
142    );
143    rt.stop(Duration::ZERO);
144    rt.join();
145}

Source

pub fn stop(&self, grace: Duration)

Starts a graceful stop: listeners close, every connection gets crate::Connection::on_stop, and connections still open after grace are aborted. Loops held back by Config::lazy_loops are not started any more.

Examples found in repository ?

examples/raw_http.rs (line 159)

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}

More examples

Hide additional examples

examples/loadgen.rs (line 143)

76fn main() {
77    let mut addr: SocketAddr = "127.0.0.1:8080".parse().unwrap();
78    let (mut conns, mut threads, mut secs, mut pipeline) = (64usize, 4usize, 10u64, 1usize);
79    let mut cpus: Option<Vec<usize>> = None;
80    let v: Vec<String> = std::env::args().skip(1).collect();
81    let mut i = 0;
82    while i < v.len() {
83        let val = v.get(i + 1).cloned().unwrap_or_default();
84        match v[i].as_str() {
85            "--addr" => addr = val.parse().expect("--addr"),
86            "--conns" => conns = val.parse().expect("--conns"),
87            "--threads" => threads = val.parse().expect("--threads"),
88            "--secs" => secs = val.parse().expect("--secs"),
89            "--pipeline" => pipeline = val.parse().expect("--pipeline"),
90            "--cpus" => cpus = Some(val.split(',').map(|s| s.trim().parse().expect("--cpus")).collect()),
91            other => panic!("unknown argument {other}"),
92        }
93        i += 2;
94    }
95    let rt = Runtime::builder()
96        .configure(|c| {
97            c.threads = threads;
98            c.thread_name = "loadgen".into();
99            if let Some(cpus) = cpus {
100                c.pin_threads = true;
101                c.cpus = Some(cpus);
102            }
103        })
104        .start()
105        .expect("start");
106    let (htx, hrx) = mpsc::channel::<ConnHandle>();
107    for k in 0..conns {
108        let sock = Socket::tcp_connect(addr).expect("connect");
109        let htx = htx.clone();
110        rt.loops()[k % threads].run(move |lcx| {
111            let h = lcx.adopt(sock, Box::new(Client { pipeline })).expect("adopt");
112            htx.send(h).unwrap();
113        });
114    }
115    drop(htx);
116    let handles: Vec<ConnHandle> = hrx.iter().take(conns).collect();
117    for h in &handles {
118        h.send(Box::new(()));
119    }
120    let count = |rt: &simoom_rt::RuntimeHandle| -> (u64, u64) {
121        let (tx, rx) = mpsc::channel();
122        for l in rt.loops() {
123            let tx = tx.clone();
124            l.run(move |_| tx.send((DONE.with(|c| c.get()), ERRORS.with(|c| c.get()))).unwrap());
125        }
126        drop(tx);
127        rx.iter().fold((0, 0), |a, b| (a.0 + b.0, a.1 + b.1))
128    };
129    std::thread::sleep(Duration::from_secs(1));
130    let (r0, _) = count(&rt);
131    let s0 = rt.stats().total().driver;
132    let t0 = Instant::now();
133    std::thread::sleep(Duration::from_secs(secs));
134    let (r1, errors) = count(&rt);
135    let el = t0.elapsed().as_secs_f64();
136    let s1 = rt.stats().total().driver;
137    let rps = (r1 - r0) as f64 / el;
138    let lat_us = (conns * pipeline) as f64 / rps * 1e6;
139    println!(
140        "{conns} conns x {pipeline} pipelined, {threads} threads: {rps:.0} req/s, mean latency {lat_us:.1} us (Little), closed connections {errors}, client syscalls/req {:.2}",
141        (s1.syscalls() - s0.syscalls()) as f64 / (r1 - r0).max(1) as f64
142    );
143    rt.stop(Duration::ZERO);
144    rt.join();
145}

Source

pub fn join(self)

Waits for every loop to exit (after RuntimeHandle::stop), then stops the offload pool.

Examples found in repository ?

examples/raw_http.rs (line 160)

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}

More examples

Hide additional examples

examples/loadgen.rs (line 144)

76fn main() {
77    let mut addr: SocketAddr = "127.0.0.1:8080".parse().unwrap();
78    let (mut conns, mut threads, mut secs, mut pipeline) = (64usize, 4usize, 10u64, 1usize);
79    let mut cpus: Option<Vec<usize>> = None;
80    let v: Vec<String> = std::env::args().skip(1).collect();
81    let mut i = 0;
82    while i < v.len() {
83        let val = v.get(i + 1).cloned().unwrap_or_default();
84        match v[i].as_str() {
85            "--addr" => addr = val.parse().expect("--addr"),
86            "--conns" => conns = val.parse().expect("--conns"),
87            "--threads" => threads = val.parse().expect("--threads"),
88            "--secs" => secs = val.parse().expect("--secs"),
89            "--pipeline" => pipeline = val.parse().expect("--pipeline"),
90            "--cpus" => cpus = Some(val.split(',').map(|s| s.trim().parse().expect("--cpus")).collect()),
91            other => panic!("unknown argument {other}"),
92        }
93        i += 2;
94    }
95    let rt = Runtime::builder()
96        .configure(|c| {
97            c.threads = threads;
98            c.thread_name = "loadgen".into();
99            if let Some(cpus) = cpus {
100                c.pin_threads = true;
101                c.cpus = Some(cpus);
102            }
103        })
104        .start()
105        .expect("start");
106    let (htx, hrx) = mpsc::channel::<ConnHandle>();
107    for k in 0..conns {
108        let sock = Socket::tcp_connect(addr).expect("connect");
109        let htx = htx.clone();
110        rt.loops()[k % threads].run(move |lcx| {
111            let h = lcx.adopt(sock, Box::new(Client { pipeline })).expect("adopt");
112            htx.send(h).unwrap();
113        });
114    }
115    drop(htx);
116    let handles: Vec<ConnHandle> = hrx.iter().take(conns).collect();
117    for h in &handles {
118        h.send(Box::new(()));
119    }
120    let count = |rt: &simoom_rt::RuntimeHandle| -> (u64, u64) {
121        let (tx, rx) = mpsc::channel();
122        for l in rt.loops() {
123            let tx = tx.clone();
124            l.run(move |_| tx.send((DONE.with(|c| c.get()), ERRORS.with(|c| c.get()))).unwrap());
125        }
126        drop(tx);
127        rx.iter().fold((0, 0), |a, b| (a.0 + b.0, a.1 + b.1))
128    };
129    std::thread::sleep(Duration::from_secs(1));
130    let (r0, _) = count(&rt);
131    let s0 = rt.stats().total().driver;
132    let t0 = Instant::now();
133    std::thread::sleep(Duration::from_secs(secs));
134    let (r1, errors) = count(&rt);
135    let el = t0.elapsed().as_secs_f64();
136    let s1 = rt.stats().total().driver;
137    let rps = (r1 - r0) as f64 / el;
138    let lat_us = (conns * pipeline) as f64 / rps * 1e6;
139    println!(
140        "{conns} conns x {pipeline} pipelined, {threads} threads: {rps:.0} req/s, mean latency {lat_us:.1} us (Little), closed connections {errors}, client syscalls/req {:.2}",
141        (s1.syscalls() - s0.syscalls()) as f64 / (r1 - r0).max(1) as f64
142    );
143    rt.stop(Duration::ZERO);
144    rt.join();
145}