dope 0.5.5

The manifold runtime
Documentation 
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

//! Thread-per-core deployment.
//!
//! The recommended production model is thread-per-core: one [`Executor`] per
//! CPU, with the worker count equal to the number of cores. [`Launcher::run`]
//! spawns one OS thread per CPU in `cpus`, pins each via [`Launcher::pin_to_cpu`]
//! ([`Ctx::cpu`]), and runs the closure on every core. Each worker builds its
//! own driver from the [`Production`] profile via
//! [`DriverCfg::for_tcp_profile::<Production>(max_conn)`][for_tcp_profile],
//! tagging the ring with [`with_cpu_id`]. Cores share one listening port through
//! the `SO_REUSEPORT` listener, enabled with
//! [`ListenerOpts::reuseport`][reuseport]. Adapt to the host with the `max_conn`,
//! `backlog`, and CPU-pinning knobs.
//!
//! [`Launcher::from_affinity`] binds one worker per CPU the process is actually
//! allowed to run on, honouring a non-contiguous `--cpuset-cpus` such as
//! `0-31,64-95`; [`Launcher::allowed_cpus`] returns that list for custom slicing.
//!
//! ```ignore
//! use dope::{DriverConfig, Executor, runtime::profile::Production};
//! use dope::launcher::{Ctx, Launcher};
//!
//! Launcher::from_affinity().run(|ctx: Ctx| {
//!     let cfg = dope::DriverCfg::for_tcp_profile::<Production>(8192)
//!         .with_cpu_id(Some(ctx.cpu));
//!     let _exec = Executor::new(cfg)?;
//!     // bind a SO_REUSEPORT listener and run the dispatcher loop here
//!     Ok(())
//! })?;
//! # Ok::<(), std::io::Error>(())
//! ```
//!
//! [`Executor`]: crate::Executor
//! [`Production`]: crate::runtime::profile::Production
//! [for_tcp_profile]: crate::DriverConfig::for_tcp_profile
//! [`with_cpu_id`]: crate::DriverConfig::with_cpu_id
//! [reuseport]: crate::transport::config::tcp::ListenerOpts::reuseport

use std::io;

use crate::backend;

#[derive(Clone, Copy, Debug)]
pub struct Ctx {
    pub cpu: u16,
}

impl Ctx {
    fn enter<F>(self, run: &F) -> io::Result<()>
    where
        F: Fn(Ctx) -> io::Result<()>,
    {
        <backend::Default as backend::Backend>::init_thread(self.cpu)?;
        run(self)
    }
}

pub struct Launcher {
    cpus: Vec<u16>,
}

impl Launcher {
    pub fn new(cpus: Vec<u16>) -> Self {
        Self { cpus }
    }

    pub fn allowed_cpus() -> Vec<u16> {
        <backend::Default as backend::Backend>::allowed_cpus()
    }

    pub fn from_affinity() -> Self {
        Self::new(Self::allowed_cpus())
    }

    pub fn pin_to_cpu(cpu: u16) -> io::Result<()> {
        <backend::Default as backend::Backend>::init_thread(cpu)
    }

    pub fn run<F>(self, run_thread: F) -> io::Result<()>
    where
        F: Fn(Ctx) -> io::Result<()> + Send + Sync,
    {
        let Self { cpus } = self;
        let (last, rest) = match cpus.split_last() {
            Some(split) => split,
            None => {
                return Err(io::Error::new(
                    io::ErrorKind::InvalidInput,
                    "Launcher::run requires at least one cpu",
                ));
            }
        };

        let run_ref = &run_thread;
        std::thread::scope(|s| -> io::Result<()> {
            let mut handles = Vec::with_capacity(rest.len());
            for &cpu in rest {
                let ctx = Ctx { cpu };
                handles.push(s.spawn(move || ctx.enter(run_ref)));
            }
            let leader_result = Ctx { cpu: *last }.enter(run_ref);
            for h in handles {
                h.join()
                    .map_err(|_| io::Error::other("launcher worker panicked"))??;
            }
            leader_result
        })
    }
}