[#139] events with bitset

Cargo.toml

@@ -20,7 +20,8 @@ members = [
 
     "examples",
 
-    "benchmarks/publish-subscribe"
+    "benchmarks/publish-subscribe",
+    "benchmarks/event"
 ]
 
 [workspace.package]

benchmarks/README.md

@@ -2,19 +2,38 @@
 
 ## Publish-Subscribe
 
-The benchmark quantifies the latency between a publisher sending a message and
-a subscriber receiving it. In the setup, a bidirectional connection is
+The benchmark quantifies the latency between a `Publisher` sending a message and
+a `Subscriber` receiving it. In the setup, a bidirectional connection is
 established from process `a` to `b` (service name `a2b`) and back
-(service name `b2a`). Subscribers employ multithreaded busy waiting and promptly
+(service name `b2a`). `Subscriber`s employ multithreaded busy waiting and promptly
 respond upon message reception. This process repeats `n` times, and the average
 latency is subsequently computed.
 
 ```sh
-cargo run --release benchmark-publish-subscribe -- --bench-all
+cargo run --bin benchmark-publish-subscribe --release -- --bench-all
 ```
 
 For more benchmark configuration details, see
 
 ```sh
-cargo run --release benchmark-publish-subscribe -- --help
+cargo run --bin benchmark-publish-subscribe --release -- --help
+```
+
+## Event
+
+The event quantifies the latency between a `Notifier` sending a notification and
+a `Listener` waking up from and responding to it. In the setup, a bidirectional connection is
+established from process `a` to `b` (service name `a2b`) and back
+(service name `b2a`). The `Listener` employs a blocking wait and wakes up on signal
+reception to promptly respond with a return signal notification. This process repeats `n`
+times, and the average latency is subsequently computed.
+
+```sh
+cargo run --bin benchmark-event --release -- -i 10000
+```
+
+For more benchmark configuration details, see
+
+```sh
+cargo run --bin benchmark-event --release -- --help
 ```

benchmarks/event/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "benchmark-event"
+description = "iceoryx2: [internal] benchmark for the iceoryx2 event services"
+categories = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+keywords = { workspace = true }
+license = { workspace = true }
+repository = { workspace = true }
+rust-version = { workspace = true }
+version = { workspace = true }
+
+[dependencies]
+iceoryx2 = { workspace = true }
+iceoryx2-bb-log = { workspace = true }
+
+clap = { workspace = true }

benchmarks/event/src/main.rs

@@ -0,0 +1,102 @@
+// Copyright (c) 2024 Contributors to the Eclipse Foundation
+//
+// See the NOTICE file(s) distributed with this work for additional
+// information regarding copyright ownership.
+//
+// This program and the accompanying materials are made available under the
+// terms of the Apache Software License 2.0 which is available at
+// https://www.apache.org/licenses/LICENSE-2.0, or the MIT license
+// which is available at https://opensource.org/licenses/MIT.
+//
+// SPDX-License-Identifier: Apache-2.0 OR MIT
+
+use std::{sync::Barrier, time::Instant};
+
+use clap::Parser;
+use iceoryx2::prelude::*;
+use iceoryx2_bb_log::set_log_level;
+
+fn perform_benchmark<T: Service>(args: &Args) {
+    let service_name_a2b = ServiceName::new("a2b").unwrap();
+    let service_name_b2a = ServiceName::new("b2a").unwrap();
+
+    let service_a2b = T::new(&service_name_a2b)
+        .event()
+        .max_notifiers(1)
+        .max_listeners(1)
+        .create()
+        .unwrap();
+
+    let service_b2a = T::new(&service_name_b2a)
+        .event()
+        .max_notifiers(1)
+        .max_listeners(1)
+        .create()
+        .unwrap();
+
+    let barrier = Barrier::new(3);
+
+    std::thread::scope(|s| {
+        let t1 = s.spawn(|| {
+            let notifier_a2b = service_a2b.notifier().create().unwrap();
+            let mut listener_b2a = service_b2a.listener().create().unwrap();
+
+            barrier.wait();
+            notifier_a2b.notify().expect("failed to notify");
+
+            for _ in 0..args.iterations {
+                while listener_b2a.blocking_wait().unwrap().is_empty() {}
+                notifier_a2b.notify().expect("failed to notify");
+            }
+        });
+
+        let t2 = s.spawn(|| {
+            let notifier_b2a = service_b2a.notifier().create().unwrap();
+            let mut listener_a2b = service_a2b.listener().create().unwrap();
+
+            barrier.wait();
+            for _ in 0..args.iterations {
+                while listener_a2b.blocking_wait().unwrap().is_empty() {}
+                notifier_b2a.notify().expect("failed to notify");
+            }
+        });
+
+        std::thread::sleep(std::time::Duration::from_millis(100));
+        let start = Instant::now();
+        barrier.wait();
+
+        t1.join().expect("thread failure");
+        t2.join().expect("thread failure");
+
+        let stop = start.elapsed();
+        println!(
+            "{} ::: Iterations: {}, Time: {}, Latency: {} ns",
+            std::any::type_name::<T>(),
+            args.iterations,
+            stop.as_secs_f64(),
+            stop.as_nanos() / (args.iterations as u128 * 2)
+        );
+    });
+}
+
+const ITERATIONS: usize = 1000000;
+const CAPACITY: usize = 128;
+
+#[derive(Parser, Debug)]
+#[clap(version, about, long_about = None)]
+struct Args {
+    /// Number of iterations the A --> B --> A communication is repeated
+    #[clap(short, long, default_value_t = ITERATIONS)]
+    iterations: usize,
+    /// Capacity of the bitset
+    #[clap(short, long, default_value_t = CAPACITY)]
+    capacity: usize,
+}
+
+fn main() {
+    let args = Args::parse();
+    set_log_level(iceoryx2_bb_log::LogLevel::Error);
+
+    perform_benchmark::<zero_copy::Service>(&args);
+    perform_benchmark::<process_local::Service>(&args);
+}

iceoryx2-bb/lock-free/src/mpmc/bit_set.rs

@@ -31,15 +31,15 @@
 //!  bitset.set(5);
 //!
 //!  // resets the bitset and calls the callback for every bit that was set
-//!  bitset.reset(|id| {
+//!  bitset.reset_all(|id| {
 //!     println!("bit {} was set", id );
 //!  });
 //!  ```
 
 use std::{
     alloc::Layout,
     fmt::Debug,
-    sync::atomic::{AtomicBool, AtomicU8, Ordering},
+    sync::atomic::{AtomicBool, AtomicU8, AtomicUsize, Ordering},
 };
 
 use iceoryx2_bb_elementary::{
@@ -59,7 +59,6 @@ pub type RelocatableBitSet = details::BitSet<RelocatablePointer<details::BitsetE
 
 #[doc(hidden)]
 pub mod details {
-    use std::sync::atomic::AtomicUsize;
 
     use super::*;
 
@@ -87,7 +86,6 @@ pub mod details {
         capacity: usize,
         array_capacity: usize,
         reset_position: AtomicUsize,
-        active_bits: AtomicUsize,
         is_memory_initialized: AtomicBool,
     }
 
@@ -115,7 +113,6 @@ pub mod details {
                 array_capacity,
                 is_memory_initialized: AtomicBool::new(true),
                 reset_position: AtomicUsize::new(0),
-                active_bits: AtomicUsize::new(0),
             }
         }
     }
@@ -128,7 +125,6 @@ pub mod details {
                 array_capacity: Self::array_capacity(capacity),
                 is_memory_initialized: AtomicBool::new(false),
                 reset_position: AtomicUsize::new(0),
-                active_bits: AtomicUsize::new(0),
             }
         }
 
