msrt 0.1.6

Portable MSRT protocol implementation.
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

//! Reliable packet recovery state.

use crate::core::{MessageId, PacketKey, PacketType, Result};
#[cfg(feature = "dynamic-recovery")]
use crate::reliability::{DynamicRecoveryConfig, DynamicRecoveryState};

use self::inflight::{InFlightPacket, InFlightPackets};

pub(crate) mod inflight;
pub(crate) mod retransmit;

/// Reliable-send recovery state machine.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) struct RecoveryState {
    in_flight: InFlightPackets,
    last_tick_ms: Option<u64>,
    #[cfg(feature = "dynamic-recovery")]
    dynamic: DynamicRecoveryState,
}

impl RecoveryState {
    #[cfg(not(feature = "dynamic-recovery"))]
    pub(crate) const fn new() -> Self {
        Self {
            in_flight: InFlightPackets::new(),
            last_tick_ms: None,
        }
    }

    #[cfg(feature = "dynamic-recovery")]
    pub(crate) const fn new(dynamic_config: DynamicRecoveryConfig) -> Self {
        Self {
            in_flight: InFlightPackets::new(),
            last_tick_ms: None,
            dynamic: DynamicRecoveryState::new(dynamic_config),
        }
    }

    pub(crate) fn track(&mut self, packet: InFlightPacket) -> Result<()> {
        self.in_flight.track(packet)
    }

    pub(crate) fn apply_ack(&mut self, key: PacketKey) {
        self.in_flight.apply_ack(key);
    }

    #[cfg(feature = "dynamic-recovery")]
    pub(crate) fn apply_ack_at(&mut self, key: PacketKey, now_ms: u64) {
        if let Some(packet) = self.in_flight.packet(key)
            && packet.sent
        {
            self.dynamic
                .observe_ack(now_ms.saturating_sub(packet.last_sent_ms));
        }

        self.apply_ack(key);
    }

    pub(crate) fn packets(&self) -> impl Iterator<Item = &InFlightPacket> {
        self.in_flight.packets()
    }

    #[cfg_attr(not(feature = "tracing"), allow(dead_code))]
    pub(crate) const fn in_flight_len(&self) -> usize {
        self.in_flight.len()
    }

    pub(crate) const fn available_in_flight(&self) -> usize {
        self.in_flight.available()
    }

    pub(crate) fn should_tick(&mut self, now_ms: u64) -> bool {
        if self.last_tick_ms == Some(now_ms) {
            return false;
        }

        self.last_tick_ms = Some(now_ms);
        true
    }

    pub(crate) fn remove_message(&mut self, packet_type: PacketType, message_id: MessageId) {
        self.in_flight.remove_message(packet_type, message_id);
    }

    pub(crate) fn note_sent(&mut self, key: PacketKey, now_ms: u64) {
        self.in_flight.note_sent(key, now_ms);
    }

    pub(crate) fn note_retransmit_sent(&mut self, key: PacketKey, now_ms: u64) {
        self.in_flight.note_retransmit_sent(key, now_ms);
    }

    #[cfg(feature = "dynamic-recovery")]
    pub(crate) fn dynamic_timeout_ms(&self, config: DynamicRecoveryConfig, attempts: u8) -> u64 {
        self.dynamic.timeout_ms(config, attempts)
    }
}