rust/ffi: add safe thread storage wrapper

Add a typed ThreadStorage<T> wrapper around the thread storage bindings.

Ticket: #8445

doc/userguide/devguide/extending/threads.rst

@@ -38,3 +38,49 @@ registered. Registered callbacks are kept for the Suricata process lifetime.
 ``ThreadVars`` carries a lifetime tied to the callback invocation, so the
 borrow checker prevents it from being stored beyond the call. Rust callbacks
 must not panic, as they are invoked across an FFI boundary.
+
+Thread Storage
+==============
+
+``thread::ThreadStorage<T>`` provides typed, per-thread storage backed by
+Suricata's thread storage API. Each registered slot holds an independent value
+of type ``T`` for every thread.
+
+Register a slot once during initialization with
+``ThreadStorage::<T>::register``. Registration must happen before Suricata
+finalizes its storage registration, which is the case during plugin
+initialization.
+
+.. code-block:: rust
+
+   use suricata_ffi::thread::{self, ThreadStorage, ThreadVars};
+
+   #[derive(Default)]
+   struct ThreadState {
+       flows: u64,
+   }
+
+   fn register(storage: ThreadStorage<ThreadState>) -> Result<(), &'static str> {
+       thread::register_init_callback(move |tv| on_thread_init(storage, tv))
+   }
+
+Values are owned by Suricata's thread storage and are dropped automatically when
+the thread's storage is freed.
+
+Access the value for a thread through the ``ThreadVars`` wrapper. ``get`` takes
+``&ThreadVars`` and returns ``Option<&T>``. ``get_mut`` takes ``&mut
+ThreadVars`` and returns ``Option<&mut T>``. ``get_or_insert_with`` also takes
+``&mut ThreadVars`` and returns ``Result<&mut T, _>``, inserting a value
+produced by the closure if the slot is empty:
+
+.. code-block:: rust
+
+   fn on_thread_init(storage: ThreadStorage<ThreadState>, tv: &mut ThreadVars) {
+       let _ = storage.get_or_insert_with(tv, ThreadState::default);
+   }
+
+   fn on_flow_init(storage: ThreadStorage<ThreadState>, tv: &mut ThreadVars) {
+       if let Some(state) = storage.get_mut(tv) {
+           state.flows += 1;
+       }
+   }

examples/plugins/rust/src/mod.rs

@@ -3,21 +3,37 @@ use std::ptr::null_mut;
 use suricata_ffi::eve::{self, SCJsonBuilder};
 use suricata_ffi::flow;
 use suricata_ffi::jsonbuilder::JsonBuilder;
-use suricata_ffi::thread::{self, ThreadVars};
-use suricata_ffi::{SCLogError, SCLogNotice};
+use suricata_ffi::thread::{self, ThreadStorage, ThreadVars};
+use suricata_ffi::{SCLogError, SCLogNotice, SCLogWarning};
 use suricata_sys::sys::{self, Flow, Packet, SCEveRegisterCallback, SCPlugin};
 
+/// Per-thread state stored in Suricata thread storage.
+#[derive(Default)]
+struct ThreadState {
+    flows: u64,
+}
+
 unsafe extern "C" fn init() {
     suricata_ffi::plugin::init();
     SCLogNotice!("Initializing rust example plugin");
 
+    // Register per-thread storage once, then hand the (copyable) handle to the
+    // callbacks that use it.
+    let thread_storage = match ThreadStorage::<ThreadState>::register("rust-example-thread") {
+        Ok(storage) => storage,
+        Err(err) => {
+            SCLogError!("Failed to register rust example thread storage: {}", err);
+            return;
+        }
+    };
+
     if let Err(err) = register_eve_callbacks() {
         SCLogError!("Failed to register rust example EVE callbacks: {}", err);
     }
-    if let Err(err) = register_flow_callbacks() {
+    if let Err(err) = register_flow_callbacks(thread_storage) {
         SCLogError!("Failed to register rust example flow callbacks: {}", err);
     }
-    if let Err(err) = register_thread_callbacks() {
+    if let Err(err) = register_thread_callbacks(thread_storage) {
         SCLogError!("Failed to register rust example thread callbacks: {}", err);
     }
 }
@@ -29,15 +45,15 @@ fn register_eve_callbacks() -> Result<(), &'static str> {
     eve::register_callback(log_eve_wrapped)
 }
 
-fn register_flow_callbacks() -> Result<(), &'static str> {
-    flow::register_init_callback(log_flow_init)?;
+fn register_flow_callbacks(storage: ThreadStorage<ThreadState>) -> Result<(), &'static str> {
+    flow::register_init_callback(move |tv, f, p| log_flow_init(storage, tv, f, p))?;
     flow::register_update_callback(log_flow_update)?;
     flow::register_finish_callback(log_flow_finish)?;
     Ok(())
 }
 
