dope-core 0.15.0

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
109
110
111
112
113
114
115
116
117
118
119
120
121

use std::{ops, process};

use crate::{
    backend::{
        self, fixed,
        uring::{
            self,
            engine::{completions, tuning},
            ring,
        },
    },
    driver::{
        self, flight,
        ops::{poll, reactors},
        schedule,
    },
    io,
};

impl<'d> poll::Source<'d> for driver::Context<'_, 'd> {
    fn dispatch(
        &mut self,
        work: schedule::Reactor<'_, 'd>,
        mut dispatch: impl FnMut(io::Event<'d>, &mut Self) -> ops::ControlFlow<io::Event<'d>>,
    ) -> poll::Dispatch<'d> {
        let work = reactors::Backend::prepare(self, work);
        let total = work.remaining();
        let cursor = self.backend().next_reactor_cursor();
        let maintenance_limit = reactors::Fair::share::<{ super::REACTOR_LANES }>(total, 0, cursor);
        let maintenance = {
            let mut budget = work.budget::<super::Maintenance>(maintenance_limit);
            super::maintain(self, &mut budget)
        };

        let source = ring::Drain::begin(&mut self.backend().ring);
        let mut retained = None;
        {
            let limit = work.remaining();
            let mut budget = work.budget::<super::Completion>(limit);
            while let schedule::Admission::Item(item) = budget.admit_with(|| {
                source
                    .next(&mut self.backend().ring)
                    .map(completions::Cqe::new)
            }) {
                let event = {
                    let (backend, drain) = self.backend_drain();
                    let backend::Uring {
                        ring,
                        tuning,
                        fixed_slots,
                        ..
                    } = backend;
                    resolve_event(ring.buffers().provided(), tuning, fixed_slots, &drain, item)
                };
                let Some(event) = event else {
                    continue;
                };
                match dispatch(event, self) {
                    ops::ControlFlow::Continue(()) => {}
                    ops::ControlFlow::Break(event) => {
                        retained = Some(event);
                        break;
                    }
                }
            }
        }
        let source_pending = source.finish(&mut self.backend().ring);

        if maintenance == super::MaintenanceState::Pending && work.remaining() != 0 {
            let limit = work.remaining();
            let mut budget = work.budget::<super::Maintenance>(limit);
            let _ = super::maintain(self, &mut budget);
        }
        self.backend().ring.buffers().provided().flush();
        let drain = if source_pending || retained.is_some() {
            poll::Drain::Pending
        } else {
            poll::Drain::Drained
        };
        poll::Dispatch::new(drain, retained)
    }
}

fn resolve_event<'d>(
    provided: &mut uring::ffi::ProvidedRing,
    tuning: &mut tuning::Table,
    fixed_slots: &mut fixed::Slots,
    drain: &flight::Drain<'_, 'd>,
    item: completions::Cqe,
) -> Option<io::Event<'d>> {
    let driver = drain.driver();
    let disposition =
        completions::Resolver::new(provided, tuning, fixed_slots, drain).resolve(item);
    match disposition {
        completions::Disposition::Consumed(buffer) => {
            if let Some(buffer) = buffer {
                provided.defer(buffer);
            }
            None
        }
        completions::Disposition::Public(completion) => Some(io::Event::from_completion(
            completion,
            driver,
            |len, buffer| provided.region(buffer, buffer_len(len)),
        )),
        completions::Disposition::Closed(closed) => {
            if closed.result() < 0 || closed.work().retires_slot() {
                process::abort();
            }
            closed.settle(fixed_slots, driver);
            None
        }
    }
}

fn buffer_len(len: u32) -> usize {
    let Ok(len) = usize::try_from(len) else {
        process::abort();
    };
    len
}