retina_core/conntrack/
timerwheel.rs

1use crate::conntrack::{Conn, ConnId};
2use crate::subscription::{Subscription, Trackable};
3
4use crossbeam_channel::{tick, Receiver};
5use hashlink::linked_hash_map::LinkedHashMap;
6use hashlink::linked_hash_map::RawEntryMut;
7use std::collections::VecDeque;
8use std::time::{Duration, Instant};
9
10/// Tracks inactive connection expiration.
11pub(super) struct TimerWheel {
12    /// Period to check for inactive connections (in milliseconds).
13    period: usize,
14    /// Start time of the `TimerWheel`.
15    start_ts: Instant,
16    /// Timeout ticker, fires every `period` milliseconds.
17    ticker: Receiver<Instant>,
18    /// Index of the next bucket to expire.
19    next_bucket: usize,
20    /// List of timers.
21    timers: Vec<VecDeque<ConnId>>,
22}
23
24impl TimerWheel {
25    /// Creates a new `TimerWheel` with a maximum timeout of `max_timeout` and a timeout check
26    /// period of `timeout_resolution`.
27    pub(super) fn new(max_timeout: usize, timeout_resolution: usize) -> Self {
28        if timeout_resolution > max_timeout {
29            panic!("Timeout check period must be smaller than maximum inactivity timeout")
30        }
31        let start_ts = Instant::now();
32        let ticker = tick(Duration::from_millis(timeout_resolution as u64));
33        TimerWheel {
34            period: timeout_resolution,
35            start_ts,
36            ticker,
37            next_bucket: 0,
38            timers: vec![VecDeque::new(); max_timeout / timeout_resolution],
39        }
40    }
41
42    /// Insert a new connection ID into the timerwheel.
43    #[inline]
44    pub(super) fn insert(
45        &mut self,
46        conn_id: &ConnId,
47        last_seen_ts: Instant,
48        inactivity_window: usize,
49    ) {
50        let current_time = (last_seen_ts - self.start_ts).as_millis() as usize;
51        let timer_index = ((current_time + inactivity_window) / self.period) % self.timers.len();
52        log::debug!("Inserting into index: {}, {:?}", timer_index, current_time);
53        self.timers[timer_index].push_back(conn_id.to_owned());
54    }
55
56    /// Checks for and remove inactive connections.
57    #[inline]
58    pub(super) fn check_inactive<T: Trackable>(
59        &mut self,
60        table: &mut LinkedHashMap<ConnId, Conn<T>>,
61        subscription: &Subscription<T::Subscribed>,
62    ) {
63        let table_len = table.len();
64        if let Ok(now) = self.ticker.try_recv() {
65            let nb_removed = self.remove_inactive(now, table, subscription);
66            log::debug!(
67                "expired: {} ({})",
68                nb_removed,
69                nb_removed as f64 / table_len as f64
70            );
71            log::debug!("new table size: {}", table.len());
72        }
73    }
74
75    /// Removes connections that have been inactive for at least their inactivity window time
76    /// period.
77    ///
78    /// Returns the number of connections removed.
79    #[inline]
80    pub(super) fn remove_inactive<T: Trackable>(
81        &mut self,
82        now: Instant,
83        table: &mut LinkedHashMap<ConnId, Conn<T>>,
84        subscription: &Subscription<T::Subscribed>,
85    ) -> usize {
86        let period = self.period;
87        let nb_buckets = self.timers.len();
88        let mut not_expired: Vec<(usize, ConnId)> = vec![];
89        let check_time = (now - self.start_ts).as_millis() as usize / period * period;
90
91        let mut cnt_exp = 0;
92        let last_expire_bucket = check_time / period;
93        log::debug!(
94            "check time: {}, next: {}, last: {}",
95            check_time,
96            self.next_bucket,
97            last_expire_bucket
98        );
99
100        for expire_bucket in self.next_bucket..last_expire_bucket {
101            log::debug!(
102                "bucket: {}, index: {}",
103                expire_bucket,
104                expire_bucket % nb_buckets
105            );
106            let list = &mut self.timers[expire_bucket % nb_buckets];
107
108            for conn_id in list.drain(..) {
109                if let RawEntryMut::Occupied(mut occupied) =
110                    table.raw_entry_mut().from_key(&conn_id)
111                {
112                    let conn = occupied.get_mut();
113                    let last_seen_time = (conn.last_seen_ts - self.start_ts).as_millis() as usize;
114                    log::debug!("Last seen time: {}", last_seen_time);
115                    let expire_time = last_seen_time + conn.inactivity_window;
116                    if expire_time < check_time {
117                        cnt_exp += 1;
118                        conn.terminate(subscription);
119                        occupied.remove();
120                    } else {
121                        let timer_index = (expire_time / period) % nb_buckets;
122                        not_expired.push((timer_index, conn_id));
123                    }
124                }
125            }
126            for (timer_index, conn_id) in not_expired.drain(..) {
127                self.timers[timer_index].push_back(conn_id);
128            }
129        }
130        self.next_bucket = last_expire_bucket;
131        cnt_exp
132    }
133}
retina_core/conntrack/timerwheel.rs

retina_core/conntrack/
timerwheel.rs