@@ -176,7 +172,6 @@ pub mod details {
                 array_capacity: Self::array_capacity(capacity),
                 is_memory_initialized: AtomicBool::new(true),
                 reset_position: AtomicUsize::new(0),
-                active_bits: AtomicUsize::new(0),
             }
         }
     }
@@ -224,7 +219,6 @@ pub mod details {
                     Ordering::Relaxed,
                 ) {
                     Ok(_) => {
-                        self.active_bits.fetch_add(1, Ordering::Relaxed);
                         return true;
                     }
                     Err(v) => current = v,
@@ -251,7 +245,6 @@ pub mod details {
                     Ordering::Relaxed,
                 ) {
                     Ok(_) => {
-                        self.active_bits.fetch_sub(1, Ordering::Relaxed);
                         return true;
                     }
                     Err(v) => current = v,
@@ -277,9 +270,6 @@ pub mod details {
         /// [`None`].
         pub fn reset_next(&self) -> Option<usize> {
             self.verify_init("reset_next");
-            if self.active_bits.load(Ordering::Relaxed) == 0 {
-                return None;
-            }
 
             let current_position = self.reset_position.load(Ordering::Relaxed);
             for pos in (current_position..self.capacity).chain(0..current_position) {
@@ -299,18 +289,12 @@ pub mod details {
 
             for i in 0..self.array_capacity {
                 let value = unsafe { (*self.data_ptr.as_ptr().add(i)).swap(0, Ordering::Relaxed) };
-                let mut counter = 0;
                 let main_index = i * BITSET_ELEMENT_BITSIZE;
                 for b in 0..BITSET_ELEMENT_BITSIZE {
                     if value & (1 << b) != 0 {
                         callback(main_index + b);
-                        counter += 1;
                     }
                 }
-
-                if self.active_bits.fetch_sub(counter, Ordering::Relaxed) == counter {
-                    return;
-                }
             }
         }
     }

iceoryx2-bb/posix/src/semaphore.rs

@@ -378,7 +378,7 @@ impl NamedSemaphoreCreationBuilder {
 ///                     .expect("failed to open semaphore");
 ///
 /// loop {
-///     semaphore.wait().expect("failed to wait on semaphore");
+///     semaphore.blocking_wait().expect("failed to wait on semaphore");
 ///     println!("process 1 has triggered me");
 /// }
 /// ```
@@ -712,7 +712,7 @@ impl Drop for UnnamedSemaphoreHandle {
 /// thread::scope(|s| {
 ///     s.spawn(|| {
 ///         loop {
-///             semaphore.wait().expect("failed to wait on semaphore");
+///             semaphore.blocking_wait().expect("failed to wait on semaphore");
 ///             println!("the thread was triggered");
 ///         }
 ///     });

iceoryx2-cal/src/event/common.rs

@@ -14,7 +14,7 @@
 pub mod details {
     use std::{fmt::Debug, marker::PhantomData, time::Duration};
 
-    use iceoryx2_bb_log::fail;
+    use iceoryx2_bb_log::{debug, fail};
     use iceoryx2_bb_memory::bump_allocator::BumpAllocator;
     use iceoryx2_bb_system_types::{file_name::FileName, path::Path};
 
@@ -32,7 +32,7 @@ pub mod details {
         },
     };
 
-    const TRIGGER_ID_DEFAULT_MAX: TriggerId = TriggerId::new(u16::MAX as u64);
+    const TRIGGER_ID_DEFAULT_MAX: TriggerId = TriggerId::new(u16::MAX as _);
 
     #[derive(Debug)]
     pub struct Management<Tracker: IdTracker, WaitMechanism: SignalMechanism> {
@@ -383,6 +383,7 @@ pub mod details {
         fn try_wait(
             &self,
         ) -> Result<Option<crate::event::TriggerId>, crate::event::ListenerWaitError> {
+            // collect all notifications until no more are available
             while unsafe { self.storage.get().signal_mechanism.try_wait()? } {}
             Ok(unsafe { self.storage.get().id_tracker.acquire() })
         }
@@ -395,11 +396,14 @@ pub mod details {
                 return Ok(Some(id));
             }
 
-            if unsafe { self.storage.get().signal_mechanism.timed_wait(timeout)? } {
-                return Ok(unsafe { self.storage.get().id_tracker.acquire() });
-            }
-
-            Ok(None)
+            Ok(unsafe {
+                self.storage
+                    .get()
+                    .signal_mechanism
+                    .timed_wait(timeout)?
+                    .then_some(self.storage.get().id_tracker.acquire())
+                    .flatten()
+            })
         }
 
         fn blocking_wait(
@@ -459,10 +463,13 @@ pub mod details {
             mgmt: &mut Management<Tracker, WaitMechanism>,
             allocator: &mut BumpAllocator,
         ) -> bool {
+            let origin = "init()";
             if unsafe { mgmt.id_tracker.init(allocator).is_err() } {
+                debug!(from origin, "Unable to initialize IdTracker.");
                 return false;
             }
             if unsafe { mgmt.signal_mechanism.init().is_err() } {
+                debug!(from origin, "Unable to initialize SignalMechanism.");
                 return false;
             }
 
@@ -489,7 +496,7 @@ pub mod details {
             ListenerCreateError,
         > {
             let msg = "Failed to create Listener";
-            let id_tracker_capacity = self.trigger_id_max.as_u64() as usize;
+            let id_tracker_capacity = self.trigger_id_max.as_u64() as usize + 1;
 
             match Storage::Builder::new(&self.name)
                 .config(&self.config.convert())

iceoryx2-cal/src/event/id_tracker/bit_set.rs

@@ -18,11 +18,11 @@ use crate::event::{NotifierNotifyError, TriggerId};
 
 impl IdTracker for RelocatableBitSet {
     fn trigger_id_max(&self) -> TriggerId {
-        TriggerId::new(self.capacity() as u64)
+        TriggerId::new(self.capacity() as u64 - 1)
     }
 
     unsafe fn add(&self, id: TriggerId) -> Result<(), NotifierNotifyError> {
-        if self.trigger_id_max() <= id {
+        if self.trigger_id_max() < id {
             fail!(from self, with NotifierNotifyError::TriggerIdOutOfBounds,
                 "Unable to set bit {:?} since it is out of bounds (max = {:?}).",
                 id, self.trigger_id_max());