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
use std::sync::Arc;
use super::*;
/// Configuration for the underlying `io_uring` system.
#[derive(Clone, Debug, Copy)]
pub struct Config {
/// The number of entries in the submission queue.
/// The completion queue size may be specified by
/// using `raw_params` instead. By default, the
/// kernel will choose a completion queue that is 2x
/// the submission queue's size.
pub depth: usize,
/// Enable `SQPOLL` mode, which spawns a kernel
/// thread that polls for submissions without
/// needing to block as often to submit.
///
/// This is a privileged operation, and
/// will cause `start` to fail if run
/// by a non-privileged user.
pub sq_poll: bool,
/// Specify a particular CPU to pin the
/// `SQPOLL` thread onto.
pub sq_poll_affinity: u32,
/// Specify that the user will directly
/// poll the hardware for operation completion
/// rather than using the completion queue.
///
/// CURRENTLY UNSUPPORTED
pub io_poll: bool,
/// Print a profile table on drop, showing where
/// time was spent.
pub print_profile_on_drop: bool,
/// setting `raw_params` overrides everything else
pub raw_params: Option<io_uring_params>,
}
impl Default for Config {
fn default() -> Config {
Config {
depth: 256,
sq_poll: false,
io_poll: false,
sq_poll_affinity: 0,
raw_params: None,
print_profile_on_drop: false,
}
}
}
impl Config {
/// Start the `Rio` system.
pub fn start(mut self) -> io::Result<Rio> {
let mut params =
if let Some(params) = self.raw_params.take() {
params
} else {
let mut params = io_uring_params::default();
if self.sq_poll {
// set SQPOLL mode to avoid needing wakeup
params.flags = IORING_SETUP_SQPOLL;
params.sq_thread_cpu =
self.sq_poll_affinity;
}
params
};
let params_ptr: *mut io_uring_params = &mut params;
let ring_fd = setup(
u32::try_from(self.depth).unwrap(),
params_ptr,
)?;
if ring_fd < 0 {
let mut err = io::Error::last_os_error();
if let Some(12) = err.raw_os_error() {
err = io::Error::new(
io::ErrorKind::Other,
"Not enough lockable memory. You probably \
need to raise the memlock rlimit, which \
often defaults to a pretty low number.",
);
}
return Err(err);
}
let in_flight = Arc::new(InFlight::new(
params.cq_entries as usize,
));
let ticket_queue = Arc::new(TicketQueue::new(
params.cq_entries as usize,
));
let sq = Sq::new(¶ms, ring_fd)?;
let cq = Cq::new(
¶ms,
ring_fd,
in_flight.clone(),
ticket_queue.clone(),
)?;
std::thread::spawn(move || {
let mut cq = cq;
cq.reaper(ring_fd)
});
Ok(Rio(Arc::new(Uring::new(
self,
params.flags,
ring_fd,
sq,
in_flight,
ticket_queue,
))))
}
}