maybenot_simulator/
queue_event.rs

1use std::collections::BinaryHeap;
2
3use maybenot::TriggerEvent;
4
5use crate::{SimEvent, event_to_usize};
6
7use std::time::{Duration, Instant};
8
9#[derive(Debug, Clone, PartialEq)]
10pub enum Queue {
11    Blocking,
12    Bypassable,
13    Internal,
14    Base,
15}
16
17/// EventQueue represents the queue of events that are waiting to be processed
18/// in order (time-wise). The queue is split into four parts:
19/// - base: TriggerEvent::NormalSent events that are from the parsed base trace
20/// - blocking: TunnelSent events that may be blocked by blocking machines
21/// - bypassable: TunnelSent events that are blocked with bypassable blocking
22/// - internal: all other events
23#[derive(Debug, Clone)]
24pub struct EventQueue {
25    pub(crate) base: BinaryHeap<SimEvent>,
26    pub(crate) blocking: BinaryHeap<SimEvent>,
27    pub(crate) bypassable: BinaryHeap<SimEvent>,
28    pub(crate) internal: BinaryHeap<SimEvent>,
29}
30
31impl Default for EventQueue {
32    fn default() -> Self {
33        Self::new()
34    }
35}
36
37impl EventQueue {
38    pub fn new() -> EventQueue {
39        EventQueue {
40            // TriggerEvent::NormalSent is the only event in the base trace
41            base: BinaryHeap::with_capacity(4096),
42            // TriggerEvent::TunnelSent is the only event that can be blocking
43            // or bypassable
44            blocking: BinaryHeap::with_capacity(1024),
45            bypassable: BinaryHeap::with_capacity(1024),
46            // all events that are not TriggerEvent::TunnelSent or
47            // TriggerEvent::NormalSent are internal
48            internal: BinaryHeap::with_capacity(1024),
49        }
50    }
51
52    pub fn len(&self) -> usize {
53        self.blocking.len() + self.bypassable.len() + self.internal.len() + self.base.len()
54    }
55
56    pub fn is_empty(&self) -> bool {
57        self.len() == 0
58    }
59
60    /// Checks if there are no normal packets in the queue. This involves
61    /// checking the base queue for TriggerEvent::NormalSent events, the
62    /// blocking and bypassable queues for TriggerEvent::TunnelSent events
63    /// without the padding flag, and the internal queue for any TunnelRecv
64    /// without the padding flag.
65    pub fn no_normal_packets(&self) -> bool {
66        self.base.is_empty()
67            && self
68                .blocking
69                .iter()
70                .all(|e| e.event != TriggerEvent::TunnelSent && !e.contains_padding)
71            && self
72                .bypassable
73                .iter()
74                .all(|e| e.event != TriggerEvent::TunnelSent && !e.contains_padding)
75            && self
76                .internal
77                .iter()
78                .all(|e| e.event != TriggerEvent::TunnelRecv && !e.contains_padding)
79    }
80
81    pub fn push(&mut self, item: SimEvent) {
82        match item.event {
83            TriggerEvent::TunnelSent => match item.bypass {
84                true => self.bypassable.push(item),
85                false => self.blocking.push(item),
86            },
87            // from parse_trace_advanced(), the only place where we push
88            // TriggerEvent::NormalSent from a base trace
89            TriggerEvent::NormalSent => {
90                self.base.push(item);
91            }
92            _ => {
93                self.internal.push(item);
94            }
95        }
96    }
97
98    pub fn peek(
99        &self,
100        network_delay_sum: Duration,
101        current_time: Instant,
102    ) -> (Option<&SimEvent>, Queue, Duration) {
103        match self.len() {
104            0 => (None, Queue::Blocking, Duration::default()),
105            _ => {
106                // peek all, per def, it's one of them: we prioritize in order
107                // of base, bypassable, blocking, and lastly internal
108                let (mut first, mut q) = (self.bypassable.peek(), Queue::Bypassable);
109
110                let n = self.blocking.peek();
111                if n > first {
112                    first = n;
113                    q = Queue::Blocking;
114                }
115
116                let n = self.internal.peek();
117                if n > first {
118                    first = n;
119                    q = Queue::Internal;
120                }
121
122                // for the base queue, we need to consider the network delay sum
123                // to determine the actual time of the event
124                let duration_since: Duration;
125                let n = self.base.peek();
126                if before(n, first, network_delay_sum) {
127                    first = n;
128                    q = Queue::Base;
129                    duration_since =
130                        (first.unwrap().time + network_delay_sum).duration_since(current_time);
131                } else {
132                    duration_since = first.unwrap().time.duration_since(current_time);
133                }
134                (first, q, duration_since)
135            }
136        }
137    }
138
139    /// remove an event from the queue
140    pub fn pop(&mut self, q: Queue, network_delay_sum: Duration) -> Option<SimEvent> {
141        match q {
142            Queue::Blocking => self.blocking.pop(),
143            Queue::Bypassable => self.bypassable.pop(),
144            Queue::Internal => self.internal.pop(),
145            Queue::Base => {
146                if network_delay_sum == Duration::default() {
147                    self.base.pop()
148                } else {
149                    let mut item = self.base.pop().unwrap();
150                    item.time += network_delay_sum;
151                    Some(item)
152                }
153            }
154        }
155    }
156
157    /// peek the next blocking event
158    pub fn peek_blocking(&self) -> Option<&SimEvent> {
159        self.blocking.peek()
160    }
161
162    /// peek the next bypassable event
163    pub fn peek_bypassable(&self) -> Option<&SimEvent> {
164        self.bypassable.peek()
165    }
166
167    /// peek the next non-blocking event
168    pub fn peek_non_blocking(&self, network_delay_sum: Duration) -> (Option<&SimEvent>, Queue) {
169        let b = self.base.peek();
170        let i = self.internal.peek();
171        if before(b, i, network_delay_sum) {
172            (b, Queue::Base)
173        } else {
174            (i, Queue::Internal)
175        }
176    }
177
178    /// get the first time of the base queue: should only be used for the
179    /// simulator's current time at startup
180    pub fn get_first_base_time(&self) -> Option<Instant> {
181        self.base.peek().map(|e| e.time)
182    }
183}
184
185// determine if a, with network delay sum, is before or at b: uses the same
186// ordering as the binary heap, from SimEvent::cmp()
187fn before(a: Option<&SimEvent>, b: Option<&SimEvent>, a_network_delay_sum: Duration) -> bool {
188    match (a, b) {
189        (Some(a), Some(b)) => {
190            let a_time = a.time + a_network_delay_sum;
191            let b_time = b.time;
192            let ordering = a_time
193                .cmp(&b_time)
194                .then_with(|| event_to_usize(&a.event).cmp(&event_to_usize(&b.event)));
195            // prefer a if it's equal, since it's the base event
196            ordering == std::cmp::Ordering::Less || ordering == std::cmp::Ordering::Equal
197        }
198        (Some(_), None) => true,
199        _ => false,
200    }
201}
maybenot_simulator/queue_event.rs

maybenot_simulator/
queue_event.rs
