dope-net 0.27.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

//! Backend-quiescent connection teardown.
use dope_core::driver::{
    life::finish,
    receive::{Absent, Registration, credit},
    route::Identity,
};
use o3::base::{Call as _, cell::region};

use crate::{
    link::{
        egress::{self, reserve::demand},
        event,
        pool::{
            Key,
            change::{self, drain, quiesce::Strategy},
            recv::{
                input,
                storage::{queue::Ops as _, tail, tail::Slot},
            },
            send::Finish,
        },
    },
    wire::{self, send::Store as _},
};

impl<'p, 'w, 'd, ID: Identity + 'w, T, W, S, M, B, const IOV: usize, D>
    change::raw::Prepared<'p, 'w, 'd, ID, T, W, S, M, B, IOV, D>
where
    T: crate::Transport,
    W: wire::Spec + 'w + 'd,
    M: input::Mode<W>,
    D: demand::Mode,
    'd: 'w,
{
    pub(in crate::link::pool::change) fn run(
        &mut self,
        finish: &mut finish::Finish<'_, 'd>,
        mut free: impl FnMut(B, &mut region::Token<region::Identity<'d, 1>>),
    ) where
        M::Ingress: Strategy<W>,
    {
        self.retire(finish, |egress, key, finish| {
            egress.finish(key, finish, &mut free)
        });
    }

    pub(in crate::link::pool::change) fn keyed(
        &mut self,
        finish: &mut finish::Finish<'_, 'd>,
        mut free: impl FnMut(Key<'d, ID>, B),
    ) where
        M::Ingress: Strategy<W>,
    {
        self.retire(finish, |egress, key, finish| {
            egress.finish(key, finish, |payload, _| free(key, payload))
        });
    }

    fn retire(
        &mut self,
        finish: &mut finish::Finish<'_, 'd>,
        mut free: impl FnMut(
            &mut egress::Storage<'p, 'd, ID, B, IOV, D>,
            Key<'d, ID>,
            &mut finish::Finish<'_, 'd>,
        ),
    ) where
        M::Ingress: Strategy<W>,
    {
        if let Some(storage) = <M::Ingress as Strategy<W>>::credits(&self.slab) {
            finish.credit().detach(storage);
        }
        if self.slab.view().is_empty() {
            return;
        }
        for index in self.slab.view().capacity() {
            let released = {
                let Some((entry, mut recv)) = Slot::at::<M::Ingress>(&mut self.slab, index) else {
                    continue;
                };
                if let Some(handle) = self.pending.handle(entry.get().key) {
                    handle.wait_send(false);
                }
                let target = credit::Id::new(entry.get().key.target().dispatch());
                if let Some(cancel) = <M::Ingress as drain::Strategy<W>>::cancel(&recv.storage) {
                    cancel.call((&mut self.runtime, target));
                }
                while let Some(front) = recv.storage.prepare_pop() {
                    match front.pop().classify() {
                        event::Kind::Data(..) => {}
                        event::Kind::Control(control) => finish.reclaim(control.retire()),
                    }
                }
                let mut entry = tail::Store::clear(entry);
                let target = entry.get().engine.ready_handle().target();
                match Registration::try_acquire(target) {
                    Ok(reg) => reg.retire(),
                    Err(Absent) => entry.get_mut().engine.operations.recv.exhaustion().cancel(),
                }
                free(&mut self.egress, entry.get().key, finish);
                entry.remove().control.send.release()
            };
            if let Some(release) = <M::Ingress as Finish<W>>::release::<ID>(released) {
                release.call(&mut self.runtime);
            }
        }
    }
}