azur 0.3.1

A no_std Rust crate that implements an executor/reactor and futures using io_uring
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

pub(crate) mod pending_state;

use core::{
    future::Future as CoreFuture,
    marker::PhantomPinned,
    pin::Pin,
    task::{
        Context,
        Poll,
    },
};

use futures_core::future::FusedFuture;
use tracing::error;

use crate::{
    op::Op,
    reactor::{
        Reactor,
        REACTOR,
        TASK,
    },
    task::TaskHeader,
};

struct Pending<O: Op> {
    state: pending_state::Packed,
    op: O,
}

pub struct Future<O: Op> {
    pending: Option<Pending<O>>,
    _marker: PhantomPinned,
}

impl<O: Op> Future<O> {
    pub(crate) fn new(op: O) -> Self {
        Self {
            pending: Some(Pending {
                state: pending_state::Packed::from(pending_state::Unpacked::New),
                op,
            }),
            _marker: PhantomPinned,
        }
    }
}

impl<O: Op> CoreFuture for Future<O> {
    type Output = O::Output;

    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
        // SAFETY: The below code does not move `this`.
        let this = unsafe { self.get_unchecked_mut() };

        let pending = this
            .pending
            .as_mut()
            .expect("polling on a terminated future");
        match pending_state::Unpacked::from(pending.state) {
            pending_state::Unpacked::New => {
                unsafe {
                    Reactor::submit(
                        &REACTOR,
                        &mut pending.op,
                        &mut pending.state as *mut pending_state::Packed,
                    )
                    .unwrap()
                };
                let mut task = unsafe { (*TASK.get()).unwrap() };
                unsafe { task.as_mut().increment_ref_count() };
                pending.state =
                    pending_state::Packed::from(pending_state::Unpacked::Submitted { task });
                Poll::Pending
            }
            pending_state::Unpacked::Submitted { .. } => Poll::Pending,
            pending_state::Unpacked::Completed { ret } => {
                let pending = this.pending.take().unwrap();
                Poll::Ready(pending.op.complete(ret))
            }
        }
    }
}

impl<O: Op> FusedFuture for Future<O> {
    fn is_terminated(&self) -> bool {
        self.pending.is_none()
    }
}

#[cold]
fn cancel(pending_state: &mut pending_state::Packed) {
    match pending_state::Unpacked::from(*pending_state) {
        pending_state::Unpacked::Submitted { task } => {
            unsafe { TaskHeader::decrement_ref_count(task) };
        }
        _ => return,
    };
    if let Err(err) = Reactor::cancel(&REACTOR, pending_state as *mut pending_state::Packed) {
        error!("failed to cancel operation: {err}");
        crate::freeze();
    }
}

impl<O: Op> Drop for Future<O> {
    fn drop(&mut self) {
        if let Some(pending) = &mut self.pending {
            cancel(&mut pending.state);
        }
    }
}