Merge pull request #365 from elfenpiff/iox2-337-samples-rejected-afte…

…r-publisher-disconnected

[#337] samples rejected after publisher disconnected

config/README.md

@@ -59,3 +59,4 @@ Adjusting `global` settings ensures a non-interfering setup.
  * `defaults.publish-subscribe.publisher-max-loaned-samples` - [int]: Maximum samples a publisher can loan.
  * `defaults.publish-subscribe.enable-safe-overflow` - [`true`|`false`]: Default overflow behavior.
  * `defaults.publish-subscribe.unable-to-deliver-strategy` - [`Block`|`DiscardSample`]: Default strategy for non-overflowing setups when delivery fails.
+ * `defaults.publish-subscribe.subscriber-expired-connection-buffer` - [int]: Expired connection buffer size of the subscriber. Connections to publishers are expired when the publisher disconnected from the service and the connection contains unconsumed samples.

config/iceoryx2.toml

@@ -31,6 +31,7 @@ subscriber-max-borrowed-samples             = 2
 publisher-max-loaned-samples                = 2
 enable-safe-overflow                        = true
 unable-to-deliver-strategy                  = 'Block' # or 'DiscardSample'
+subscriber-expired-connection-buffer        = 128
 
 [defaults.event]
 max-listeners                               = 2

doc/release-notes/iceoryx2-unreleased.md

@@ -71,6 +71,7 @@
  * Mem-leak in `iceoryx2-bb-posix::Directory::contents()` and skip empty file names [#287](https://github.com/eclipse-iceoryx/iceoryx2/issues/287)
  * Log macros do no longer return values [#292](https://github.com/eclipse-iceoryx/iceoryx2/issues/292)
  * Fix cross-compilation issue with `scandir.c` [#318](https://github.com/eclipse-iceoryx/iceoryx2/issues/318)
+ * Fix sample loss when publisher disconnected before subscriber called receive [#337](https://github.com/eclipse-iceoryx/iceoryx2/issues/337)
  * Service-creation-timeout is considered also for the data segment and zero copy connection [#361](https://github.com/eclipse-iceoryx/iceoryx2/issues/361)
 
 ### Refactoring

iceoryx2-bb/container/src/queue.rs

@@ -149,17 +149,29 @@ pub mod details {
             unsafe { self.clear_impl() }
         }
 
-        /// Acquire an element from the queue. If the queue is empty it returns [`None`].
+        /// Returns a reference to the element from the beginning of the queue without removing it.
+        /// If the queue is empty it returns [`None`].
+        pub fn peek(&self) -> Option<&T> {
+            unsafe { self.peek_impl() }
+        }
+
+        /// Returns a mutable reference to the element from the beginning of the queue without removing it.
+        /// If the queue is empty it returns [`None`].
+        pub fn peek_mut(&mut self) -> Option<&mut T> {
+            unsafe { self.peek_mut_impl() }
+        }
+
+        /// Removes the element from the beginning of the queue. If the queue is empty it returns [`None`].
         pub fn pop(&mut self) -> Option<T> {
             unsafe { self.pop_impl() }
         }
 
-        /// Adds an element to the queue. If the queue is full it returns false, otherwise true.
+        /// Adds an element at the end of the queue. If the queue is full it returns false, otherwise true.
         pub fn push(&mut self, value: T) -> bool {
             unsafe { self.push_impl(value) }
         }
 
-        /// Adds an element to the queue. If the queue is full it returns the oldest element,
+        /// Adds an element at the end of the queue. If the queue is full it returns the oldest element,
         /// otherwise [`None`].
         pub fn push_with_overflow(&mut self, value: T) -> Option<T> {
             unsafe { self.push_with_overflow_impl(value) }
@@ -259,7 +271,29 @@ pub mod details {
             self.clear_impl()
         }
 
-        /// Acquire an element from the queue. If the queue is empty it returns [`None`].
+        /// Returns a reference to the element from the beginning of the queue without removing it.
+        /// If the queue is empty it returns [`None`].
+        ///
+        /// # Safety
+        ///
+        ///  * [`Queue::init()`] must have been called once before
+        ///
+        pub fn peek(&self) -> Option<&T> {
+            unsafe { self.peek_impl() }
+        }
+
+        /// Returns a mutable reference to the element from the beginning of the queue without removing it.
+        /// If the queue is empty it returns [`None`].
+        ///
+        /// # Safety
+        ///
+        ///  * [`Queue::init()`] must have been called once before
+        ///
+        pub fn peek_mut(&mut self) -> Option<&mut T> {
+            unsafe { self.peek_mut_impl() }
+        }
+
+        /// Removes the element from the beginning of the queue. If the queue is empty it returns [`None`].
         ///
         /// # Safety
         ///
@@ -269,7 +303,7 @@ pub mod details {
             self.pop_impl()
         }
 
-        /// Adds an element to the queue. If the queue is full it returns false, otherwise true.
+        /// Adds an element at the end of the queue. If the queue is full it returns false, otherwise true.
         ///
         /// # Safety
         ///
@@ -279,7 +313,7 @@ pub mod details {
             self.push_impl(value)
         }
 
-        /// Adds an element to the queue. If the queue is full it returns the oldest element,
+        /// Adds an element at the end of the queue. If the queue is full it returns the oldest element,
         /// otherwise [`None`].
         ///
         /// # Safety
@@ -331,12 +365,37 @@ pub mod details {
             while self.pop_impl().is_some() {}
         }
 
-        pub(crate) unsafe fn pop_impl(&mut self) -> Option<T> {
+        pub(crate) unsafe fn peek_mut_impl(&mut self) -> Option<&mut T> {
+            self.verify_init(&format!("Queue<{}>::pop()", std::any::type_name::<T>()));
+
+            if self.is_empty() {
+                return None;
+            }
+
+            let index = (self.start - self.len) % self.capacity;
+
+            Some((*self.data_ptr.as_mut_ptr().add(index)).assume_init_mut())
+        }
+
+        pub(crate) unsafe fn peek_impl(&self) -> Option<&T> {
+            self.verify_init(&format!("Queue<{}>::pop()", std::any::type_name::<T>()));
+
             if self.is_empty() {
                 return None;
             }
 
+            let index = (self.start - self.len) % self.capacity;
+
+            Some((*self.data_ptr.as_ptr().add(index)).assume_init_ref())
+        }
+
+        pub(crate) unsafe fn pop_impl(&mut self) -> Option<T> {
             self.verify_init(&format!("Queue<{}>::pop()", std::any::type_name::<T>()));
+
+            if self.is_empty() {
+                return None;
+            }
+
             let index = (self.start - self.len) % self.capacity;
             self.len -= 1;
             let value = std::mem::replace(
@@ -347,27 +406,28 @@ pub mod details {
         }
 
         pub(crate) unsafe fn push_impl(&mut self, value: T) -> bool {
+            self.verify_init(&format!("Queue<{}>::push()", std::any::type_name::<T>()));
+
             if self.len == self.capacity {
                 return false;
             }
 
-            self.verify_init(&format!("Queue<{}>::push()", std::any::type_name::<T>()));
-
             self.unchecked_push(value);
             true
         }
 
         pub(crate) unsafe fn push_with_overflow_impl(&mut self, value: T) -> Option<T> {
+            self.verify_init(&format!(
+                "Queue<{}>::push_with_overflow()",
+                std::any::type_name::<T>()
+            ));
+
             let overridden_value = if self.len() == self.capacity() {
                 self.pop_impl()
             } else {
                 None
             };
 
-            self.verify_init(&format!(
-                "Queue<{}>::push_with_overflow()",
-                std::any::type_name::<T>()
-            ));
             self.unchecked_push(value);
             overridden_value
         }
@@ -458,17 +518,29 @@ impl<T, const CAPACITY: usize> FixedSizeQueue<T, CAPACITY> {
         unsafe { self.state.clear_impl() }
     }
 
-    /// Acquire an element from the queue. If the queue is empty it returns [`None`].
+    /// Returns a reference to the element from the beginning of the queue without removing it.
+    /// If the queue is empty it returns [`None`].
+    pub fn peek(&self) -> Option<&T> {
+        unsafe { self.state.peek_impl() }
+    }
+
+    /// Returns a mutable reference to the element from the beginning of the queue without removing it.
+    /// If the queue is empty it returns [`None`].
+    pub fn peek_mut(&mut self) -> Option<&mut T> {
+        unsafe { self.state.peek_mut_impl() }
+    }
+
+    /// Removes the element from the beginning of the queue. If the queue is empty it returns [`None`].
     pub fn pop(&mut self) -> Option<T> {
         unsafe { self.state.pop_impl() }
     }
 
-    /// Adds an element to the queue. If the queue is full it returns false, otherwise true.
+    /// Adds an element at the end of the queue. If the queue is full it returns false, otherwise true.
     pub fn push(&mut self, value: T) -> bool {
         unsafe { self.state.push_impl(value) }
     }
 
-    /// Adds an element to the queue. If the queue is full it returns the oldest element,
+    /// Adds an element at the end of the queue. If the queue is full it returns the oldest element,
     /// otherwise [`None`].
     pub fn push_with_overflow(&mut self, value: T) -> Option<T> {
         unsafe { self.state.push_with_overflow_impl(value) }

iceoryx2-bb/container/tests/queue_tests.rs

@@ -307,4 +307,22 @@ mod queue {
         assert_that!(unsafe {sut.assume_init_mut()}.pop(), eq Some(123));
         assert_that!(unsafe {sut.assume_init_mut()}.pop(), eq Some(456));
     }
+
+    #[test]
+    fn peek_works() {
+        let mut sut = Sut::new();
+
+        assert_that!(sut.peek(), is_none);
+        assert_that!(sut.peek_mut(), is_none);
+
+        sut.push(8781);
+
+        assert_that!(*sut.peek().unwrap(), eq 8781);
+        assert_that!(*sut.peek_mut().unwrap(), eq 8781);
+
+        *sut.peek_mut().unwrap() = 99182;
+
+        assert_that!(*sut.peek().unwrap(), eq 99182);
+        assert_that!(*sut.peek_mut().unwrap(), eq 99182);
+    }
 }

iceoryx2-cal/src/static_storage/file.rs

@@ -438,7 +438,7 @@ impl crate::static_storage::StaticStorageBuilder<Storage> for Builder {
             "{} due to a failure while reading the files metadata.", msg);
 
             if metadata.permission() != FINAL_PERMISSIONS {
-                if elapsed_time >= timeout {
+                if elapsed_time > timeout {
                     fail!(from origin,
                         with StaticStorageOpenError::InitializationNotYetFinalized,
                         "{} since the static storage is still being created (in locked  state), try later.",

iceoryx2-cal/src/static_storage/process_local.rs

@@ -325,7 +325,7 @@ impl StaticStorageBuilder<Storage> for Builder {
 
             let entry = entry.unwrap();
             if entry.content.is_locked {
-                if elapsed_time >= timeout {
+                if elapsed_time > timeout {
                     fail!(from self, with StaticStorageOpenError::InitializationNotYetFinalized,
                         "{} since the static storage is still being created (in locked state), try later.", msg);
                 }

iceoryx2-ffi/ffi/src/api/subscriber.rs

@@ -96,7 +96,7 @@ impl SubscriberUnion {
 #[repr(C)]
 #[repr(align(16))] // alignment of Option<SubscriberUnion>
 pub struct iox2_subscriber_storage_t {
-    internal: [u8; 512], // magic number obtained with size_of::<Option<SubscriberUnion>>()
+    internal: [u8; 816], // magic number obtained with size_of::<Option<SubscriberUnion>>()
 }
 
 #[repr(C)]

iceoryx2/src/config.rs

@@ -241,6 +241,12 @@ pub struct PublishSubscribe {
     /// [`crate::port::publisher::Publisher`] when the [`crate::port::subscriber::Subscriber`]s
     /// buffer is full.
     pub unable_to_deliver_strategy: UnableToDeliverStrategy,
+    /// Defines the size of the internal [`Subscriber`](crate::port::subscriber::Subscriber)
+    /// buffer that contains expired connections. An
+    /// connection is expired when the [`Publisher`](crate::port::publisher::Publisher)
+    /// disconnected from a service and the connection
+    /// still contains unconsumed [`Sample`](crate::sample::Sample)s.
+    pub subscriber_expired_connection_buffer: usize,
 }
 
 /// Default settings for the event messaging pattern. These settings are used unless
@@ -312,6 +318,7 @@ impl Default for Config {
                     publisher_max_loaned_samples: 2,
                     enable_safe_overflow: true,
                     unable_to_deliver_strategy: UnableToDeliverStrategy::Block,
+                    subscriber_expired_connection_buffer: 128,
                 },
                 event: Event {
                     max_listeners: 2,

iceoryx2/src/port/details/publisher_connections.rs

@@ -78,7 +78,7 @@ impl<Service: service::Service> Connection<Service> {
 }
 #[derive(Debug)]
 pub(crate) struct PublisherConnections<Service: service::Service> {
-    connections: Vec<UnsafeCell<Option<Connection<Service>>>>,
+    connections: Vec<UnsafeCell<Option<Arc<Connection<Service>>>>>,
     subscriber_id: UniqueSubscriberId,
     pub(crate) service_state: Arc<ServiceState<Service>>,
     pub(crate) static_config: StaticConfig,
@@ -106,13 +106,13 @@ impl<Service: service::Service> PublisherConnections<Service> {
         self.subscriber_id
     }
 
-    pub(crate) fn get(&self, index: usize) -> &Option<Connection<Service>> {
+    pub(crate) fn get(&self, index: usize) -> &Option<Arc<Connection<Service>>> {
         unsafe { &*self.connections[index].get() }
     }
 
     // only used internally as convinience function
     #[allow(clippy::mut_from_ref)]
-    pub(crate) fn get_mut(&self, index: usize) -> &mut Option<Connection<Service>> {
+    pub(crate) fn get_mut(&self, index: usize) -> &mut Option<Arc<Connection<Service>>> {
         #[deny(clippy::mut_from_ref)]
         unsafe {
             &mut *self.connections[index].get()
@@ -124,7 +124,7 @@ impl<Service: service::Service> PublisherConnections<Service> {
         index: usize,
         details: &PublisherDetails,
     ) -> Result<(), ConnectionFailure> {
-        *self.get_mut(index) = Some(Connection::new(self, details)?);
+        *self.get_mut(index) = Some(Arc::new(Connection::new(self, details)?));
 
         Ok(())
     }