Merge pull request #680 from elfenpiff/iox2-429-add-response-channel

[#429] add response channel

Author: elfenpiff

Cargo.lock

@@ -119,6 +119,16 @@ Adjusting `global` settings ensures a non-interfering setup.
 
 ### Service: Request Response Messaging Pattern
 
+* `defaults.request-response.client-unable-to-deliver-strategy` -
+  [`Block`|`DiscardSample`]: Default strategy for non-overflowing setups
+  when delivery fails.
+* `defaults.request-response.client-expired-connection-buffer` - [int]:
+  Expired connection buffer size of the client. Connections to servers
+  are expired when the server disconnected from the service and the
+  connection contains unconsumed responses.
+* `defaults.request-response.enable-fire-and-forget-requests` -
+  [`true`|`false`]: Enables the client to send requests without
+  expecting a response.
 * `defaults.request-response.enable-safe-overflow-for-requests` -
   [`true`|`false`]: Defines if the request buffer of the service safely
   overflows.
@@ -130,14 +140,17 @@ Adjusting `global` settings ensures a non-interfering setup.
 * `defaults.request-response.max-borrowed-responses-per-pending-response` -
   [int]: The maximum number of borrowed responses a client can hold in
   parallel per pending response.
+* `defaults.request-response.max-clients` - [int]:
+  The maximum amount of supported clients.
+* `defaults.request-response.max-nodes` - [int]:
+  The maximum amount of supported nodes. Defines indirectly how many
+  processes can open the service at the same time.
 * `defaults.request-response.max-response-buffer-size` - [int]:
   The maximum buffer size for responses for an active request.
 * `defaults.request-response.max-request-buffer-size` - [int]:
   The maximum buffer size for requests for a server.
 * `defaults.request-response.max-servers` - [int]:
   The maximum amount of supported servers.
-* `defaults.request-response.max-clients` - [int]:
-  The maximum amount of supported clients.
-* `defaults.request-response.max-nodes` - [int]:
-  The maximum amount of supported nodes. Defines indirectly how many
-  processes can open the service at the same time.
+* `defaults.request-response.server-unable-to-deliver-strategy` -
+  [`Block`|`DiscardSample`]: Default strategy for non-overflowing setups
+  when delivery fails.

config/README.md

@@ -22,6 +22,9 @@ creation-timeout.secs = 0
 creation-timeout.nanos = 500000000
 
 [defaults.request-response]
+client-expired-connection-buffer = 128
+client-unable-to-deliver-strategy = 'Block'     # or 'DiscardSample'
+enable-fire-and-forget-requests = true
 enable-safe-overflow-for-requests = true
 enable-safe-overflow-for-responses = true
 max-active-requests-per-client = 4
@@ -30,9 +33,8 @@ max-servers = 2
 max-clients = 8
 max-nodes = 20
 max-borrowed-responses-per-pending-response = 2
-client-max-loaned-requests = 2
+max-loaned-requests = 2
 server-max-loaned-responses-per-request = 2
-client-unable-to-deliver-strategy = 'Block'     # or 'DiscardSample'
 server-unable-to-deliver-strategy = 'Block'     # or 'DiscardSample'
 
 [defaults.publish-subscribe]

config/iceoryx2.toml

@@ -15,6 +15,8 @@
     [#264](https://github.com/eclipse-iceoryx/iceoryx2/issues/264)
 * Read LogLevel from environment variable
     [#396](https://github.com/eclipse-iceoryx/iceoryx2/issues/396)
+* Add Request-Response messaging pattern
+    [#429](https://github.com/eclipse-iceoryx/iceoryx2/issues/429)
 * Lookup config file in default locations
     [#442](https://github.com/eclipse-iceoryx/iceoryx2/issues/442)
 * Introduce `socket_pair` abstraction in POSIX wrapper
@@ -82,7 +84,6 @@
 * Remove the `print_system_configuration()` function in
 `iceoryx2-bb/posix/src/system_configuration.rs` file and move it into the CLI `iox2-config`
     [#432](https://github.com/eclipse-iceoryx/iceoryx2/issues/432)
-
 * Remove obsolete POSIX wrapper
     [#594](https://github.com/eclipse-iceoryx/iceoryx2/issues/594)
 * Updated all dependencies and increased MSRV to 1.81

doc/release-notes/iceoryx2-unreleased.md

@@ -29,12 +29,21 @@ fn main() -> Result<(), Box<dyn core::error::Error>> {
 
     let mut counter: u64 = 0;
 
+    // sending first request by using slower, inefficient copy API
+    println!("Send request {} ...", counter);
+    let mut pending_response = client.send_copy(counter)?;
     while node.wait(CYCLE_TIME).is_ok() {
+        // acquire all responses from our buffer
+        while let Some(response) = pending_response.receive()? {
+            println!("  received response: {:?}", *response);
+        }
+
         counter += 1;
+        // send all other requests by using zero copy API
         let request = client.loan_uninit()?;
         let request = request.write_payload(counter);
 
-        let _pending_response = request.send()?;
+        pending_response = request.send()?;
 
         println!("Send request {} ...", counter);
     }

examples/rust/request_response/client.rs

@@ -14,7 +14,7 @@ use core::time::Duration;
 use examples_common::TransmissionData;
 use iceoryx2::prelude::*;
 
-const CYCLE_TIME: Duration = Duration::from_millis(10);
+const CYCLE_TIME: Duration = Duration::from_millis(100);
 
 fn main() -> Result<(), Box<dyn core::error::Error>> {
     set_log_level_from_env_or(LogLevel::Info);
@@ -27,10 +27,38 @@ fn main() -> Result<(), Box<dyn core::error::Error>> {
 
     let server = service.server_builder().create()?;
 
+    let mut counter = 0;
     while node.wait(CYCLE_TIME).is_ok() {
         while let Some(active_request) = server.receive()? {
             println!("received request: {:?}", *active_request);
+
+            // send first response by using the slower, non-zero-copy
+            // API
+            active_request.send_copy(TransmissionData {
+                x: 5 + counter,
+                y: 6 * counter,
+                funky: 7.77,
+            })?;
+
+            // use zero copy API, send out some samples to utilize
+            // the streaming API
+            for n in 0..*active_request % 2 {
+                let sample = active_request.loan_uninit()?;
+                let sample = sample.write_payload(TransmissionData {
+                    x: counter as i32 * (n as i32 + 1),
+                    y: counter as i32 + n as i32,
+                    funky: counter as f64 * 0.1234,
+                });
+                sample.send()?;
+            }
+
+            // when active_request is dropped it marks the connection so
+            // that the corresponding pending response sees that no more
+            // responses are arriving
+            drop(active_request);
         }
+
+        counter += 1;
     }
 
     println!("exit");

examples/rust/request_response/server.rs

@@ -294,6 +294,17 @@ pub mod details {
             true
         }
 
+        unsafe fn remove_impl(&mut self, index: usize) -> T {
+            self.verify_init("remove()");
+            debug_assert!(index < self.len());
+
+            let ptr = self.as_mut_ptr().add(index);
+            let value = core::ptr::read(ptr);
+            core::ptr::copy(ptr.add(1), ptr, self.len - index - 1);
+            self.len -= 1;
+            value
+        }
+
         unsafe fn pop_impl(&mut self) -> Option<T> {
             if self.is_empty() {
                 return None;
@@ -329,7 +340,7 @@ pub mod details {
     }
 
     impl<T> MetaVec<T, GenericOwningPointer> {
-        /// Creates a new [`Queue`] with the provided capacity
+        /// Creates a new [`Vec`] with the provided capacity
         pub fn new(capacity: usize) -> Self {
             Self {
                 data_ptr: OwningPointer::<MaybeUninit<T>>::new_with_alloc(capacity),
@@ -340,6 +351,16 @@ pub mod details {
             }
         }
 
+        /// Creates a new [`Vec`] with the provided capacity and fills it by
+        /// using the provided callback
+        pub fn from_fn<F: FnMut(usize) -> T>(capacity: usize, mut callback: F) -> Self {
+            let mut new_self = Self::new(capacity);
+            for n in 0..capacity {
+                new_self.push(callback(n));
+            }
+            new_self
+        }
+
         /// Adds an element at the end of the vector. If the vector is full and the element cannot be
         /// added it returns false, otherwise true.
         pub fn push(&mut self, value: T) -> bool {
@@ -373,6 +394,11 @@ pub mod details {
             unsafe { self.pop_impl() }
         }
 
+        /// Removes the element at the provided index and returns it.
+        pub fn remove(&mut self, index: usize) -> T {
+            unsafe { self.remove_impl(index) }
+        }
+
         /// Removes all elements from the vector
         pub fn clear(&mut self) {
             unsafe { self.clear_impl() }
@@ -455,6 +481,16 @@ pub mod details {
             self.pop_impl()
         }
 
+        /// Removes the element at the provided index and returns it.
+        ///
+        /// # Safety
+        ///
+        ///  * [`RelocatableVec::init()`] must be called once before
+        ///
+        pub unsafe fn remove(&mut self, index: usize) -> T {
+            unsafe { self.remove_impl(index) }
+        }
+
         /// Removes all elements from the vector
         ///
         /// # Safety
@@ -681,6 +717,11 @@ impl<T, const CAPACITY: usize> FixedSizeVec<T, CAPACITY> {
         unsafe { self.state.pop() }
     }
 
+    /// Removes the element at the provided index and returns it.
+    pub fn remove(&mut self, index: usize) -> T {
+        unsafe { self.state.remove(index) }
+    }
+
     /// Removes all elements from the vector
     pub fn clear(&mut self) {
         unsafe { self.state.clear() }

iceoryx2-bb/container/src/vec.rs

@@ -333,4 +333,40 @@ mod vec {
         assert_that!(sut.pop(), eq Some(TEST_VALUE));
         assert_that!(sut, len 0);
     }
+
+    #[test]
+    fn remove_reverse_order_works() {
+        const CAPACITY: usize = 12;
+        let mut sut = Vec::<usize>::new(CAPACITY);
+
+        for n in 0..CAPACITY {
+            sut.push(n);
+        }
+
+        for n in (0..CAPACITY).rev() {
+            assert_that!(sut.remove(n), eq n);
+
+            for (idx, v) in sut.iter().enumerate() {
+                assert_that!(*v, eq idx);
+            }
+        }
+    }
+
+    #[test]
+    fn remove_works() {
+        const CAPACITY: usize = 12;
+        let mut sut = Vec::<usize>::new(CAPACITY);
+
+        for n in 0..CAPACITY {
+            sut.push(n);
+        }
+
+        for n in 0..CAPACITY {
+            assert_that!(sut.remove(0), eq n);
+
+            for (idx, v) in sut.iter().enumerate() {
+                assert_that!(*v, eq idx + n + 1);
+            }
+        }
+    }
 }

iceoryx2-bb/container/tests/vec_tests.rs

@@ -30,6 +30,7 @@ pub enum MemoryLockCreationError {
     UnknownError(i32),
 }
 
+#[derive(Debug, Clone, Copy, Eq, Hash, PartialEq)]
 pub enum MemoryLockAllError {
     UnableToLock,
     WouldExceedMainMemory,

iceoryx2-bb/posix/src/memory_lock.rs

@@ -72,6 +72,17 @@ macro_rules! assert_that {
             }
         }
     };
+    ($lhs:expr, aligned_to $rhs:expr) => {
+        {
+            let lval = $lhs as usize;
+            let rval = $rhs as usize;
+            let act_result = lval % rval;
+
+            if !(act_result == 0) {
+                assert_that!(message $lhs, $rhs, lval, rval, "aligned to");
+            }
+        }
+    };
     ($lhs:expr, mod $rhs:expr, is $result:expr) => {
         {
             let lval = &$lhs;

iceoryx2-bb/testing/src/assert.rs

@@ -44,6 +44,9 @@ rust_library(
         "@crate_index//:tiny-fn",
         "@crate_index//:toml",
     ],
+    proc_macro_deps = [
+        "//iceoryx2-bb/derive-macros:iceoryx2-bb-derive-macros",
+    ],
 )
 
 rust_test_suite(

iceoryx2-cal/BUILD.bazel

@@ -19,6 +19,7 @@ dev_permissions = []
 [dependencies]
 iceoryx2-bb-posix = { workspace = true }
 iceoryx2-bb-container = { workspace = true }
+iceoryx2-bb-derive-macros = { workspace = true }
 iceoryx2-bb-system-types = { workspace = true }
 iceoryx2-bb-log = { workspace = true }
 iceoryx2-bb-elementary = { workspace = true }

iceoryx2-cal/Cargo.toml

@@ -106,7 +106,12 @@ enum_gen! {
 pub trait DynamicStorageBuilder<'builder, T: Send + Sync, D: DynamicStorage<T>>:
     Debug + Sized + NamedConceptBuilder<D>
 {
-    /// Defines if a newly created [`DynamicStorage`] owns the underlying resources
+    /// Defines if `T::Drop` shall be called when the [`DynamicStorage`] is removed. The default
+    /// is [`true`].
+    fn call_drop_on_destruction(self, value: bool) -> Self;
+
+    /// Defines if a newly created [`DynamicStorage`] owns the underlying resources. The default
+    /// is [`true`].
     fn has_ownership(self, value: bool) -> Self;
 
     /// Sets the size of the supplementary data. Only relevant when it is newly created otherwise