dope-uring 0.21.0

Owned io_uring backend for dope
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

use std::convert::Infallible;

use dope_sys::{
    engine::{
        admit::{self, Context},
        ring,
    },
    io::{
        reactor::submit::{self, Control, Queue as _, Ready as _},
        request,
    },
    ring::ready::Ready as _,
    table::token::{self, Codec, Tuning as _},
};
use o3::base::num::bounded;

use crate::{
    Uring,
    driver::{
        Engine, Key,
        tuning::{self, Progress as _},
    },
    reactor::sq,
};

/// An SQ reservation for one key-bound cancellation.
pub enum Cancellation<'a, T: Codec> {
    Pending {
        vacant: sq::raw::Vacancy<'a, T>,
        key: Key<T>,
    },
    Tuning {
        vacant: sq::raw::Vacancy<'a, T>,
        target: T::Tuning,
        stopped: &'a mut bool,
    },
    Completed,
}

impl<T: Codec> admit::Submission for Cancellation<'_, T> {
    fn submit(self) -> admit::Cancelled {
        match self {
            Self::Pending { vacant, key } => {
                let entry = Uring::cancel(T::flight(key), T::cancel_key(key));
                submit::raw::Vacancy::prepare(vacant, &entry).push();
                admit::Cancelled::Submitted
            }
            Self::Tuning {
                vacant,
                target,
                stopped,
            } => {
                let entry = tuning::cancel::<T>(target);
                *stopped = true;
                submit::raw::Vacancy::prepare(vacant, &entry).push();
                admit::Cancelled::Submitted
            }
            Self::Completed => admit::Cancelled::Completed,
        }
    }
}

impl<'a, C> admit::Reserve<'a, C, Engine<C>> for Cancellation<'a, C::Token>
where
    C: ring::Context + Context<Native = Uring, Key = Key<C::Token>>,
    C::Tuning: tuning::Progress<C::Token>,
    Engine<C>: 'a,
{
    fn try_new(
        backend: &'a mut Engine<C>,
        key: C::Key,
        pending: bool,
    ) -> Result<Self, admit::Error<Infallible>> {
        if !pending {
            return Ok(Cancellation::Completed);
        }
        let vacant = backend
            .ring
            .submission()
            .try_one()
            .map_err(admit::Error::Full)?;
        Ok(Cancellation::Pending { vacant, key })
    }

    fn try_at(
        backend: &'a mut Engine<C>,
        key: C::Key,
        pending: bool,
        slot: &<C::Inputs as request::Context>::Target,
    ) -> Result<Self, admit::Error<Infallible>> {
        if !pending {
            return Ok(Cancellation::Completed);
        }
        let vacant = backend
            .ring
            .submission()
            .try_one()
            .map_err(admit::Error::Full)?;
        let slot = <Uring as request::Source<C::Inputs>>::target(*slot);
        let Some(stopped) = backend.tuning.cancel_at(slot, Some(key)) else {
            return Ok(Cancellation::Pending { vacant, key });
        };
        let index: <<C::Token as Codec>::Context as token::Context>::Slot =
            bounded::U32::clamp(slot.index() as usize).into();
        Ok(Cancellation::Tuning {
            vacant,
            target: <C::Token as Codec>::Tuning::new(index),
            stopped,
        })
    }
}