gmf 2.0.0

A runtime-agnostic, high-performance gRPC server framework using thread-per-core architecture.
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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148

use std::cell::Cell;
use std::future::Future;
use std::io;
use std::net::SocketAddr;
use std::rc::Rc;

use monoio::net::{TcpListener as MonoioTcpListener, TcpStream as MonoioTcpStream};

use crate::server::error::GmfError;
use crate::server::runtime::{
    Runtime, RuntimeExecutor, RuntimeSemaphore, RuntimeTcpListener, RuntimeTcpStream,
};

/// Thread-per-core runtime using monoio (io_uring on Linux, kqueue on macOS).
pub struct MonoioRuntime;

impl Runtime for MonoioRuntime {
    type TcpListener = MonoioListener;
    type Executor = MonoioExec;
    type Semaphore = MonoioSemaphore;

    fn run_multi_core<F, Fut>(cores: usize, f: F) -> Result<(), GmfError>
    where
        F: Fn(usize) -> Fut + Send + Clone + 'static,
        Fut: Future<Output = Result<(), GmfError>> + 'static,
    {
        let mut handles = Vec::with_capacity(cores);

        for cpu in 0..cores {
            let f = f.clone();
            let handle = std::thread::Builder::new()
                .name(format!("gmf_core_{cpu}"))
                .spawn(move || -> Result<(), GmfError> {
                    // Pin to CPU core on Linux for optimal thread-per-core performance.
                    #[cfg(target_os = "linux")]
                    {
                        unsafe {
                            let mut cpuset: libc::cpu_set_t = std::mem::zeroed();
                            libc::CPU_SET(cpu, &mut cpuset);
                            libc::sched_setaffinity(
                                0,
                                std::mem::size_of::<libc::cpu_set_t>(),
                                &cpuset,
                            );
                        }
                    }

                    let mut rt = monoio::RuntimeBuilder::<monoio::FusionDriver>::new()
                        .enable_timer()
                        .build()
                        .map_err(|e| GmfError::SpawnExecutor { cpu, source: e })?;

                    rt.block_on(f(cpu))
                })
                .map_err(|e| GmfError::SpawnExecutor { cpu, source: e })?;

            handles.push(handle);
        }

        for handle in handles {
            handle
                .join()
                .map_err(|_| GmfError::Io(io::Error::other("thread panicked")))??;
        }

        Ok(())
    }
}

// -- TCP Listener --

pub struct MonoioListener(MonoioTcpListener);

impl RuntimeTcpListener for MonoioListener {
    type Stream = MonoioStream;

    async fn bind(addr: SocketAddr) -> io::Result<Self> {
        let listener = MonoioTcpListener::bind(addr)?;
        Ok(MonoioListener(listener))
    }

    async fn accept(&self) -> io::Result<(Self::Stream, SocketAddr)> {
        let (stream, addr) = self.0.accept().await?;
        Ok((MonoioStream(stream), addr))
    }
}

// -- TCP Stream --

pub struct MonoioStream(MonoioTcpStream);

impl RuntimeTcpStream for MonoioStream {
    // monoio TcpStream is completion-based; wrap with StreamWrapper for poll-based compat,
    // then MonoioIo for hyper::rt::{Read, Write}.
    type HyperIo = monoio_compat::hyper::MonoioIo<monoio_compat::StreamWrapper<MonoioTcpStream>>;

    fn into_hyper_io(self) -> Self::HyperIo {
        let compat = monoio_compat::StreamWrapper::new(self.0);
        monoio_compat::hyper::MonoioIo::new(compat)
    }
}

// -- Executor --

#[derive(Clone, Default)]
pub struct MonoioExec;

impl RuntimeExecutor for MonoioExec {
    fn spawn<F: Future<Output = ()> + 'static>(&self, fut: F) {
        monoio::spawn(fut);
    }
}

impl<F> hyper::rt::Executor<F> for MonoioExec
where
    F: Future + 'static,
    F::Output: 'static,
{
    fn execute(&self, fut: F) {
        monoio::spawn(async move {
            fut.await;
        });
    }
}

// -- Semaphore --

pub struct MonoioSemaphore {
    permits: Rc<Cell<usize>>,
}

impl RuntimeSemaphore for MonoioSemaphore {
    fn new(permits: usize) -> Self {
        MonoioSemaphore {
            permits: Rc::new(Cell::new(permits)),
        }
    }

    fn try_acquire(&self) -> bool {
        let current = self.permits.get();
        if current > 0 {
            self.permits.set(current - 1);
            true
        } else {
            false
        }
    }
}