dope-net 0.24.0

Owned network transports 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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147

use dope_core::driver::{
    receive::credit,
    route::{self, Identity, table::entries::occupied},
};
use storage::queue::Ops;

use crate::{
    link::{
        duplex,
        egress::reserve::demand,
        pool::{
            self, pending,
            recv::{self, Missing, delivery, input, storage},
        },
        slot::types,
    },
    wire,
};

pub(in crate::link::pool::recv) struct Resume<
    'a,
    'w,
    'd,
    ID: Identity + 'w,
    W: wire::Spec + 'w + 'd,
    S,
    M: input::Mode<W>,
> where
    'd: 'w,
{
    entry: occupied::Entry<'a, types::Connection<'w, 'd, ID, W, S>, ID>,
    runtime: &'a mut W::Runtime<'w, 'd, ID>,
    recv: recv::Borrow<'a, 'd, ID, M::Ingress>,
    pending: &'a pending::Queue<'d, ID>,
}

impl<'a, 'w, 'd, ID: Identity + 'w, W, S, M> Resume<'a, 'w, 'd, ID, W, S, M>
where
    'd: 'w,
    W: wire::Spec + 'w + 'd,
    M: input::Mode<W>,
{
    pub(in crate::link::pool::recv) fn try_new<'p, T, B, const IOV: usize, D>(
        pool: &'a mut pool::Connections<'p, 'w, 'd, ID, T, W, S, M, B, IOV, D>,
        target: route::Target<'d, ID>,
    ) -> Result<Self, Missing<'d, ID>>
    where
        T: crate::Transport,
        D: demand::Mode,
    {
        let key = pool::Key::from_target(target);
        let (entry, storage) = pool
            .prepared
            .slab
            .try_entry(key.parts())
            .map_err(|cause| Missing { key, cause })?;
        Ok(Self {
            entry,
            runtime: &mut pool.prepared.runtime,
            recv: storage,
            pending: pool.prepared.pending,
        })
    }

    pub(in crate::link::pool::recv) fn direct(mut self) -> delivery::Resumed<'d, ID> {
        self.resume();
        let slot = self.entry.get_mut();
        delivery::Resumed::new(slot.key, slot.io().ready_key())
    }

    pub(in crate::link::pool::recv) fn credit(
        mut self,
        release: impl FnOnce(&mut W::Runtime<'w, 'd, ID>),
    ) -> delivery::Resumed<'d, ID>
    where
        M::Ingress: storage::raw::Mode<Bound<'a, 'd> = storage::Loan<'a, 'd>>,
    {
        let wake = self.prepare();
        match &wake {
            credit::Take::Ready(credit::Wake::Returned) => release(self.runtime),
            credit::Take::Ready(credit::Wake::Retry) | credit::Take::Idle => {}
        }
        self.finish(wake)
    }

    fn resume(&mut self) {
        let slot = self.entry.get_mut();
        if !slot.is_recv_done()
            && slot
                .control
                .engine
                .operations
                .recv
                .exhaustion()
                .try_resume()
                .is_started()
        {
            self.recv.rearm.queue();
        }
    }

    fn prepare(&mut self) -> credit::Take
    where
        M::Ingress: storage::raw::Mode<Bound<'a, 'd> = storage::Loan<'a, 'd>>,
    {
        self.resume();
        let slot = self.entry.get_mut();
        let wake = self.recv.storage.get().take();
        match &wake {
            credit::Take::Ready(credit::Wake::Returned)
            | credit::Take::Ready(credit::Wake::Retry) => {
                let close = slot.control.engine.operations.lifecycle.is_closing();
                let queued = self.recv.storage.has_queued();
                if !slot.terminal.is_read_done()
                    && slot
                        .control
                        .engine
                        .operations
                        .recv
                        .resume(close)
                        .is_some_and(duplex::Rearm::is_needed)
                    && !queued
                {
                    self.recv.rearm.queue();
                }
            }
            credit::Take::Idle => {}
        }
        wake
    }

    fn finish(self, wake: credit::Take) -> delivery::Resumed<'d, ID> {
        let slot = self.entry.get();
        match wake {
            credit::Take::Ready(credit::Wake::Returned)
            | credit::Take::Ready(credit::Wake::Retry) => {
                if slot.control.engine.operations.lifecycle.is_closing()
                    && let Some(handle) = self.pending.handle(slot.key)
                {
                    handle.wake();
                }
            }
            credit::Take::Idle => {}
        }
        delivery::Resumed::new(slot.key, slot.io().ready_key())
    }
}