[#429] Use Vec instead of Queue for to_be_removed_connections

Author: elfenpiff

iceoryx2/src/port/client.rs

@@ -40,7 +40,7 @@ use alloc::sync::Arc;
 use core::{
     cell::UnsafeCell, fmt::Debug, marker::PhantomData, mem::MaybeUninit, sync::atomic::Ordering,
 };
-use iceoryx2_bb_container::queue::Queue;
+use iceoryx2_bb_container::{queue::Queue, vec::Vec};
 
 use iceoryx2_bb_elementary::{cyclic_tagger::CyclicTagger, CallbackProgression};
 use iceoryx2_bb_lock_free::mpmc::container::{ContainerHandle, ContainerState};
@@ -371,14 +371,12 @@ impl<
                     number_of_channels: 1,
                 },
                 response_receiver: Receiver {
-                    connections: (0..server_list.capacity())
-                        .map(|_| UnsafeCell::new(None))
-                        .collect(),
+                    connections: Vec::from_fn(server_list.capacity(), |_| UnsafeCell::new(None)),
                     receiver_port_id: client_port_id.value(),
                     service_state: service.__internal_state().clone(),
                     buffer_size: static_config.max_response_buffer_size,
                     tagger: CyclicTagger::new(),
-                    to_be_removed_connections: Some(UnsafeCell::new(Queue::new(
+                    to_be_removed_connections: Some(UnsafeCell::new(Vec::new(
                         service
                             .__internal_state()
                             .shared_node

iceoryx2/src/port/details/receiver.rs

@@ -24,7 +24,7 @@ use crate::service::static_config::message_type_details::MessageTypeDetails;
 use crate::service::ServiceState;
 use crate::service::{self, config_scheme::connection_config, naming_scheme::connection_name};
 use alloc::sync::Arc;
-use iceoryx2_bb_container::queue::Queue;
+use iceoryx2_bb_container::vec::Vec;
 use iceoryx2_bb_elementary::cyclic_tagger::*;
 use iceoryx2_bb_log::{fail, warn};
 use iceoryx2_cal::named_concept::NamedConceptBuilder;
@@ -111,7 +111,7 @@ pub(crate) struct Receiver<Service: service::Service> {
     pub(crate) service_state: Arc<ServiceState<Service>>,
     pub(crate) buffer_size: usize,
     pub(crate) tagger: CyclicTagger,
-    pub(crate) to_be_removed_connections: Option<UnsafeCell<Queue<Arc<Connection<Service>>>>>,
+    pub(crate) to_be_removed_connections: Option<UnsafeCell<Vec<Arc<Connection<Service>>>>>,
     pub(crate) degradation_callback: Option<DegradationCallback<'static>>,
     pub(crate) message_type_details: MessageTypeDetails,
     pub(crate) receiver_max_borrowed_samples: usize,
@@ -278,22 +278,45 @@ impl<Service: service::Service> Receiver<Service> {
         &self,
         channel_id: ChannelId,
     ) -> Result<Option<(ChunkDetails<Service>, Chunk)>, ReceiveError> {
+        let msg = "Unable to receive data";
         if let Some(to_be_removed_connections) = &self.to_be_removed_connections {
             let to_be_removed_connections = unsafe { &mut *to_be_removed_connections.get() };
 
-            if let Some(connection) = to_be_removed_connections.peek() {
+            let mut clean_connections = Vec::new(to_be_removed_connections.capacity());
+            for (n, connection) in to_be_removed_connections.iter_mut().enumerate() {
+                if connection.receiver.borrow_count(channel_id)
+                    == connection.receiver.max_borrowed_samples()
+                {
+                    continue;
+                }
+
                 if let Some((details, absolute_address)) =
                     self.receive_from_connection(connection, channel_id)?
                 {
                     return Ok(Some((details, absolute_address)));
                 } else {
-                    to_be_removed_connections.pop();
+                    clean_connections.push(n);
                 }
             }
+
+            for idx in clean_connections.iter().rev() {
+                to_be_removed_connections.remove(*idx);
+            }
         }
 
+        let mut active_channel_count = 0;
+        let mut all_channels_exceed_max_borrows = true;
         for id in 0..self.len() {
             if let Some(ref mut connection) = &mut self.get_mut(id) {
+                active_channel_count += 0;
+                if connection.receiver.borrow_count(channel_id)
+                    >= connection.receiver.max_borrowed_samples()
+                {
+                    continue;
+                } else {
+                    all_channels_exceed_max_borrows = false;
+                }
+
                 if let Some((details, absolute_address)) =
                     self.receive_from_connection(connection, channel_id)?
                 {
@@ -302,6 +325,11 @@ impl<Service: service::Service> Receiver<Service> {
             }
         }
 
+        if all_channels_exceed_max_borrows && active_channel_count != 0 {
+            fail!(from self, with ReceiveError::ExceedsMaxBorrows,
+                 "{msg} since every channel exceeds the max number of borrows.");
+        }
+
         Ok(None)
     }
 

iceoryx2/src/port/server.rs

@@ -37,6 +37,7 @@ extern crate alloc;
 use alloc::sync::Arc;
 use core::{cell::UnsafeCell, sync::atomic::Ordering};
 use core::{fmt::Debug, marker::PhantomData};
+use iceoryx2_bb_container::vec::Vec;
 use iceoryx2_cal::zero_copy_connection::ChannelId;
 use iceoryx2_pal_concurrency_sync::iox_atomic::{IoxAtomicBool, IoxAtomicUsize};
 
@@ -236,9 +237,7 @@ impl<
             .clients;
 
         let request_receiver = Receiver {
-            connections: (0..client_list.capacity())
-                .map(|_| UnsafeCell::new(None))
-                .collect(),
+            connections: Vec::from_fn(client_list.capacity(), |_| UnsafeCell::new(None)),
             receiver_port_id: server_port_id.value(),
             service_state: service.__internal_state().clone(),
             message_type_details: static_config.request_message_type_details.clone(),

iceoryx2/src/port/subscriber.rs

@@ -39,7 +39,7 @@ use core::sync::atomic::Ordering;
 
 extern crate alloc;
 
-use iceoryx2_bb_container::queue::Queue;
+use iceoryx2_bb_container::vec::Vec;
 use iceoryx2_bb_elementary::cyclic_tagger::CyclicTagger;
 use iceoryx2_bb_elementary::CallbackProgression;
 use iceoryx2_bb_lock_free::mpmc::container::{ContainerHandle, ContainerState};
@@ -146,17 +146,15 @@ impl<Service: service::Service, Payload: Debug + ?Sized, UserHeader: Debug>
         };
 
         let receiver = Receiver {
-            connections: (0..publisher_list.capacity())
-                .map(|_| UnsafeCell::new(None))
-                .collect(),
+            connections: Vec::from_fn(publisher_list.capacity(), |_| UnsafeCell::new(None)),
             receiver_port_id: subscriber_id.value(),
             service_state: service.__internal_state().clone(),
             message_type_details: static_config.message_type_details.clone(),
             receiver_max_borrowed_samples: static_config.subscriber_max_borrowed_samples,
             enable_safe_overflow: static_config.enable_safe_overflow,
             buffer_size,
             tagger: CyclicTagger::new(),
-            to_be_removed_connections: Some(UnsafeCell::new(Queue::new(
+            to_be_removed_connections: Some(UnsafeCell::new(Vec::new(
                 service
                     .__internal_state()
                     .shared_node