Events: use channels instead of polling

pkcs11/Cargo.toml

@@ -39,6 +39,7 @@ config_file = { path = "config_file" }
 ureq = { version = "=3.3.0", default-features  = false }
 rustls-pki-types = "1.11.0"
 socket2 = { version = "0.6", features = ["all"] }
+crossbeam = "0.8.4"
 
 [dev-dependencies]
 once_cell = "1.19.0"

pkcs11/src/api/mod.rs

@@ -14,10 +14,7 @@ use std::{ptr::addr_of_mut, sync::Arc};
 
 use crate::config::device::{start_background_timer, stop_background_timer};
 use crate::{
-    backend::{
-        events::{fetch_slots_state, EventsManager},
-        Pkcs11Error,
-    },
+    backend::{events::fetch_slots_state, Pkcs11Error},
     data::{self, DEVICE, EVENTS_MANAGER, THREADS_ALLOWED, TOKENS_STATE},
     defs,
     utils::padded_str,
@@ -123,8 +120,8 @@ fn initialize(init_args_ptr: CK_VOID_PTR) -> Result<(), Pkcs11Error> {
     }
 
     // Initialize the events manager
-    *EVENTS_MANAGER.write().unwrap() = EventsManager::new();
-    *TOKENS_STATE.lock().unwrap() = std::collections::HashMap::new();
+    EVENTS_MANAGER.reset();
+    TOKENS_STATE.lock().unwrap().clear();
 
     fetch_slots_state()
 }
@@ -140,7 +137,7 @@ fn finalize(reserved_ptr: CK_VOID_PTR) -> Result<(), Pkcs11Error> {
     }
     DEVICE.store(None);
     stop_background_timer();
-    EVENTS_MANAGER.write().unwrap().finalized = true;
+    EVENTS_MANAGER.reset();
     Ok(())
 }
 

pkcs11/src/api/token.rs

