[#116] Fix retrieve buffer overflow

doc/release-notes/iceoryx2-unreleased.md

@@ -15,6 +15,7 @@
 
  <!-- NOTE: Add new entries sorted by issue number to minimize the possibility of conflicts when merging. -->
 
+ * Fix retrieve channel overflow caused by big publisher loans [#116](https://github.com/eclipse-iceoryx/iceoryx2/issues/116)
  * Fix `open_or_create` race [#108](https://github.com/eclipse-iceoryx/iceoryx2/issues/108)
  * Fix undefined behavior in `spsc::{queue|index_queue}` [#87](https://github.com/eclipse-iceoryx/iceoryx2/issues/87)
 

iceoryx2/src/port/publisher.rs

@@ -471,9 +471,10 @@ impl<'a, Service: service::Service, MessageType: Debug> UninitLoan<MessageType>
     for Publisher<'a, Service, MessageType>
 {
     fn loan_uninit(&self) -> Result<SampleMut<MaybeUninit<MessageType>>, PublisherLoanError> {
-        self.retrieve_returned_samples();
         let msg = "Unable to loan Sample";
 
+        self.retrieve_returned_samples();
+
         if self.loan_counter.load(Ordering::Relaxed) >= self.config.max_loaned_samples {
             fail!(from self, with PublisherLoanError::ExceedsMaxLoanedChunks,
                 "{} since already {} samples were loaned and it would exceed the maximum of parallel loans of {}. Release or send a loaned sample to loan another sample.",
@@ -534,6 +535,7 @@ impl<'a, Service: service::Service, MessageType: Debug> PublishMgmt
     }
 
     fn send_impl(&self, address_to_chunk: usize) -> Result<usize, ConnectionFailure> {
+        self.retrieve_returned_samples();
         fail!(from self, when self.update_connections(),
             "Unable to send sample since the connections could not be updated.");
 

iceoryx2/tests/service_publish_subscribe_tests.rs

@@ -946,6 +946,128 @@ mod service_publish_subscribe {
         );
     }
 
+    fn retrieve_channel_capacity_is_never_exceeded<Sut: Service>(
+        publisher_borrow_size: usize,
+        buffer_size: usize,
+        max_borrow: usize,
+    ) {
+        const ITERATIONS: usize = 16;
+        let service_name = generate_name();
+
+        let sut = Sut::new(&service_name)
+            .publish_subscribe()
+            .max_publishers(1)
+            .max_subscribers(1)
+            .enable_safe_overflow(false)
+            .history_size(0)
+            .subscriber_max_buffer_size(buffer_size)
+            .subscriber_max_borrowed_samples(max_borrow)
+            .create::<usize>()
+            .unwrap();
+
+        let sut_publisher = sut
+            .publisher()
+            .max_loaned_samples(publisher_borrow_size)
+            .create()
+            .unwrap();
+        let sut_subscriber = sut.subscriber().create().unwrap();
+
+        let mut borrowed_samples = vec![];
+        let mut cached_samples = vec![];
+
+        let mut send_sample = || {
+            if borrowed_samples.is_empty() {
+                for _ in 0..publisher_borrow_size {
+                    borrowed_samples.push(sut_publisher.loan().unwrap());
+                }
+            }
+
+            let sample = borrowed_samples.pop().unwrap();
+            sample.send().unwrap();
+        };
+
+        for _ in 0..ITERATIONS {
+            for _ in 0..max_borrow {
+                send_sample();
+                cached_samples.push(sut_subscriber.receive().unwrap().unwrap());
+            }
+
+            for _ in 0..buffer_size {
+                send_sample();
+            }
+
+            cached_samples.clear();
+            for _ in 0..buffer_size {
+                sut_subscriber.receive().unwrap().unwrap();
+            }
+        }
+    }
+
+    #[test]
+    fn retrieve_channel_capacity_is_never_exceeded_with_large_publisher_borrow_size<
+        Sut: Service,
+    >() {
+        const PUBLISHER_BORROW_SIZE: usize = 10;
+        const BUFFER_SIZE: usize = 1;
+        const MAX_BORROW: usize = 1;
+        retrieve_channel_capacity_is_never_exceeded::<Sut>(
+            PUBLISHER_BORROW_SIZE,
+            BUFFER_SIZE,
+            MAX_BORROW,
+        );
+    }
+
+    #[test]
+    fn retrieve_channel_capacity_is_never_exceeded_with_large_buffer_size<Sut: Service>() {
+        const PUBLISHER_BORROW_SIZE: usize = 1;
+        const BUFFER_SIZE: usize = 10;
+        const MAX_BORROW: usize = 1;
+        retrieve_channel_capacity_is_never_exceeded::<Sut>(
+            PUBLISHER_BORROW_SIZE,
+            BUFFER_SIZE,
+            MAX_BORROW,
+        );
+    }
+
+    #[test]
+    fn retrieve_channel_capacity_is_never_exceeded_with_large_max_borrow<Sut: Service>() {
+        const PUBLISHER_BORROW_SIZE: usize = 1;
+        const BUFFER_SIZE: usize = 1;
+        const MAX_BORROW: usize = 10;
+
+        retrieve_channel_capacity_is_never_exceeded::<Sut>(
+            PUBLISHER_BORROW_SIZE,
+            BUFFER_SIZE,
+            MAX_BORROW,
+        );
+    }
+
+    #[test]
+    fn retrieve_channel_capacity_is_never_exceeded_with_large_settings<Sut: Service>() {
+        const PUBLISHER_BORROW_SIZE: usize = 20;
+        const BUFFER_SIZE: usize = 14;
+        const MAX_BORROW: usize = 15;
+
+        retrieve_channel_capacity_is_never_exceeded::<Sut>(
+            PUBLISHER_BORROW_SIZE,
+            BUFFER_SIZE,
+            MAX_BORROW,
+        );
+    }
+
+    #[test]
+    fn retrieve_channel_capacity_is_never_exceeded_with_small_settings<Sut: Service>() {
+        const PUBLISHER_BORROW_SIZE: usize = 1;
+        const BUFFER_SIZE: usize = 1;
+        const MAX_BORROW: usize = 1;
+
+        retrieve_channel_capacity_is_never_exceeded::<Sut>(
+            PUBLISHER_BORROW_SIZE,
+            BUFFER_SIZE,
+            MAX_BORROW,
+        );
+    }
+
     #[test]
     fn creating_max_supported_amount_of_ports_work<Sut: Service>() {
         const MAX_PUBLISHERS: usize = 4;