use msrt::{
Engine, EngineConfig,
core::{ChannelId, PacketType},
engine::{EnginePoll, ReceiveReport},
};
const TX_BUF_BYTES: usize = 128;
#[test]
fn reliable_transport_survives_drops_corruption_reordering_and_duplex_load() {
let mut mac = Engine::new(EngineConfig {
fragment_bytes: 8,
retransmit_timeout_ms: 1,
..EngineConfig::default()
});
let mut mcu = Engine::new(EngineConfig {
fragment_bytes: 9,
retransmit_timeout_ms: 1,
..EngineConfig::default()
});
let nav = ChannelId::new(16);
let telemetry = ChannelId::new(17);
let mac_messages = [
ExpectedMessage::new(nav, b"mac nav message one"),
ExpectedMessage::new(telemetry, b"mac telemetry message two"),
ExpectedMessage::new(nav, b"mac command message three"),
];
let mcu_messages = [
ExpectedMessage::new(telemetry, b"mcu telemetry response one"),
ExpectedMessage::new(nav, b"mcu nav response two"),
ExpectedMessage::new(telemetry, b"mcu status response three"),
];
let mut link = SimLink::new();
let mut mac_delivered = DeliveredMessages::new();
let mut mcu_delivered = DeliveredMessages::new();
for message in mac_messages {
mac.send_on(message.channel_id, message.bytes)
.expect("queue mac message");
}
for message in mcu_messages {
mcu.send_on(message.channel_id, message.bytes)
.expect("queue mcu message");
}
for _ in 0..64 {
pump_until_idle(
&mut mac,
&mut mcu,
&mut link,
&mut mac_delivered,
&mut mcu_delivered,
);
link.flush_reordered(&mut mac, &mut mcu);
pump_until_idle(
&mut mac,
&mut mcu,
&mut link,
&mut mac_delivered,
&mut mcu_delivered,
);
if mac_delivered.contains_all(&mcu_messages) && mcu_delivered.contains_all(&mac_messages) {
assert_no_unexpected_events(&mut mac);
assert_no_unexpected_events(&mut mcu);
return;
}
}
panic!(
"simulation did not deliver all messages: mac={:?}, mcu={:?}",
mac_delivered, mcu_delivered
);
}
fn pump_until_idle(
mac: &mut Engine,
mcu: &mut Engine,
link: &mut SimLink,
mac_delivered: &mut DeliveredMessages,
mcu_delivered: &mut DeliveredMessages,
) {
for _ in 0..256 {
let mut progressed = false;
progressed |= pump_one(mac, mcu, &mut link.mac_to_mcu, mac_delivered);
progressed |= pump_one(mcu, mac, &mut link.mcu_to_mac, mcu_delivered);
if !progressed {
break;
}
}
}
fn pump_one(
src: &mut Engine,
dst: &mut Engine,
direction: &mut SimDirection,
delivered: &mut DeliveredMessages,
) -> bool {
let mut tx_buf = [0; TX_BUF_BYTES];
match src.poll(0, &mut tx_buf).expect("poll engine") {
EnginePoll::Transmit { bytes, .. } => {
let write = SimWrite::from_bytes(bytes);
direction.deliver(dst, write);
true
}
EnginePoll::Message(message) => {
delivered.push(message.channel_id, message.as_bytes());
true
}
EnginePoll::SendFailed(failed) => {
panic!("reliable simulation should not fail sends: {failed:?}");
}
EnginePoll::Idle => false,
}
}
fn assert_no_unexpected_events(engine: &mut Engine) {
let mut tx_buf = [0; TX_BUF_BYTES];
loop {
match engine.poll(0, &mut tx_buf).expect("poll engine") {
EnginePoll::Transmit { .. } | EnginePoll::Message(_) => {}
EnginePoll::SendFailed(failed) => {
panic!("simulation completed but found send failure: {failed:?}");
}
EnginePoll::Idle => break,
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct ExpectedMessage {
channel_id: ChannelId,
bytes: &'static [u8],
}
impl ExpectedMessage {
const fn new(channel_id: ChannelId, bytes: &'static [u8]) -> Self {
Self { channel_id, bytes }
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
struct DeliveredMessages {
messages: [(ChannelId, [u8; 64], usize); 8],
len: usize,
}
impl DeliveredMessages {
const fn new() -> Self {
Self {
messages: [(ChannelId::DEFAULT, [0; 64], 0); 8],
len: 0,
}
}
fn push(&mut self, channel_id: ChannelId, bytes: &[u8]) {
assert!(
self.len < self.messages.len(),
"delivered message buffer full"
);
assert!(bytes.len() <= 64, "delivered message too large for test");
let mut stored = [0; 64];
stored[..bytes.len()].copy_from_slice(bytes);
self.messages[self.len] = (channel_id, stored, bytes.len());
self.len += 1;
}
fn contains_all(&self, expected: &[ExpectedMessage]) -> bool {
expected
.iter()
.all(|message| self.contains(message.channel_id, message.bytes))
}
fn contains(&self, channel_id: ChannelId, bytes: &[u8]) -> bool {
self.messages[..self.len]
.iter()
.any(|(current_channel, current_bytes, current_len)| {
*current_channel == channel_id && ¤t_bytes[..*current_len] == bytes
})
}
}
#[derive(Debug)]
struct SimLink {
mac_to_mcu: SimDirection,
mcu_to_mac: SimDirection,
}
impl SimLink {
const fn new() -> Self {
Self {
mac_to_mcu: SimDirection::new(),
mcu_to_mac: SimDirection::new(),
}
}
fn flush_reordered(&mut self, mac: &mut Engine, mcu: &mut Engine) {
self.mac_to_mcu.flush(mcu);
self.mcu_to_mac.flush(mac);
}
}
#[derive(Debug)]
struct SimDirection {
seen_data: [bool; 64],
held: [Option<SimWrite>; 8],
held_len: usize,
}
impl SimDirection {
const fn new() -> Self {
Self {
seen_data: [false; 64],
held: [None; 8],
held_len: 0,
}
}
fn deliver(&mut self, dst: &mut Engine, write: SimWrite) {
if !is_data(write) {
receive_ok(dst, write.as_bytes());
return;
}
let packet_number = packet_number(write.as_bytes()) as usize;
let first_seen = packet_number < self.seen_data.len() && !self.seen_data[packet_number];
if packet_number < self.seen_data.len() {
self.seen_data[packet_number] = true;
}
if first_seen && packet_number % 4 == 3 {
self.hold(write);
return;
}
receive_ok(dst, write.as_bytes());
}
fn hold(&mut self, write: SimWrite) {
assert!(self.held_len < self.held.len(), "held packet buffer full");
self.held[self.held_len] = Some(write);
self.held_len += 1;
}
fn flush(&mut self, dst: &mut Engine) {
while self.held_len > 0 {
self.held_len -= 1;
let write = self.held[self.held_len].take().expect("held packet");
receive_ok(dst, write.as_bytes());
}
}
}
fn receive_ok(dst: &mut Engine, bytes: &[u8]) {
match dst.receive(bytes) {
ReceiveReport::Packet { .. }
| ReceiveReport::Duplicate { .. }
| ReceiveReport::Ack { .. }
| ReceiveReport::Ping { .. }
| ReceiveReport::Pong { .. } => {}
other => panic!("unexpected receive report in simulation: {other:?}"),
}
}
fn is_data(write: SimWrite) -> bool {
write.as_bytes().get(8).copied() == Some(PacketType::Data.code())
}
fn packet_number(bytes: &[u8]) -> u32 {
u32::from_le_bytes(
bytes[msrt::wire::WIRE_HEADER_LEN + 2..msrt::wire::WIRE_HEADER_LEN + 6]
.try_into()
.expect("packet number bytes"),
)
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct SimWrite {
bytes: [u8; TX_BUF_BYTES],
len: usize,
}
impl SimWrite {
fn from_bytes(bytes: &[u8]) -> Self {
let mut stored = [0; TX_BUF_BYTES];
stored[..bytes.len()].copy_from_slice(bytes);
Self {
bytes: stored,
len: bytes.len(),
}
}
fn as_bytes(&self) -> &[u8] {
&self.bytes[..self.len]
}
}