@@ -1,3 +1,6 @@
+use std::time::Duration;
+
+use crossbeam::channel::TryRecvError;
 use cryptoki_sys::{
     CKF_RNG, CKF_TOKEN_INITIALIZED, CKF_USER_PIN_INITIALIZED, CK_SLOT_ID, CK_SLOT_INFO,
     CK_TOKEN_INFO, CK_ULONG,
@@ -385,34 +388,27 @@ fn wait_for_slot_event(
 
     fetch_slots_state()?;
 
+    let recv = &EVENTS_MANAGER.receiver.load();
     loop {
         // check if there is an event in the queue
+        let slot = if flags & cryptoki_sys::CKF_DONT_BLOCK == cryptoki_sys::CKF_DONT_BLOCK {
+            match recv.try_recv() {
+                Ok(s) => Some(s),
+                Err(TryRecvError::Empty) => return Err(Pkcs11Error::NoEvent),
+                Err(TryRecvError::Disconnected) => return Err(Pkcs11Error::CryptokiNotInitialized),
+            }
+        } else {
+            recv.recv_timeout(Duration::from_secs(1)).ok()
+        };
 
-        let slot = EVENTS_MANAGER.write().unwrap().events.pop();
         if let Some(slot) = slot {
             unsafe {
                 std::ptr::write(slot_ptr, slot);
             }
             return Ok(());
         }
 
-        // if the dont block flag is set, return no event
-        if flags & cryptoki_sys::CKF_DONT_BLOCK == 1 {
-            return Err(Pkcs11Error::NoEvent);
-        } else {
-            // Otherwise, wait for an event
-
-            // If C_Finalize() has been called, return an error
-            if EVENTS_MANAGER.read().unwrap().finalized {
-                return Err(Pkcs11Error::CryptokiNotInitialized);
-            }
-
-            // sleep for 1 second
-            std::thread::sleep(std::time::Duration::from_secs(1));
-
-            // fetch the slots state so we get the latest events in the next iteration
-            fetch_slots_state()?;
-        }
+        fetch_slots_state()?;
     }
 }
 
@@ -422,10 +418,7 @@ mod tests {
 
     use crate::{
         api::C_Finalize,
-        backend::{
-            events::{update_slot_state, EventsManager},
-            slot::init_for_tests,
-        },
+        backend::{events::update_slot_state, slot::init_for_tests},
         data::{SESSION_MANAGER, TOKENS_STATE},
     };
 
@@ -437,8 +430,8 @@ mod tests {
     #[ignore]
     fn test_wait_for_slot_event_no_event() {
         let _guard = init_for_tests();
-        *EVENTS_MANAGER.write().unwrap() = EventsManager::new();
-        *TOKENS_STATE.lock().unwrap() = std::collections::HashMap::new();
+        EVENTS_MANAGER.reset();
+        TOKENS_STATE.lock().unwrap().clear();
 
         let mut slot = 0;
         let result = C_WaitForSlotEvent(CKF_DONT_BLOCK, &mut slot, std::ptr::null_mut());
@@ -451,14 +444,12 @@ mod tests {
     #[ignore]
     fn test_wait_for_slot_event_one_event() {
         let _guard = init_for_tests();
-        *EVENTS_MANAGER.write().unwrap() = EventsManager::new();
-        *TOKENS_STATE.lock().unwrap() = std::collections::HashMap::new();
+        EVENTS_MANAGER.reset();
+        TOKENS_STATE.lock().unwrap().clear();
 
         update_slot_state(0, false);
         update_slot_state(0, true);
 
-        println!("Events: {:?}", EVENTS_MANAGER.read().unwrap().events);
-
         let mut slot = 15;
         let result = C_WaitForSlotEvent(CKF_DONT_BLOCK, &mut slot, std::ptr::null_mut());
         assert_eq!(result, cryptoki_sys::CKR_OK);
@@ -471,8 +462,8 @@ mod tests {
     #[ignore]
     fn test_wait_for_slot_event_blocking_one_event() {
         let _guard = init_for_tests();
-        *EVENTS_MANAGER.write().unwrap() = EventsManager::new();
-        *TOKENS_STATE.lock().unwrap() = std::collections::HashMap::new();
+        EVENTS_MANAGER.reset();
+        TOKENS_STATE.lock().unwrap().clear();
 
         // update the slot state in a separate thread
 
@@ -495,8 +486,8 @@ mod tests {
     #[ignore]
     fn test_wait_for_slot_event_blocking_finalize() {
         let _guard = init_for_tests();
-        *EVENTS_MANAGER.write().unwrap() = EventsManager::new();
-        *TOKENS_STATE.lock().unwrap() = std::collections::HashMap::new();
+        EVENTS_MANAGER.reset();
+        TOKENS_STATE.lock().unwrap().clear();
 
         // update the slot state in a separate thread
 

pkcs11/src/backend/events.rs

@@ -1,24 +1,34 @@
+use std::sync::Arc;
+
+use arc_swap::ArcSwap;
 use cryptoki_sys::CK_SLOT_ID;
 use nethsm_sdk_rs::{apis::default_api, models::SystemState};
 
+use crossbeam::channel::{bounded, Receiver, Sender, TrySendError};
+
 use crate::data::{self, EVENTS_MANAGER, TOKENS_STATE};
 
 use super::{login::LoginCtx, Pkcs11Error};
 
 pub struct EventsManager {
-    pub events: Vec<CK_SLOT_ID>, // list of slots that changed
-
-    // Used when CKF_DONT_BLOCK is clear and C_Finalize is called, then every blocking call to C_WaitForSlotEvent should return CKR_CRYPTOKI_NOT_INITIALIZED
-    pub finalized: bool,
+    pub sender: ArcSwap<Sender<CK_SLOT_ID>>,
+    pub receiver: ArcSwap<Receiver<CK_SLOT_ID>>,
 }
 
 impl EventsManager {
-    pub const fn new() -> Self {
+    pub fn new() -> Self {
+        let (tx, rx) = bounded(128);
         EventsManager {
-            events: Vec::new(),
-            finalized: false,
+            sender: ArcSwap::new(Arc::new(tx)),
+            receiver: ArcSwap::new(Arc::new(rx)),
         }
     }
+
+    pub fn reset(&self) {
+        let (tx, rx) = bounded(128);
+        self.sender.store(Arc::new(tx));
+        self.receiver.store(Arc::new(rx));
+    }
 }
 
 pub fn update_slot_state(slot_id: CK_SLOT_ID, present: bool) {
@@ -28,7 +38,17 @@ pub fn update_slot_state(slot_id: CK_SLOT_ID, present: bool) {
             return;
         } else {
             // new event
-            EVENTS_MANAGER.write().unwrap().events.push(slot_id);
+            loop {
+                match EVENTS_MANAGER.sender.load().try_send(slot_id) {
+                    Ok(_) => break,
+                    Err(TrySendError::Full(_)) => {
+                        log::warn!("Dropping slot event to avoid filling up memory")
+                    }
+                    Err(TrySendError::Disconnected(_)) => {
+                        continue;
+                    }
+                };
+            }
         }
     }
     tokens_state.insert(slot_id, present);

pkcs11/src/data.rs

@@ -1,5 +1,6 @@
 use std::collections::HashMap;
-use std::sync::{atomic::AtomicBool, Arc, Mutex, MutexGuard, RwLock};
+use std::sync::LazyLock;
+use std::sync::{atomic::AtomicBool, Arc, Mutex, MutexGuard};
 
 use crate::backend::{events::EventsManager, Pkcs11Error};
 
@@ -28,7 +29,7 @@ lazy_static! {
 }
 
 // Storage of events
-pub static EVENTS_MANAGER: RwLock<EventsManager> = RwLock::new(EventsManager::new());
+pub static EVENTS_MANAGER: LazyLock<EventsManager> = LazyLock::new(EventsManager::new);
 
 // If the calling application allows threads to be used
 pub static THREADS_ALLOWED: AtomicBool = AtomicBool::new(true);