dope-net 0.46.0

Owned network transports for dope
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

mod cell;

pub(in crate::link::oneshot) use cell::Cell;
use dope_core::{
    driver::{self, result, route::Identity},
    io::{
        event::net::{connect, send, socket::Completion},
        socket,
    },
};
use o3::depot::{self, arena::slab::external::errors};

use crate::link::{
    egress::data,
    oneshot::{
        Pool, Progress,
        entry::{Access, Entry},
    },
};

/// Retained operations on one exact installed pool and its driver turn.
pub struct Io<'a, 'turn, 'd, ID: Identity, A: 'd, P: socket::Prepared, B: 'd>
where
    B: data::raw::StableOwner<'d>,
{
    pool: &'a mut Pool<'d, ID, A, P, B>,
    driver: &'a mut driver::Context<'turn, 'd>,
}

impl<'a, 'turn, 'd, ID: Identity, A: 'd, P: socket::Prepared, B: 'd> Io<'a, 'turn, 'd, ID, A, P, B>
where
    B: data::raw::StableOwner<'d>,
{
    /// # Safety
    /// This pool belongs to `driver`'s installed owner; retain its fixed backing,
    /// addresses and payloads until completion or quiescence, beyond this access.
    pub unsafe fn new(
        pool: &'a mut Pool<'d, ID, A, P, B>,
        driver: &'a mut driver::Context<'turn, 'd>,
    ) -> Self {
        Self { pool, driver }
    }

    pub fn socket(&mut self, done: Completion<'d>) -> Progress<A> {
        let local = match self.pool.routes.parse(done.token()) {
            Ok(local) => local,
            Err(_) => {
                done.reclaim();
                return Progress::Idle;
            }
        };
        // SAFETY: this Pool just admitted the completion's own physical segment.
        let access =
            Access::new(unsafe { depot::raw::Proof::new((&mut *self.pool, local.index())) });
        match Entry::try_new(access, local.parts().epoch()) {
            Ok(entry) => entry.socket(done, self.driver),
            Err(errors::Lookup::Physical { .. } | errors::Lookup::Generation { .. }) => {
                done.reclaim();
                Progress::Idle
            }
        }
    }

    pub fn connect(
        &mut self,
        done: connect::Completion<'d>,
    ) -> Result<Progress<A>, connect::Completion<'d>> {
        let local = match self.pool.routes.parse(done.token()) {
            Ok(local) => local,
            Err(_) => {
                return Err(done);
            }
        };
        // SAFETY: this Pool just admitted the completion's own physical segment.
        let access =
            Access::new(unsafe { depot::raw::Proof::new((&mut *self.pool, local.index())) });
        match Entry::try_new(access, local.parts().epoch()) {
            Ok(entry) => entry.connect(done, self.driver),
            Err(errors::Lookup::Physical { .. } | errors::Lookup::Generation { .. }) => Err(done),
        }
    }

    pub fn send(
        &mut self,
        done: send::Completion<'d>,
    ) -> Result<Progress<A>, send::Completion<'d>> {
        let local = match self.pool.routes.parse(done.token()) {
            Ok(local) => local,
            Err(_) => {
                return Err(done);
            }
        };
        // SAFETY: this Pool just admitted the completion's own physical segment.
        let access =
            Access::new(unsafe { depot::raw::Proof::new((&mut *self.pool, local.index())) });
        match Entry::try_new(access, local.parts().epoch()) {
            Ok(entry) => entry.send(done, self.driver),
            Err(errors::Lookup::Physical { .. } | errors::Lookup::Generation { .. }) => Err(done),
        }
    }

    /// Retries one operation; a full submission queue retains it for a later turn.
    pub fn retry_one(&mut self) -> Result<Progress<A>, result::SubmitError> {
        let Some(local) = self.pool.retries.pop_front() else {
            return Ok(Progress::Idle);
        };
        // SAFETY: this exact Pool's retry queue retained the admitted position.
        let access =
            Access::new(unsafe { depot::raw::Proof::new((&mut *self.pool, local.index())) });
        match Entry::try_new(access, local.parts().epoch()) {
            Ok(entry) => entry.retry(self.driver),
            Err(errors::Lookup::Physical { .. } | errors::Lookup::Generation { .. }) => {
                Ok(Progress::Idle)
            }
        }
    }
}