-pub fn register_thread_callbacks() -> Result<(), &'static str> {
-    thread::register_init_callback(on_thread_init)
+fn register_thread_callbacks(storage: ThreadStorage<ThreadState>) -> Result<(), &'static str> {
+    thread::register_init_callback(move |tv| on_thread_init(storage, tv))
 }
 
 unsafe extern "C" fn log_eve_raw(
@@ -66,15 +82,39 @@ fn log_eve_wrapped(
     Ok(())
 }
 
-fn on_thread_init(tv: &mut ThreadVars) {
+fn on_thread_init(storage: ThreadStorage<ThreadState>, tv: &mut ThreadVars) {
+    // Initialize the per-thread storage for this thread.
+    if let Err(err) = storage.get_or_insert_with(tv, ThreadState::default) {
+        SCLogError!("failed to initialize rust example thread storage: {}", err);
+    }
     SCLogNotice!(
         "rust example thread init callback: thread={:p}",
         tv.as_ptr()
     );
 }
 
-fn log_flow_init(_tv: &mut ThreadVars, _f: *mut Flow, _p: *const Packet) {
-    SCLogNotice!("rust example flow init callback: flow={:p}", _f);
+fn log_flow_init(
+    storage: ThreadStorage<ThreadState>,
+    tv: &mut ThreadVars,
+    f: *mut Flow,
+    _p: *const Packet,
+) {
+    // Count flows seen by this thread using the per-thread storage.
+    let flows = match storage.get_mut(tv) {
+        Some(state) => {
+            state.flows += 1;
+            state.flows
+        }
+        None => {
+            SCLogWarning!("rust example thread storage was not initialized");
+            0
+        }
+    };
+    SCLogNotice!(
+        "rust example flow init callback: flow={:p}, thread_flows={}",
+        f,
+        flows
+    );
 }
 
 fn log_flow_update(_tv: &mut ThreadVars, _f: *mut Flow, _p: *mut Packet) {

rust/ffi/src/thread.rs

@@ -15,10 +15,14 @@
  * 02110-1301, USA.
  */
 
+use std::ffi::CString;
 use std::marker::PhantomData;
 use std::os::raw::c_void;
 
-use suricata_sys::sys::{self, SCThreadRegisterInitCallback};
+use suricata_sys::sys::{
+    self, SCThreadGetStorageById, SCThreadRegisterInitCallback, SCThreadSetStorageById,
+    SCThreadStorageId, SCThreadStorageRegister,
+};
 
 /// A safe wrapper around a Suricata `sys::ThreadVars` pointer.
 ///
@@ -45,6 +49,129 @@ impl<'a> ThreadVars<'a> {
     pub fn as_ptr(&self) -> *const sys::ThreadVars {
         self.tv
     }
+
+    /// Return the underlying raw `ThreadVars` pointer for mutable access.
+    ///
+    /// Requires `&mut self` so that mutable use of the underlying
+    /// `ThreadVars` (such as setting thread storage) is gated by an exclusive
+    /// borrow of the wrapper.
+    fn as_mut_ptr(&mut self) -> *mut sys::ThreadVars {
+        self.tv
+    }
+}
+
+/// A typed handle to a per-thread storage slot.
+///
+/// `ThreadStorage<T>` wraps the `SCThreadStorageId` returned when registering
+/// thread storage with Suricata. Values are stored as a `Box<T>` owned by
+/// Suricata's thread storage and are dropped automatically when the thread's
+/// storage is freed.
+///
+/// The handle only holds the storage id, so it is `Copy` and `Send`/`Sync`
+/// regardless of `T`, and can be passed by value into the callbacks that need
+/// it.
+pub struct ThreadStorage<T> {
+    id: SCThreadStorageId,
+    _marker: PhantomData<fn() -> T>,
+}
+
+// Manual `Copy`/`Clone` impls so the handle is copyable regardless of whether
+// `T` is; it only holds the storage id.
+impl<T> Clone for ThreadStorage<T> {
+    fn clone(&self) -> Self {
+        *self
+    }
+}
+
+impl<T> Copy for ThreadStorage<T> {}
+
+impl<T: Send + 'static> ThreadStorage<T> {
+    /// Register a new thread storage slot for values of type `T`.
+    ///
+    /// `name` must be unique among registered thread storage. Registration has
+    /// to happen during initialization, before Suricata finalizes storage
+    /// registration (`SCStorageFinalize`).
+    ///
+    /// Returns an error if `name` contains an interior nul byte or if Suricata
+    /// rejects the registration.
+    pub fn register(name: &str) -> Result<Self, &'static str> {
+        let name = CString::new(name).map_err(|_| "thread storage name contains a nul byte")?;
+        let id = unsafe { SCThreadStorageRegister(name.as_ptr(), Some(Self::free)) };
+        if id.id < 0 {
+            return Err("Failed to register thread storage");
+        }
+
+        // Suricata keeps the storage name pointer in its storage mapping for
+        // the lifetime of the process, so the CString is intentionally leaked.
+        std::mem::forget(name);
+
+        Ok(Self {
+            id,
+            _marker: PhantomData,
+        })
+    }
+
+    /// Return a reference to the value stored for `tv`, if any.
+    pub fn get<'t>(&self, tv: &'t ThreadVars<'_>) -> Option<&'t T> {
+        let ptr = unsafe { SCThreadGetStorageById(tv.as_ptr(), self.id) };
+        if ptr.is_null() {
+            None
+        } else {
+            Some(unsafe { &*(ptr as *const T) })
+        }
+    }
+
+    /// Return a mutable reference to the value stored for `tv`, if any.
+    ///
+    /// Takes `&mut ThreadVars` so the returned `&mut T` is the only live
+    /// reference to the stored value for the duration of the borrow.
+    pub fn get_mut<'t>(&self, tv: &'t mut ThreadVars<'_>) -> Option<&'t mut T> {
+        let ptr = unsafe { SCThreadGetStorageById(tv.as_ptr(), self.id) };
+        if ptr.is_null() {
+            None
+        } else {
+            Some(unsafe { &mut *(ptr as *mut T) })
+        }
+    }
+
+    /// Return a mutable reference to the value stored for `tv`, inserting the
+    /// value produced by `init` if none is present yet.
+    ///
+    /// Takes `&mut ThreadVars` so the returned `&mut T` is the only live
+    /// reference to the stored value for the duration of the borrow.
+    pub fn get_or_insert_with<'t>(
+        &self, tv: &'t mut ThreadVars<'_>, init: impl FnOnce() -> T,
+    ) -> Result<&'t mut T, &'static str> {
+        let ptr = unsafe { SCThreadGetStorageById(tv.as_ptr(), self.id) };
+        if !ptr.is_null() {
+            return Ok(unsafe { &mut *(ptr as *mut T) });
+        }
+
+        // `SCThreadSetStorageById` overwrites the slot without freeing any
+        // previous value; we only reach here when the slot is empty.
+        let ptr = Box::into_raw(Box::new(init()));
+        let rc = unsafe { SCThreadSetStorageById(tv.as_mut_ptr(), self.id, ptr.cast()) };
+        if rc != 0 {
+            unsafe {
+                drop(Box::from_raw(ptr));
+            }
+            return Err("Failed to set thread storage");
+        }
+
+        Ok(unsafe { &mut *ptr })
+    }
+
+    /// Free callback registered with Suricata thread storage that drops the
+    /// `Box<T>` backing a stored value.
+    unsafe extern "C" fn free(ptr: *mut c_void) {
+        if !ptr.is_null() {
+            // The drop runs across an FFI boundary, so guard against unwinding
+            // into C if `T`'s `Drop` panics.
+            let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                drop(Box::from_raw(ptr as *mut T));
+            }));
+        }
+    }
 }
 
 /// Register a thread initialization callback.