v2x: Reduce SPAT rate

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: <https://docs.espressif.com/projects/esp-idf/en/stable/esp32/api-reference/system/app_image_format.html#application-description>
@@ -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<chrono::NaiveDateTime> =
+        UnsafeCell::new(chrono::NaiveDateTime::MIN);
+
+    #[cfg(feature = "spat")]
+    static mut SPAT_RATELIMIT_CACHE: UnsafeCell<cache::Cache<u32, RatelimitData>> = UnsafeCell::new(
+        cache::Cache::<u32, RatelimitData>::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<u32> for RatelimitData {
+    fn key(&self) -> u32 {
+        self.station_id
+    }
+}
+impl cache::Initable<Self, u32> for RatelimitData {
+    fn init(key: u32) -> Self {
+        Self {
+            station_id: key,
+            ..Default::default()
+        }
+    }
 }
 
 fn get_its_header<'p>(