Struct Builder

Source

pub struct Builder { /* private fields */ }

Expand description

Builds and starts a runtime. See crate::Runtime::builder.

Implementations§

Source §

impl Builder

Source

pub fn new() -> Builder

A builder with the default Config.

Source

pub fn config(self, cfg: Config) -> Builder

Replaces the whole configuration.

Source

pub fn configure(self, f: impl FnOnce(&mut Config)) -> Builder

Changes the configuration in place.

Examples found in repository ?

examples/raw_http.rs (lines 116-124)

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 (lines 96-103)

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 threads(self, n: usize) -> Builder

Loop threads (0 = one per CPU).

Source

pub fn backend(self, b: Backend) -> Builder

The I/O back end.

Source

pub fn listen( self, addr: impl ToSocketAddrs, acceptor: impl Acceptor, ) -> Builder

Accepts TCP connections on addr and runs them with acceptor. On Linux each loop gets its own SO_REUSEPORT listener; elsewhere one listener is shared and connections are handed to the less busy of two loops (on Windows, through one AcceptEx queue). Address errors are reported by Builder::start.

Examples found in repository ?

examples/raw_http.rs (line 125)

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 listen_on( self, listener: TcpListener, acceptor: impl Acceptor, ) -> Builder

Accepts on an existing listener (socket activation, a shared port). It is shared by all loops.

Source

pub fn udp( self, addr: impl ToSocketAddrs, opts: UdpOptions, service: impl UdpService, ) -> Builder

Binds UDP on addr and creates an endpoint per loop with service (one SO_REUSEPORT socket per loop on Linux; one socket on loop 0 elsewhere). Feature udp.

Source

pub fn start(self) -> Result<RuntimeHandle>

Binds every listener, starts the loops and returns once they all run.

Examples found in repository ?

examples/raw_http.rs (line 126)

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 104)

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}