dope_core/backend/uring/ops/
source.rs

1use std::{ops, process};
2
3use crate::{
4    backend::{
5        self, fixed,
6        uring::{
7            self,
8            engine::{completions, tuning},
9            ring,
10        },
11    },
12    driver::{
13        self, flight,
14        ops::{poll, reactors},
15        schedule,
16    },
17    io,
18};
19
20impl<'d> poll::Source<'d> for driver::Context<'_, 'd> {
21    fn dispatch(
22        &mut self,
23        work: schedule::Reactor<'_, 'd>,
24        mut dispatch: impl FnMut(io::Event<'d>, &mut Self) -> ops::ControlFlow<io::Event<'d>>,
25    ) -> poll::Dispatch<'d> {
26        let work = reactors::Backend::prepare(self, work);
27        let total = work.remaining();
28        let cursor = self.backend().next_reactor_cursor();
29        let maintenance_limit = reactors::Fair::share::<{ super::REACTOR_LANES }>(total, 0, cursor);
30        let maintenance = {
31            let mut budget = work.budget::<super::Maintenance>(maintenance_limit);
32            super::maintain(self, &mut budget)
33        };
34
35        let source = ring::Drain::begin(&mut self.backend().ring);
36        let mut retained = None;
37        {
38            let limit = work.remaining();
39            let mut budget = work.budget::<super::Completion>(limit);
40            while let schedule::Admission::Item(item) = budget.admit_with(|| {
41                source
42                    .next(&mut self.backend().ring)
43                    .map(completions::Cqe::new)
44            }) {
45                let event = {
46                    let (backend, drain) = self.backend_drain();
47                    let backend::Uring {
48                        ring,
49                        tuning,
50                        fixed_slots,
51                        ..
52                    } = backend;
53                    resolve_event(ring.buffers().provided(), tuning, fixed_slots, &drain, item)
54                };
55                let Some(event) = event else {
56                    continue;
57                };
58                match dispatch(event, self) {
59                    ops::ControlFlow::Continue(()) => {}
60                    ops::ControlFlow::Break(event) => {
61                        retained = Some(event);
62                        break;
63                    }
64                }
65            }
66        }
67        let source_pending = source.finish(&mut self.backend().ring);
68
69        if maintenance == super::MaintenanceState::Pending && work.remaining() != 0 {
70            let limit = work.remaining();
71            let mut budget = work.budget::<super::Maintenance>(limit);
72            let _ = super::maintain(self, &mut budget);
73        }
74        self.backend().ring.buffers().provided().flush();
75        let drain = if source_pending || retained.is_some() {
76            poll::Drain::Pending
77        } else {
78            poll::Drain::Drained
79        };
80        poll::Dispatch::new(drain, retained)
81    }
82}
83
84fn resolve_event<'d>(
85    provided: &mut uring::ffi::ProvidedRing,
86    tuning: &mut tuning::Table,
87    fixed_slots: &mut fixed::Slots,
88    drain: &flight::Drain<'_, 'd>,
89    item: completions::Cqe,
90) -> Option<io::Event<'d>> {
91    let driver = drain.driver();
92    let disposition =
93        completions::Resolver::new(provided, tuning, fixed_slots, drain).resolve(item);
94    match disposition {
95        completions::Disposition::Consumed(buffer) => {
96            if let Some(buffer) = buffer {
97                provided.defer(buffer);
98            }
99            None
100        }
101        completions::Disposition::Public(completion) => Some(io::Event::from_completion(
102            completion,
103            driver,
104            |len, buffer| provided.region(buffer, buffer_len(len)),
105        )),
106        completions::Disposition::Closed(closed) => {
107            if closed.result() < 0 || closed.work().retires_slot() {
108                process::abort();
109            }
110            closed.settle(fixed_slots, driver);
111            None
112        }
113    }
114}
115
116fn buffer_len(len: u32) -> usize {
117    let Ok(len) = usize::try_from(len) else {
118        process::abort();
119    };
120    len
121}
dope_core/backend/uring/ops/source.rs

dope_core/backend/uring/ops/
source.rs
