dope_core/driver/
control.rs

1use std::io;
2use std::os::fd::BorrowedFd;
3
4use super::token::Token;
5use super::{DriverContext, OutboundReservation, PushError};
6
7pub trait ContextControl {
8    fn prepare_drop(&mut self);
9    /// # Safety
10    /// `fd` must remain open through this call.
11    unsafe fn register_shutdown_fd(&mut self, fd: BorrowedFd<'_>) -> io::Result<()>;
12    fn reserve_outbound(&mut self, count: u32) -> io::Result<OutboundReservation>;
13    fn reserve_route(&mut self, id: u8) -> bool;
14    fn release_route(&mut self, id: u8);
15    fn poison_route(&mut self, id: u8);
16    fn quiesce(&mut self, targets: &[Token]) -> bool;
17    fn set(
18        &mut self,
19        fixed_idx: u32,
20        level: u32,
21        optname: u32,
22        value: i32,
23    ) -> Result<(), PushError>;
24}
25
26cfg_select! {
27    target_os = "linux" => {
28        use std::io::ErrorKind;
29        use std::mem::size_of;
30        use std::os::fd::AsRawFd;
31        use std::process::abort;
32
33        use io_uring::opcode::SetSockOpt;
34        use io_uring::types::{CancelBuilder, Fixed};
35
36        use super::submission::Submission;
37        use crate::backend::uring::sqe::Sqe;
38
39        impl ContextControl for DriverContext<'_, '_> {
40            fn prepare_drop(&mut self) {
41                self.backend().shutdown();
42            }
43
44            /// # Safety
45            /// `fd` must remain open through this call.
46            unsafe fn register_shutdown_fd(&mut self, fd: BorrowedFd<'_>) -> io::Result<()> {
47                Submission::push(self, Sqe::poll_shutdown(fd.as_raw_fd())).map_err(io::Error::from)?;
48                self.backend().uring.submit().map(|_| ())
49            }
50
51            fn reserve_outbound(&mut self, count: u32) -> io::Result<OutboundReservation> {
52                let base = self.backend().alloc_fixed_range(count)?;
53                Ok(OutboundReservation::new(base, count))
54            }
55
56            fn reserve_route(&mut self, id: u8) -> bool {
57                self.backend().routes.reserve(id)
58            }
59
60            fn release_route(&mut self, id: u8) {
61                self.backend().routes.release(id);
62            }
63
64            fn poison_route(&mut self, id: u8) {
65                self.backend().routes.poison(id);
66            }
67
68            fn quiesce(&mut self, targets: &[Token]) -> bool {
69                if targets.is_empty() {
70                    return false;
71                }
72                let state = self.backend();
73                if state.uring.submit().is_err() {
74                    abort();
75                }
76                for target in targets {
77                    match state
78                        .uring
79                        .submitter()
80                        .register_sync_cancel(None, CancelBuilder::user_data(target.raw()).all())
81                    {
82                        Ok(()) => {}
83                        Err(error) if error.kind() == ErrorKind::NotFound => {}
84                        Err(_) => abort(),
85                    }
86                }
87                true
88            }
89
90            fn set(
91                &mut self,
92                fixed_idx: u32,
93                level: u32,
94                optname: u32,
95                value: i32,
96            ) -> Result<(), PushError> {
97                let state = self.backend();
98                let Ok((key, stored)) = state.setsockopt.insert_entry(value) else {
99                    return Err(PushError);
100                };
101                let optval_ptr = (&raw const *stored).cast::<libc::c_void>();
102                let ud = Token::from_key(key);
103                let sqe = SetSockOpt::new(
104                    Fixed(fixed_idx),
105                    level,
106                    optname,
107                    optval_ptr,
108                    size_of::<libc::c_int>() as u32,
109                )
110                .build()
111                .user_data(ud.raw());
112                if unsafe { state.uring.submission().push(&sqe) }.is_ok() {
113                    Ok(())
114                } else {
115                    state.setsockopt.remove(key);
116                    Err(PushError)
117                }
118            }
119        }
120    }
121    _ => {
122        use std::mem::size_of;
123        use std::os::fd::AsRawFd;
124        use std::ptr::{null, null_mut};
125
126        use crate::backend::kqueue::driver::TAG_SHUTDOWN;
127        use crate::backend::kqueue::driver::retry::Retry;
128        use crate::backend::kqueue::driver::udata::Udata;
129        use crate::io::fd::FdSlot;
130
131        impl ContextControl for DriverContext<'_, '_> {
132            fn prepare_drop(&mut self) {
133                self.backend().shutdown();
134            }
135
136            /// # Safety
137            /// `fd` must remain open through this call.
138            unsafe fn register_shutdown_fd(&mut self, fd: BorrowedFd<'_>) -> io::Result<()> {
139                let state = self.backend();
140                let event = libc::kevent {
141                    ident: fd.as_raw_fd() as libc::uintptr_t,
142                    filter: libc::EVFILT_READ,
143                    flags: libc::EV_ADD | libc::EV_CLEAR,
144                    fflags: 0,
145                    data: 0,
146                    udata: Udata::pack(TAG_SHUTDOWN, 0, 0).into_kevent(),
147                };
148                let rc = unsafe {
149                    libc::kevent(
150                        state.kq.as_raw_fd(),
151                        &event,
152                        1,
153                        null_mut(),
154                        0,
155                        null(),
156                    )
157                };
158                if rc < 0 {
159                    Err(io::Error::last_os_error())
160                } else {
161                    Ok(())
162                }
163            }
164
165            fn reserve_outbound(&mut self, count: u32) -> io::Result<OutboundReservation> {
166                let base = self.backend().alloc_fixed_range(count)?;
167                Ok(OutboundReservation::new(base, count))
168            }
169
170            fn reserve_route(&mut self, id: u8) -> bool {
171                self.backend().routes.reserve(id)
172            }
173
174            fn release_route(&mut self, id: u8) {
175                self.backend().routes.release(id);
176            }
177
178            fn poison_route(&mut self, id: u8) {
179                self.backend().routes.poison(id);
180            }
181
182            fn quiesce(&mut self, targets: &[Token]) -> bool {
183                if targets.is_empty() {
184                    return false;
185                }
186                let state = self.backend();
187                for target in targets {
188                    state.quiesce_accept(*target);
189                    state.quiesce_recv(*target);
190                    state.retire_write_token(*target);
191                }
192                let mut extracted = state.pending.extract_targets(targets);
193                while let Some(completion) = state.pending.pop_extracted(&mut extracted) {
194                    state.reclaim(completion);
195                }
196                false
197            }
198
199            fn set(
200                &mut self,
201                fixed_idx: u32,
202                level: u32,
203                optname: u32,
204                value: i32,
205            ) -> Result<(), PushError> {
206                let Some(raw) = self.backend_ref().raw_fd(FdSlot::new(fixed_idx)) else {
207                    return Err(PushError);
208                };
209                let rc = unsafe {
210                    libc::setsockopt(
211                        raw,
212                        level as libc::c_int,
213                        optname as libc::c_int,
214                        (&value as *const libc::c_int).cast(),
215                        size_of::<libc::c_int>() as libc::socklen_t,
216                    )
217                };
218                if rc == 0 { Ok(()) } else { Err(PushError) }
219            }
220        }
221    }
222}
dope_core/driver/control.rs

dope_core/driver/
control.rs
