ace_sim/
bus.rs

1// region: Imports
2
3use crate::clock::{Clock, Duration, Instant, SimClock};
4use crate::fault::FaultConfig;
5use crate::io::NodeAddress;
6use crate::rng::{Rng, Xorshift64};
7use ace_core::Vec;
8
9// endregion: Imports
10
11// region: Envelope
12
13/// A message in-flight on the simulation bus.
14#[derive(Debug, Clone)]
15pub struct Envelope<const MAX_DATA: usize> {
16    pub src: NodeAddress,
17    pub dst: NodeAddress,
18    pub data: Vec<u8, MAX_DATA>,
19    /// Earliest time this message may be delivered.
20    pub deliver_at: Instant,
21}
22
23// endregion: Envelope
24
25// region: SimBus
26
27/// The simulation message bus.
28///
29/// Connects all nodes in the simulation. Drives time forward, delivers messages, and injects
30/// faults according to [`FaultConfig`].
31///
32/// `N` - max message payload bytes
33/// `Q` - max messages in-flight simultaneously
34#[derive(Debug)]
35pub struct SimBus<const MAX_DATA: usize, const MAX_QUEUED: usize> {
36    clock: SimClock,
37    rng: Xorshift64,
38    faults: FaultConfig,
39    queue: Vec<Envelope<MAX_DATA>, MAX_QUEUED>,
40}
41
42impl<const MAX_DATA: usize, const MAX_QUEUED: usize> SimBus<MAX_DATA, MAX_QUEUED> {
43    pub fn new(seed: u64, faults: FaultConfig) -> Self {
44        Self {
45            clock: SimClock::new(),
46            rng: Xorshift64::new(seed),
47            faults,
48            queue: Vec::new(),
49        }
50    }
51
52    /// Returns the current simulation time.
53    pub fn now(&self) -> Instant {
54        self.clock.now()
55    }
56
57    /// Advances simulation time by `duration` and returns all messages that are due for delivery
58    /// at or before the new time.
59    pub fn tick(&mut self, duration: Duration) -> Vec<Envelope<MAX_DATA>, MAX_QUEUED> {
60        self.clock.advance(duration);
61        let now = self.clock.now();
62
63        let mut delivered = Vec::new();
64        let mut remaining = Vec::new();
65
66        for envelope in self.queue.drain(..) {
67            if envelope.deliver_at <= now {
68                let _ = delivered.push(envelope);
69            } else {
70                let _ = remaining.push(envelope);
71            }
72        }
73
74        if delivered.len() > 1 {
75            for i in 0..delivered.len() - 1 {
76                if self
77                    .rng
78                    .chance(self.faults.message_reorder.0, self.faults.message_reorder.1)
79                {
80                    delivered.swap(i, i + 1);
81                }
82            }
83        }
84
85        self.queue = remaining;
86        delivered
87    }
88
89    /// Enqueues a message from `src` to `dst` with fault injection applied.
90    ///
91    /// Returns `true` if the message was enqueued, `false` if it was dropped by fault injection or
92    /// the queue is full
93    pub fn send(&mut self, src: NodeAddress, dst: NodeAddress, data: &[u8]) -> bool {
94        let r = self
95            .rng
96            .chance(self.faults.message_loss.0, self.faults.message_loss.1);
97
98        if r {
99            return false;
100        }
101
102        if self
103            .rng
104            .chance(self.faults.timeout.0, self.faults.timeout.1)
105        {
106            return false;
107        }
108
109        let mut payload = Vec::new();
110
111        for &byte in data {
112            if self
113                .rng
114                .chance(self.faults.corruption.0, self.faults.corruption.1)
115            {
116                let _ = payload.push(byte ^ self.rng.next_u8());
117            } else {
118                let _ = payload.push(byte);
119            }
120        }
121
122        let deliver_at = if self
123            .rng
124            .chance(self.faults.message_delay.0, self.faults.message_delay.1)
125        {
126            let delay_us = self.rng.next_u64() % self.faults.max_delay.as_micros().max(1);
127            self.clock.now() + Duration::from_micros(delay_us)
128        } else {
129            self.clock.now()
130        };
131
132        let envelope = Envelope {
133            src,
134            dst,
135            data: payload,
136            deliver_at,
137        };
138
139        if self.queue.len() >= MAX_QUEUED {
140            return false;
141        }
142
143        let _ = self.queue.push(envelope);
144
145        true
146    }
147
148    /// Returns a reference to the current fault config
149    pub fn faults(&self) -> &FaultConfig {
150        &self.faults
151    }
152
153    /// Returns the fault config - allows escalating fault severity during a simulation run.
154    pub fn set_faults(&mut self, faults: FaultConfig) {
155        self.faults = faults;
156    }
157
158    /// Returns the RNG seed-derived next value - useful for injecting spontaneous NRCs at the node
159    /// level.
160    pub fn next_u8(&mut self) -> u8 {
161        self.rng.next_u8()
162    }
163
164    pub fn chance(&mut self, numerator: u32, denominator: u32) -> bool {
165        self.rng.chance(numerator, denominator)
166    }
167}
168
169// endregion: SimBus
ace_sim/bus.rs

ace_sim/
bus.rs
