Send data loss events

Author: Christopher-06

rustmeter-beacon/rustmeter-beacon-target/src/cortex/tracing_rtt.rs

@@ -1,22 +1,53 @@
-use core::sync::atomic::Ordering;
-
 use crate::cortex::rtt_minimal::rtt_write_core;
+use core::sync::atomic::Ordering;
+use rustmeter_beacon_core::{protocol::raw_writers::event_ids, time_delta::TimeDelta};
 
-// TODO: Move Dropped Events Counter to rtt_minimal::rtt_write_core and core-specific dropped counters
-static DROPPED_EVENTS_COUNTER: portable_atomic::AtomicU32 = portable_atomic::AtomicU32::new(0);
+// static mut okay because accessed in critical section
+static mut DROPPED_EVENTS_COUNTER: [u32; 2] = [0, 0];
 
 #[unsafe(no_mangle)]
-// #[unsafe(link_section = ".iram1")] ESP32
 #[unsafe(link_section = ".data")]
 fn write_tracing_data(data: &[u8]) {
     let core_id = crate::core_id::get_current_core_id() as usize;
     let idx = if core_id > 1 { 0 } else { core_id as usize };
 
     // Write to Buffer
     cortex_m::interrupt::free(|_| {
-        let written = rtt_write_core(idx, data);
-        if written.is_none() {
-            DROPPED_EVENTS_COUNTER.fetch_add(1, Ordering::Relaxed);
+        // Check for dropped events
+        let dropped_events = unsafe { DROPPED_EVENTS_COUNTER[idx] };
+        if dropped_events > 0 {
+            // Create dropped events event
+            let mut buffer = [0u8; 12];
+            buffer[0] = event_ids::DATA_LOSS_EVENT << 3;
+            buffer[1..5].copy_from_slice(&dropped_events.to_le_bytes());
+            let timestamp = TimeDelta::from_now();
+            let pos = timestamp.write_bytes_mut(&mut buffer[5..]);
+
+            // Try to write dropped events data
+            match rtt_write_core(idx, &buffer[..5 + pos]) {
+                Some(_) => {
+                    // Successfully written, reset counter
+                    unsafe {
+                        DROPPED_EVENTS_COUNTER[idx] = 0;
+                    }
+                }
+                None => {
+                    // Still cannot write, increment counter
+                    unsafe {
+                        DROPPED_EVENTS_COUNTER[idx] += 1;
+                    }
+                }
+            }
+
+            // do not send actual data this round because we took TimeDelta::now() and it would be inconsistent
+        } else {
+            // No dropped events, try to write actual data
+            let written = rtt_write_core(idx, data);
+            if written.is_none() {
+                unsafe {
+                    DROPPED_EVENTS_COUNTER[idx] += 1;
+                }
+            }
         }
     });
 }

rustmeter-beacon/rustmeter-beacon-target/src/espressif/tracing_esp.rs

@@ -11,10 +11,18 @@ use embassy_sync::{
     blocking_mutex::{Mutex, raw::CriticalSectionRawMutex},
     pipe::{Pipe, TryWriteError},
 };
-use rustmeter_beacon_core::{buffer::BufferWriter, protocol::EventPayload, time_delta::TimeDelta};
+use rustmeter_beacon_core::{
+    buffer::BufferWriter,
+    protocol::{EventPayload, raw_writers::event_ids},
+    time_delta::TimeDelta,
+};
 
 use crate::{
-    NUM_CORES, espressif::local_critical_section::{enter_local_critical_section, exit_local_critical_section}, ringbuffer::{AtomicRingBuffer, SimpleRingBuffer}
+    NUM_CORES,
+    espressif::local_critical_section::{
+        enter_local_critical_section, exit_local_critical_section,
+    },
+    ringbuffer::{AtomicRingBuffer, SimpleRingBuffer},
 };
 
 const BUFFER_SIZE: usize = 4096;
@@ -28,7 +36,8 @@ static TRACE_BUFFERS: [PerCoreSync<AtomicRingBuffer<BUFFER_SIZE>>; NUM_CORES] =
 static NEW_DATA_SIGNAL: embassy_sync::signal::Signal<CriticalSectionRawMutex, ()> =
     embassy_sync::signal::Signal::new();
 
-static DROPPED_EVENTS_COUNTER: portable_atomic::AtomicU32 = portable_atomic::AtomicU32::new(0);
+// static mut okay because accessed in critical section
+static mut DROPPED_EVENTS_COUNTER: [u32; 2] = [0, 0];
 
 pub fn get_tracing_buffers_and_signaller() -> (
     &'static [PerCoreSync<AtomicRingBuffer<BUFFER_SIZE>>; NUM_CORES],
@@ -45,15 +54,47 @@ fn write_tracing_data(data: &[u8]) {
 
     // Enter local critical section
     let prev_interrupt_state = enter_local_critical_section();
-
-    // Write to Buffer
     let buf = unsafe { &mut *TRACE_BUFFERS[core_id].get() };
-    if let Some(free) = buf.push_slice_fast(data) {
-        if free < (buf.capacity() * 3 / 4)  {
-            NEW_DATA_SIGNAL.signal(());
+
+    let dropped_events = unsafe { DROPPED_EVENTS_COUNTER[idx] };
+    if dropped_events > 0 {
+        // Create dropped events event
+        let mut buffer = [0u8; 12];
+        buffer[0] = event_ids::DATA_LOSS_EVENT << 3;
+        buffer[1..5].copy_from_slice(&dropped_events.to_le_bytes());
+        let timestamp = TimeDelta::from_now();
+        let pos = timestamp.write_bytes_mut(&mut buffer[5..]);
+
+        // Try to write dropped events data
+        match buf.push_slice_fast(&buffer[..5 + pos]) {
+            Some(_) => {
+                // Successfully written, reset counter
+                unsafe {
+                    DROPPED_EVENTS_COUNTER[idx] = 0;
+                }
+            }
+            None => {
+                // Still cannot write, increment counter
+                unsafe {
+                    DROPPED_EVENTS_COUNTER[idx] += 1;
+                }
+            }
         }
+
+        NEW_DATA_SIGNAL.signal(()); // signal because we added data
+
+    // do not send actual data this round because we took TimeDelta::now() and it would be inconsistent
     } else {
-        DROPPED_EVENTS_COUNTER.fetch_add(1, Ordering::Relaxed);
+        // write actual data
+        if let Some(free) = buf.push_slice_fast(data) {
+            if free < (buf.capacity() * 3 / 4) {
+                NEW_DATA_SIGNAL.signal(()); // signal when buffer is less than 75% free <=> 25% used
+            }
+        } else {
+            unsafe {
+                DROPPED_EVENTS_COUNTER[idx] += 1;
+            }
+        }
     }
 
     // Exit local critical section