cartel-core 0.28.0

Shared core for cartel database drivers
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

use std::{mem, pin, task};

use dope::core::driver::schedule::clock::timer;
use dope_fiber::context;
use o3::base::time::Positive;

/// The bounded operation phase in which an absolute request deadline elapsed.
#[derive(Clone, Copy, Eq, PartialEq, Debug)]
pub enum RequestPhase {
    Admission,
    Response,
    Settlement,
    Lease,
    Contention,
}

#[derive(Clone, Copy, Eq, PartialEq, Debug)]
pub struct Elapsed {
    phase: RequestPhase,
}

impl Elapsed {
    pub const fn new(phase: RequestPhase) -> Self {
        Self { phase }
    }

    pub const fn phase(self) -> RequestPhase {
        self.phase
    }
}

/// Copyable authority to create deadlines tied to one exact driver lifetime.
#[derive(Clone, Copy)]
pub struct Source<'d> {
    timer: timer::Timer<'d>,
    window: Positive,
}

impl<'d> Source<'d> {
    pub const fn new(timer: timer::Timer<'d>, window: Positive) -> Self {
        Self { timer, window }
    }

    #[inline]
    pub fn guard(self) -> Result<Guard<'d>, timer::Exhausted> {
        Ok(Guard {
            state: GuardState::Live {
                alarm: self.timer.try_alarm()?,
                window: Some(self.window),
            },
        })
    }

    pub const fn into_parts(self) -> (timer::Timer<'d>, Positive) {
        (self.timer, self.window)
    }
}

/// One first-poll-anchored absolute deadline. Phase changes never rearm it.
pub struct Guard<'d> {
    state: GuardState<'d>,
}

enum GuardState<'d> {
    Live {
        alarm: timer::Alarm<'d>,
        window: Option<Positive>,
    },
    Elapsed,
    Complete,
}

impl<'d> Guard<'d> {
    pub fn poll_expired(
        self: pin::Pin<&mut Self>,
        mut ctx: pin::Pin<&mut context::Context<'_, 'd>>,
        phase: RequestPhase,
    ) -> task::Poll<Elapsed> {
        let state = &mut self.get_mut().state;
        let (alarm, window) = match state {
            GuardState::Live { alarm, window } => (alarm, window),
            GuardState::Elapsed => return task::Poll::Ready(Elapsed::new(phase)),
            GuardState::Complete => return task::Poll::Pending,
        };
        let wake = ctx.as_ref().completion();
        let now = ctx.as_mut().driver().driver_ref().clock().now();
        let expired = match window.take() {
            Some(duration) => alarm.arm_and_poll(now.clamp_add(duration), now, wake),
            None => alarm.poll(now, wake),
        };
        if expired.is_ready() {
            *state = GuardState::Elapsed;
            task::Poll::Ready(Elapsed::new(phase))
        } else {
            task::Poll::Pending
        }
    }

    pub fn complete(&mut self) {
        self.state = GuardState::Complete;
    }
}

const _: () = {
    const LIVE: usize = mem::size_of::<timer::Alarm<'static>>() + mem::size_of::<Positive>();

    assert!(mem::size_of::<Guard<'static>>() == LIVE);
    assert!(mem::size_of::<Option<Guard<'static>>>() == LIVE);
    assert!(mem::size_of::<Result<Guard<'static>, timer::Exhausted>>() == LIVE);
};