[#390] Use temporarily uds in process local; add waitset tests

elfenpiff · elfenpiff · commit 0c7ca92b3d86 · 2024-10-04T16:25:56.000+02:00
diff --git a/iceoryx2/src/node/mod.rs b/iceoryx2/src/node/mod.rs
@@ -121,7 +121,6 @@ pub mod node_name;
 pub mod testing;
 
 use crate::node::node_name::NodeName;
-use crate::prelude::WaitEvent;
 use crate::service::builder::{Builder, OpenDynamicStorageFailure};
 use crate::service::config_scheme::{
     node_details_path, node_monitoring_config, service_tag_config,
@@ -196,6 +195,23 @@ impl std::fmt::Display for NodeCreationFailure {
 
 impl std::error::Error for NodeCreationFailure {}
 
+/// The failures that can occur when a list of [`NodeState`]s is created with [`Node::list()`].
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+pub enum NodeWaitFailure {
+    /// The process received an interrupt signal while acquiring the list of all [`Node`]s.
+    Interrupt,
+    /// A termination signal `SIGTERM` was received.
+    TerminationRequest,
+}
+
+impl std::fmt::Display for NodeWaitFailure {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        std::write!(f, "NodeWaitFailure::{:?}", self)
+    }
+}
+
+impl std::error::Error for NodeWaitFailure {}
+
 /// The failures that can occur when a list of [`NodeState`]s is created with [`Node::list()`].
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub enum NodeListFailure {
@@ -791,20 +807,26 @@ impl<Service: service::Service> Node<Service> {
 
     /// Waits until an event was received. It returns
     /// [`WaitEvent::Tick`] when the `cycle_time` has passed, otherwise event that occurred.
-    pub fn wait(&self, cycle_time: Duration) -> WaitEvent {
+    pub fn wait(&self, cycle_time: Duration) -> Result<(), NodeWaitFailure> {
+        let msg = "Unable to wait on node";
         if SignalHandler::termination_requested() {
-            return WaitEvent::TerminationRequest;
+            fail!(from self, with NodeWaitFailure::TerminationRequest,
+                "{msg} since a termination request was received.");
         }
 
         match nanosleep(cycle_time) {
             Ok(()) => {
                 if SignalHandler::termination_requested() {
-                    WaitEvent::TerminationRequest
+                    fail!(from self, with NodeWaitFailure::TerminationRequest,
+                        "{msg} since a termination request was received.");
                 } else {
-                    WaitEvent::Tick
+                    Ok(())
                 }
             }
-            Err(NanosleepError::InterruptedBySignal(_)) => WaitEvent::Interrupt,
+            Err(NanosleepError::InterruptedBySignal(_)) => {
+                fail!(from self, with NodeWaitFailure::Interrupt,
+                        "{msg} since a interrupt signal was received.");
+            }
             Err(v) => {
                 fatal_panic!(from self,
                     "Failed to wait with cycle time {:?} in main event look, caused by ({:?}).",
diff --git a/iceoryx2/src/port/waitset.rs b/iceoryx2/src/port/waitset.rs
@@ -64,7 +64,7 @@
 //!     }
 //! };
 //!
-//! while waitset.run(event_handler) != Ok(WaitEvent::TerminationRequest) {}
+//! while waitset.run(event_handler).is_ok() {}
 //!
 //! # Ok(())
 //! # }
@@ -97,7 +97,7 @@
 //!     }
 //! };
 //!
-//! while waitset.run(event_handler) != Ok(WaitEvent::TerminationRequest) {}
+//! while waitset.run(event_handler).is_ok() {}
 //!
 //! # Ok(())
 //! # }
@@ -132,7 +132,7 @@
 //!     }
 //! };
 //!
-//! while waitset.run(event_handler) != Ok(WaitEvent::TerminationRequest) {}
+//! while waitset.run(event_handler).is_ok() {}
 //!
 //! # Ok(())
 //! # }
@@ -172,7 +172,7 @@
 //!             println!("received notification {:?}", event_id);
 //!         }
 //!     }
-//! }) != Ok(WaitEvent::TerminationRequest) {}
+//! }).is_ok() {}
 //!
 //! # Ok(())
 //! # }
@@ -193,20 +193,6 @@ use iceoryx2_bb_posix::{
 use iceoryx2_cal::reactor::*;
 use iceoryx2_pal_concurrency_sync::iox_atomic::IoxAtomicUsize;
 
-/// Defines the type of that triggered [`WaitSet::try_wait()`], [`WaitSet::timed_wait()`] or
-/// [`WaitSet::blocking_wait()`].
-#[derive(Debug, PartialEq, Eq, Copy, Clone)]
-pub enum WaitEvent {
-    /// A termination signal `SIGTERM` was received.
-    TerminationRequest,
-    /// An interrupt signal `SIGINT` was received.
-    Interrupt,
-    /// No event was triggered.
-    Tick,
-    /// One or more event notifications were received.
-    Notification,
-}
-
 /// Defines the failures that can occur when attaching something with [`WaitSet::attach()`].
 #[derive(Debug, PartialEq, Eq, Copy, Clone)]
 pub enum WaitSetAttachmentError {
@@ -226,27 +212,28 @@ impl std::fmt::Display for WaitSetAttachmentError {
 
 impl std::error::Error for WaitSetAttachmentError {}
 
-/// Defines the failures that can occur when calling
-///  * [`WaitSet::try_wait()`]
-///  * [`WaitSet::timed_wait()`]
-///  * [`WaitSet::blocking_wait()`]
+/// Defines the failures that can occur when calling [`WaitSet::run()`].
 #[derive(Debug, PartialEq, Eq, Copy, Clone)]
-pub enum WaitSetWaitError {
+pub enum WaitSetRunError {
     /// The process has not sufficient permissions to wait on the attachments.
     InsufficientPermissions,
     /// An internal error has occurred.
     InternalError,
     /// Waiting on an empty [`WaitSet`] would lead to a deadlock therefore it causes an error.
     NoAttachments,
+    /// A termination signal `SIGTERM` was received.
+    TerminationRequest,
+    /// An interrupt signal `SIGINT` was received.
+    Interrupt,
 }
 
-impl std::fmt::Display for WaitSetWaitError {
+impl std::fmt::Display for WaitSetRunError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        std::write!(f, "WaitSetWaitError::{:?}", self)
+        std::write!(f, "WaitSetRunError::{:?}", self)
     }
 }
 
-impl std::error::Error for WaitSetWaitError {}
+impl std::error::Error for WaitSetRunError {}
 
 /// Defines the failures that can occur when calling [`WaitSetBuilder::create()`].
 #[derive(Debug, PartialEq, Eq, Copy, Clone)]
@@ -327,12 +314,20 @@ impl<Service: crate::service::Service> AttachmentId<Service> {
         }
     }
 
-    /// Returns true if an event was emitted from the attachment corresponding to [`Guard`].
+    /// Returns true if an event was emitted from a notification or deadline attachment
+    /// corresponding to [`Guard`].
     pub fn event_from(&self, other: &Guard<Service>) -> bool {
-        if let AttachmentIdType::Deadline(..) = self.attachment_type {
-            false
+        let other_attachment = other.to_attachment_id();
+        if let AttachmentIdType::Deadline(other_waitset, other_reactor_idx, _) =
+            other_attachment.attachment_type
+        {
+            if let AttachmentIdType::Notification(waitset, reactor_idx) = self.attachment_type {
+                waitset == other_waitset && reactor_idx == other_reactor_idx
+            } else {
+                false
+            }
         } else {
-            self.attachment_type == other.to_attachment_id().attachment_type
+            self.attachment_type == other_attachment.attachment_type
         }
     }
 
@@ -481,19 +476,15 @@ impl<Service: crate::service::Service> WaitSet<Service> {
             .remove(&deadline_queue_idx);
     }
 
-    fn contains_deadlines(&self) -> bool {
-        !self.attachment_to_deadline.borrow().is_empty()
-    }
-
     fn reset_deadline(
         &self,
         reactor_idx: i32,
-    ) -> Result<Option<DeadlineQueueIndex>, WaitSetWaitError> {
+    ) -> Result<Option<DeadlineQueueIndex>, WaitSetRunError> {
         let msg = "Unable to reset deadline";
         if let Some(deadline_queue_idx) = self.attachment_to_deadline.borrow().get(&reactor_idx) {
             fail!(from self,
                   when self.deadline_queue.reset(*deadline_queue_idx),
-                  with WaitSetWaitError::InternalError,
+                  with WaitSetRunError::InternalError,
                   "{msg} since the deadline_queue guard could not be reset for the attachment {reactor_idx}. Continuing operations will lead to invalid deadline failures.");
             Ok(Some(*deadline_queue_idx))
         } else {
@@ -505,7 +496,7 @@ impl<Service: crate::service::Service> WaitSet<Service> {
         &self,
         fn_call: &mut F,
         error_msg: &str,
-    ) -> Result<(), WaitSetWaitError> {
+    ) -> Result<(), WaitSetRunError> {
         let deadline_to_attachment = self.deadline_to_attachment.borrow();
         let call = |idx: DeadlineQueueIndex| {
             if let Some(reactor_idx) = deadline_to_attachment.get(&idx) {
@@ -515,12 +506,10 @@ impl<Service: crate::service::Service> WaitSet<Service> {
             }
         };
 
-        if self.contains_deadlines() {
-            fail!(from self,
+        fail!(from self,
                   when self.deadline_queue.missed_deadlines(call),
-                  with WaitSetWaitError::InternalError,
+                  with WaitSetRunError::InternalError,
                   "{error_msg} since the missed deadlines could not be acquired.");
-        }
 
         Ok(())
     }
@@ -530,32 +519,22 @@ impl<Service: crate::service::Service> WaitSet<Service> {
         triggered_file_descriptors: &Vec<i32>,
         fn_call: &mut F,
         error_msg: &str,
-    ) -> Result<(), WaitSetWaitError> {
+    ) -> Result<(), WaitSetRunError> {
         // we need to reset the deadlines first, otherwise a long fn_call may extend the
         // deadline unintentionally
-        if self.contains_deadlines() {
-            let mut fd_and_deadline_queue_idx = Vec::new();
-            fd_and_deadline_queue_idx.reserve(triggered_file_descriptors.len());
+        let mut fd_and_deadline_queue_idx = Vec::new();
+        fd_and_deadline_queue_idx.reserve(triggered_file_descriptors.len());
 
-            for fd in triggered_file_descriptors {
-                fd_and_deadline_queue_idx.push((fd, self.reset_deadline(*fd)?));
-            }
+        for fd in triggered_file_descriptors {
+            fd_and_deadline_queue_idx.push((fd, self.reset_deadline(*fd)?));
+        }
 
-            // must be called after the deadlines have been reset, in the case that the
-            // event has been received shortly before the deadline ended.
-            self.handle_deadlines(fn_call, error_msg)?;
+        // must be called after the deadlines have been reset, in the case that the
+        // event has been received shortly before the deadline ended.
+        self.handle_deadlines(fn_call, error_msg)?;
 
-            for (fd, deadline_queue_idx) in fd_and_deadline_queue_idx {
-                if let Some(deadline_queue_idx) = deadline_queue_idx {
-                    fn_call(AttachmentId::deadline(self, *fd, deadline_queue_idx));
-                } else {
-                    fn_call(AttachmentId::notification(self, *fd));
-                }
-            }
-        } else {
-            for fd in triggered_file_descriptors {
-                fn_call(AttachmentId::notification(self, *fd));
-            }
+        for fd in triggered_file_descriptors {
+            fn_call(AttachmentId::notification(self, *fd));
         }
 
         Ok(())
@@ -591,10 +570,15 @@ impl<Service: crate::service::Service> WaitSet<Service> {
         let reactor_guard = self.attach_to_reactor(attachment)?;
         let deadline_queue_guard = self.attach_to_deadline_queue(deadline)?;
 
-        self.attachment_to_deadline.borrow_mut().insert(
-            unsafe { reactor_guard.file_descriptor().native_handle() },
-            deadline_queue_guard.index(),
-        );
+        let reactor_idx = unsafe { reactor_guard.file_descriptor().native_handle() };
+        let deadline_idx = deadline_queue_guard.index();
+
+        self.attachment_to_deadline
+            .borrow_mut()
+            .insert(reactor_idx, deadline_idx);
+        self.deadline_to_attachment
+            .borrow_mut()
+            .insert(deadline_idx, reactor_idx);
         self.attach()?;
 
         Ok(Guard {
@@ -625,21 +609,22 @@ impl<Service: crate::service::Service> WaitSet<Service> {
     pub fn run<F: FnMut(AttachmentId<Service>)>(
         &self,
         mut fn_call: F,
-    ) -> Result<WaitEvent, WaitSetWaitError> {
+    ) -> Result<(), WaitSetRunError> {
         let msg = "Unable to call WaitSet::run()";
 
         if SignalHandler::termination_requested() {
-            return Ok(WaitEvent::TerminationRequest);
+            fail!(from self, with WaitSetRunError::TerminationRequest,
+                "{msg} since a termination request was received.");
         }
 
         if self.is_empty() {
-            fail!(from self, with WaitSetWaitError::NoAttachments,
+            fail!(from self, with WaitSetRunError::NoAttachments,
                 "{msg} since the WaitSet has no attachments, therefore the call would end up in a deadlock.");
         }
 
         let next_timeout = fail!(from self,
                                  when self.deadline_queue.duration_until_next_deadline(),
-                                 with WaitSetWaitError::InternalError,
+                                 with WaitSetRunError::InternalError,
                                  "{msg} since the next timeout could not be acquired.");
 
         let mut triggered_file_descriptors = vec![];
@@ -655,19 +640,22 @@ impl<Service: crate::service::Service> WaitSet<Service> {
         ) {
             Ok(0) => {
                 self.handle_deadlines(&mut fn_call, msg)?;
-                Ok(WaitEvent::Tick)
+                Ok(())
             }
             Ok(_) => {
                 self.handle_all_attachments(&triggered_file_descriptors, &mut fn_call, msg)?;
-                Ok(WaitEvent::Notification)
+                Ok(())
+            }
+            Err(ReactorWaitError::Interrupt) => {
+                fail!(from self, with WaitSetRunError::Interrupt,
+                    "{msg} since an interrupt signal was received.");
             }
-            Err(ReactorWaitError::Interrupt) => Ok(WaitEvent::Interrupt),
             Err(ReactorWaitError::InsufficientPermissions) => {
-                fail!(from self, with WaitSetWaitError::InsufficientPermissions,
+                fail!(from self, with WaitSetRunError::InsufficientPermissions,
                     "{msg} due to insufficient permissions.");
             }
             Err(ReactorWaitError::UnknownError) => {
-                fail!(from self, with WaitSetWaitError::InternalError,
+                fail!(from self, with WaitSetRunError::InternalError,
                     "{msg} due to an internal error.");
             }
         }
diff --git a/iceoryx2/src/prelude.rs b/iceoryx2/src/prelude.rs
@@ -13,7 +13,7 @@
 pub use crate::config::Config;
 pub use crate::node::{node_name::NodeName, Node, NodeBuilder, NodeState};
 pub use crate::port::event_id::EventId;
-pub use crate::port::waitset::{AttachmentId, WaitEvent, WaitSet, WaitSetBuilder};
+pub use crate::port::waitset::{AttachmentId, WaitSet, WaitSetBuilder};
 pub use crate::service::messaging_pattern::MessagingPattern;
 pub use crate::service::{
     attribute::AttributeSet, attribute::AttributeSpecifier, attribute::AttributeVerifier, ipc,
diff --git a/iceoryx2/src/service/local.rs b/iceoryx2/src/service/local.rs
@@ -53,7 +53,8 @@ impl crate::service::Service for Service {
     type ServiceNameHasher = hash::sha1::Sha1;
     type SharedMemory = shared_memory::process_local::Memory<PoolAllocator>;
     type Connection = zero_copy_connection::process_local::Connection;
-    type Event = event::process_local::EventImpl;
+    //type Event = event::process_local::EventImpl;
+    type Event = event::unix_datagram_socket::EventImpl;
     type Monitoring = monitoring::process_local::ProcessLocalMonitoring;
     type Reactor = reactor::posix_select::Reactor;
 }
diff --git a/iceoryx2/tests/waitset_tests.rs b/iceoryx2/tests/waitset_tests.rs
