dope-tls 0.19.0

TCP TLS streams for dope runtime, shin-backed
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

use dope::net::wire::{self, receive};
use o3::collections::queue::slot;

#[derive(Clone, Copy)]
pub(crate) enum Resource {
    Recv,
    HandshakeEgress,
}

pub struct Registration<'a, 'd, const ID: u8> {
    waiters: &'a mut Queue<'d, ID>,
    target: wire::RecvCreditId<'d, ID>,
    pending: bool,
}

pub(crate) struct Queue<'d, const ID: u8> {
    recv: slot::Fifo<wire::RecvCreditGuard<'d, ID>>,
    handshake_egress: slot::Fifo<wire::RecvCreditGuard<'d, ID>>,
}

impl<'d, const ID: u8> Queue<'d, ID> {
    pub(crate) fn with_capacity(capacity: usize) -> Self {
        Self {
            recv: slot::Fifo::with_capacity(capacity),
            handshake_egress: slot::Fifo::with_capacity(capacity),
        }
    }

    pub(crate) fn register(
        &mut self,
        resource: Resource,
        credit: wire::RecvCredit<'d, ID>,
    ) -> Option<Registration<'_, 'd, ID>> {
        let target = credit.id();
        {
            let queue = self.queue(resource);
            let vacant = queue.vacant_entry(target.index())?;
            let guard = credit.claim().ok()?;
            vacant.push_back(guard);
        }
        Some(Registration {
            waiters: self,
            target,
            pending: true,
        })
    }

    pub(crate) fn wake(&mut self, resource: Resource) {
        let Some(guard) = self.queue(resource).pop_front() else {
            return;
        };
        guard.retry();
    }

    pub(crate) fn cancel(&mut self, target: wire::RecvCreditId<'d, ID>) {
        let index = target.index();
        Self::cancel_queue(&mut self.recv, index, target);
        Self::cancel_queue(&mut self.handshake_egress, index, target);
    }

    fn cancel_queue(
        queue: &mut slot::Fifo<wire::RecvCreditGuard<'d, ID>>,
        index: usize,
        target: wire::RecvCreditId<'d, ID>,
    ) {
        let Some(guard) = queue.remove_if(index, |guard| guard.id() == target) else {
            return;
        };
        guard.cancel();
    }

    fn queue(&mut self, resource: Resource) -> &mut slot::Fifo<wire::RecvCreditGuard<'d, ID>> {
        match resource {
            Resource::Recv => &mut self.recv,
            Resource::HandshakeEgress => &mut self.handshake_egress,
        }
    }
}

impl<const ID: u8> receive::Registration for Registration<'_, '_, ID> {
    fn commit(mut self) {
        self.pending = false;
    }
}

impl<const ID: u8> Drop for Registration<'_, '_, ID> {
    fn drop(&mut self) {
        if self.pending {
            self.waiters.cancel(self.target);
        }
    }
}