diff --git a/src/main.rs b/src/main.rs index 7f0fb48..79cf7d9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -10,6 +10,8 @@ use alloc::string::ToString as _; use alloc::vec::Vec; use core::cell::RefCell; +#[cfg(feature = "spat")] +use core::cell::UnsafeCell; use c_its_parser::gn as geonetworking; use critical_section::Mutex; @@ -37,6 +39,8 @@ mod radio; mod screen; const WIFI_CHANNEL: radio::Channel = 180; +#[cfg(feature = "spat")] +const SPAT_RATE_LIMIT: u8 = 5; // keep every n-th message // This creates a default app-descriptor required by the esp-idf bootloader. // For more information see: @@ -212,7 +216,7 @@ async fn main(spawner: Spawner) -> ! { use c_its_parser::ItsMessage; // use c_its_parser::standards::extensions::ItsMessageId; - // debug!("Got new {:?} message", ItsMessageId::from(&msg)); + // info!("Got new {:?} message", ItsMessageId::from(&msg)); match msg { #[cfg(feature = "spat")] @@ -272,7 +276,28 @@ async fn main(spawner: Spawner) -> ! { clippy::needless_pass_by_value, reason = "adhering to callback interface" )] +#[allow(clippy::too_many_lines)] fn handle_frame(frame: wifi::sniffer::PromiscuousPkt<'_>) { + #[cfg(feature = "spat")] + const PRUNE_INTERVAL: chrono::Duration = chrono::Duration::seconds(60); + #[cfg(feature = "spat")] + static mut RATELIMIT_CACHE_LAST_PRUNE: UnsafeCell = + UnsafeCell::new(chrono::NaiveDateTime::MIN); + + #[cfg(feature = "spat")] + static mut SPAT_RATELIMIT_CACHE: UnsafeCell> = UnsafeCell::new( + cache::Cache::::new(chrono::Duration::seconds(20)), + ); + + #[cfg(feature = "spat")] + let now = { + let time = esp_hal::time::Instant::now().duration_since_epoch(); + // unwrap is fine since value range is u64 microseconds + chrono::DateTime::from_timestamp(time.as_secs().cast_signed(), 0) + .unwrap() + .naive_utc() + }; + // Ignore frames with errors and non-data frames if frame.rx_cntl.rx_state != 0 { warn!("Received frame has RX error: {}", frame.rx_cntl.rx_state); @@ -315,16 +340,37 @@ fn handle_frame(frame: wifi::sniffer::PromiscuousPkt<'_>) { } // "parse" ITS PDU Header - if let Ok((payload, message_id, _)) = get_its_header(&packet) { - // drop all unsupported message IDs + if let Ok((payload, message_id, station_id)) = get_its_header(&packet) { + // drop all unsupported message IDs and rate-limit SPATEMs if match message_id { #[cfg(feature = "denm")] c_its_parser::standards::extensions::ItsMessageId::Denm => false, #[cfg(feature = "cam")] c_its_parser::standards::extensions::ItsMessageId::Cam => false, #[cfg(feature = "spat")] - c_its_parser::standards::extensions::ItsMessageId::Spatem - | c_its_parser::standards::extensions::ItsMessageId::Mapem => false, + c_its_parser::standards::extensions::ItsMessageId::Spatem => { + // reduce SPAT rate per station ID + let mut drop_msg = false; + unsafe { + #[allow( + static_mut_refs, + reason = "this function will only run consecutive" + )] + let spat_cache = SPAT_RATELIMIT_CACHE.get(); + + (*spat_cache).update_or_init(now, station_id, |i| { + i.msg_count = i.msg_count.wrapping_add(1); + + if (i.msg_count % SPAT_RATE_LIMIT) > 0 { + drop_msg = true; + } + }); + } + + drop_msg + } + #[cfg(feature = "spat")] + c_its_parser::standards::extensions::ItsMessageId::Mapem => false, _ => true, } { return; @@ -343,6 +389,39 @@ fn handle_frame(frame: wifi::sniffer::PromiscuousPkt<'_>) { } } } + + #[cfg(feature = "spat")] + unsafe { + #[allow(static_mut_refs, reason = "this function will only run consecutive")] + let last_prune_time = RATELIMIT_CACHE_LAST_PRUNE.get(); + if *last_prune_time + PRUNE_INTERVAL < now { + #[allow(static_mut_refs, reason = "this function will only run consecutive")] + let spat_cache = SPAT_RATELIMIT_CACHE.get(); + (*spat_cache).prune(now); + + *last_prune_time = now; + } + } +} + +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +struct RatelimitData { + station_id: u32, + msg_count: u8, +} + +impl cache::Cachable for RatelimitData { + fn key(&self) -> u32 { + self.station_id + } +} +impl cache::Initable for RatelimitData { + fn init(key: u32) -> Self { + Self { + station_id: key, + ..Default::default() + } + } } fn get_its_header<'p>(