msrt 0.1.1

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

//! Tick-driven retransmission.

use crate::engine::{
    Engine, EngineOutput, MAX_IN_FLIGHT_PACKETS, SendFailedEvent, SendFailureReason, WriteEvent,
};

pub(crate) fn tick(engine: &mut Engine, now_ms: u64) {
    engine.now_ms = now_ms;
    engine
        .reassembly
        .expire(now_ms, engine.reassembly_timeout_ms);

    let mut retransmits = [None; MAX_IN_FLIGHT_PACKETS];
    let mut retransmit_len = 0;
    let mut failures = [None; MAX_IN_FLIGHT_PACKETS];
    let mut failure_len = 0;
    let mut failed_messages = [None; MAX_IN_FLIGHT_PACKETS];
    let mut failed_message_len = 0;

    for packet in engine.in_flight.packets() {
        if now_ms.saturating_sub(packet.last_sent_ms) < engine.retransmit_timeout_ms {
            continue;
        }

        if packet.attempts >= engine.max_retransmit_attempts {
            let already_failed = failed_messages[..failed_message_len]
                .iter()
                .flatten()
                .any(|key| *key == (packet.channel_id, packet.message_id));

            if !already_failed && failure_len < failures.len() {
                failures[failure_len] = Some(*packet);
                failure_len += 1;
                failed_messages[failed_message_len] = Some((packet.channel_id, packet.message_id));
                failed_message_len += 1;
            }
        } else if retransmit_len < retransmits.len() {
            retransmits[retransmit_len] = Some(*packet);
            retransmit_len += 1;
        }
    }

    for packet in failures[..failure_len].iter().flatten() {
        engine
            .in_flight
            .remove_message(packet.channel_id, packet.message_id);
        let _ = engine
            .events
            .push(EngineOutput::SendFailed(SendFailedEvent {
                channel_id: packet.channel_id,
                message_id: packet.message_id,
                reason: SendFailureReason::RetryLimitReached,
            }));
    }

    for packet in retransmits[..retransmit_len].iter().flatten() {
        let message_failed = failed_messages[..failed_message_len]
            .iter()
            .flatten()
            .any(|key| *key == (packet.channel_id, packet.message_id));

        if message_failed {
            continue;
        }

        engine
            .in_flight
            .note_retransmit(packet.packet_number, now_ms);
        let _ = engine.events.push(EngineOutput::Write(WriteEvent {
            packet_number: packet.packet_number,
            bytes: packet.bytes,
            len: packet.len,
        }));
    }
}