[eclipse-iceoryx#690] Implement slice API in active request

elfenpiff · elfenpiff · commit bab8dd1692be · 2025-04-28T18:53:19.000+02:00
diff --git a/iceoryx2/src/active_request.rs b/iceoryx2/src/active_request.rs
@@ -42,13 +42,19 @@
 extern crate alloc;
 
 use alloc::sync::Arc;
-use core::{fmt::Debug, marker::PhantomData, mem::MaybeUninit, ops::Deref, sync::atomic::Ordering};
+use core::{
+    any::TypeId, fmt::Debug, marker::PhantomData, mem::MaybeUninit, ops::Deref,
+    sync::atomic::Ordering,
+};
 use iceoryx2_bb_elementary::zero_copy_send::ZeroCopySend;
 use iceoryx2_pal_concurrency_sync::iox_atomic::IoxAtomicUsize;
 
 use iceoryx2_bb_log::{error, fail};
 use iceoryx2_bb_posix::unique_system_id::UniqueSystemId;
-use iceoryx2_cal::zero_copy_connection::{ChannelId, ZeroCopyReceiver, ZeroCopyReleaseError};
+use iceoryx2_cal::{
+    shm_allocator::AllocationStrategy,
+    zero_copy_connection::{ChannelId, ZeroCopyReceiver, ZeroCopyReleaseError},
+};
 
 use crate::{
     port::{
@@ -60,7 +66,10 @@ use crate::{
     raw_sample::{RawSample, RawSampleMut},
     response_mut::ResponseMut,
     response_mut_uninit::ResponseMutUninit,
-    service::{self, builder::publish_subscribe::CustomPayloadMarker},
+    service::{
+        self, builder::publish_subscribe::CustomPayloadMarker,
+        static_config::message_type_details::TypeVariant,
+    },
 };
 
 /// Represents a one-to-one connection to a [`Client`](crate::port::client::Client)
@@ -217,6 +226,28 @@ impl<
     pub fn origin(&self) -> UniqueClientId {
         UniqueClientId(UniqueSystemId::from(self.details.origin))
     }
+
+    fn increment_loan_counter(&self) -> Result<(), LoanError> {
+        let mut current_loan_count = self.shared_loan_counter.load(Ordering::Relaxed);
+        loop {
+            if self.max_loan_count <= current_loan_count {
+                fail!(from self,
+                with LoanError::ExceedsMaxLoans,
+                "Unable to loan memory for Response since it would exceed the maximum number of loans of {}.",
+                self.max_loan_count);
+            }
+
+            match self.shared_loan_counter.compare_exchange(
+                current_loan_count,
+                current_loan_count + 1,
+                Ordering::Relaxed,
+                Ordering::Relaxed,
+            ) {
+                Ok(_) => return Ok(()),
+                Err(v) => current_loan_count = v,
+            }
+        }
+    }
 }
 
 ////////////////////////
@@ -259,25 +290,7 @@ impl<
         &self,
     ) -> Result<ResponseMutUninit<Service, MaybeUninit<ResponsePayload>, ResponseHeader>, LoanError>
     {
-        let mut current_loan_count = self.shared_loan_counter.load(Ordering::Relaxed);
-        loop {
-            if self.max_loan_count <= current_loan_count {
-                fail!(from self,
-                with LoanError::ExceedsMaxLoans,
-                "Unable to loan memory for Response since it would exceed the maximum number of loans of {}.",
-                self.max_loan_count);
-            }
-
-            match self.shared_loan_counter.compare_exchange(
-                current_loan_count,
-                current_loan_count + 1,
-                Ordering::Relaxed,
-                Ordering::Relaxed,
-            ) {
-                Ok(_) => break,
-                Err(v) => current_loan_count = v,
-            }
-        }
+        self.increment_loan_counter()?;
 
         let chunk = self
             .shared_state
@@ -412,32 +425,100 @@ impl<
         Service: crate::service::Service,
         RequestPayload: Debug + ZeroCopySend + ?Sized,
         RequestHeader: Debug + ZeroCopySend,
-        ResponsePayload: Debug + Default + ZeroCopySend,
+        ResponsePayload: Debug + Default + ZeroCopySend + 'static,
         ResponseHeader: Debug + ZeroCopySend,
     > ActiveRequest<Service, RequestPayload, RequestHeader, [ResponsePayload], ResponseHeader>
 {
     pub fn loan_slice(
         &self,
-        number_of_elements: usize,
+        slice_len: usize,
     ) -> Result<ResponseMut<Service, [ResponsePayload], ResponseHeader>, LoanError> {
-        todo!()
+        let response = self.loan_slice_uninit(slice_len)?;
+        Ok(response.write_from_fn(|_| ResponsePayload::default()))
     }
 }
 
 impl<
         Service: crate::service::Service,
         RequestPayload: Debug + ZeroCopySend + ?Sized,
         RequestHeader: Debug + ZeroCopySend,
-        ResponsePayload: Debug + ZeroCopySend,
+        ResponsePayload: Debug + ZeroCopySend + 'static,
         ResponseHeader: Debug + ZeroCopySend,
     > ActiveRequest<Service, RequestPayload, RequestHeader, [ResponsePayload], ResponseHeader>
 {
     pub fn loan_slice_uninit(
         &self,
-        number_of_elements: usize,
+        slice_len: usize,
     ) -> Result<ResponseMutUninit<Service, [MaybeUninit<ResponsePayload>], ResponseHeader>, LoanError>
     {
-        todo!()
+        debug_assert!(TypeId::of::<ResponsePayload>() != TypeId::of::<CustomPayloadMarker>());
+        unsafe { self.loan_slice_uninit_impl(slice_len, slice_len) }
+    }
+
+    unsafe fn loan_slice_uninit_impl(
+        &self,
+        slice_len: usize,
+        underlying_number_of_slice_elements: usize,
+    ) -> Result<ResponseMutUninit<Service, [MaybeUninit<ResponsePayload>], ResponseHeader>, LoanError>
+    {
+        let max_slice_len = self.shared_state.config.initial_max_slice_len;
+
+        if self.shared_state.config.allocation_strategy == AllocationStrategy::Static
+            && max_slice_len < slice_len
+        {
+            fail!(from self, with LoanError::ExceedsMaxLoanSize,
+                "Unable to loan slice with {} elements since it would exceed the max supported slice length of {}.",
+                slice_len, max_slice_len);
+        }
+
+        self.increment_loan_counter()?;
+
+        let response_layout = self.shared_state.response_sender.sample_layout(slice_len);
+        let chunk = self
+            .shared_state
+            .response_sender
+            .allocate(response_layout)?;
+
+        unsafe {
+            (chunk.header as *mut service::header::request_response::ResponseHeader).write(
+                service::header::request_response::ResponseHeader {
+                    server_port_id: UniqueServerId(UniqueSystemId::from(
+                        self.shared_state.response_sender.sender_port_id,
+                    )),
+                    request_id: self.request_id,
+                    number_of_elements: slice_len as _,
+                },
+            )
+        };
+
+        let ptr = unsafe {
+            RawSampleMut::<
+                service::header::request_response::ResponseHeader,
+                ResponseHeader,
+                [MaybeUninit<ResponsePayload>],
+            >::new_unchecked(
+                chunk.header.cast(),
+                chunk.user_header.cast(),
+                core::slice::from_raw_parts_mut(
+                    chunk.payload.cast(),
+                    underlying_number_of_slice_elements,
+                ),
+            )
+        };
+
+        Ok(ResponseMutUninit {
+            response: ResponseMut {
+                ptr,
+                shared_loan_counter: self.shared_loan_counter.clone(),
+                shared_state: self.shared_state.clone(),
+                offset_to_chunk: chunk.offset,
+                channel_id: self.channel_id,
+                connection_id: self.connection_id,
+                sample_size: chunk.size,
+                _response_payload: PhantomData,
+                _response_header: PhantomData,
+            },
+        })
     }
 }
 
@@ -462,7 +543,16 @@ impl<
         ResponseMutUninit<Service, [MaybeUninit<CustomPayloadMarker>], ResponseHeader>,
         LoanError,
     > {
-        todo!()
+        // TypeVariant::Dynamic == slice and only here it makes sense to loan more than one element
+        debug_assert!(
+            slice_len == 1
+                || self.shared_state.response_sender.payload_type_variant() == TypeVariant::Dynamic
+        );
+
+        self.loan_slice_uninit_impl(
+            slice_len,
+            self.shared_state.response_sender.payload_size() * slice_len,
+        )
     }
 }
 ////////////////////////
diff --git a/iceoryx2/src/port/client.rs b/iceoryx2/src/port/client.rs
@@ -309,12 +309,11 @@ impl<
 
         let global_config = service.__internal_state().shared_node.config();
         let segment_name = data_segment_name(client_port_id.value());
-        let data_segment_type = DataSegmentType::Static;
-        let max_number_of_segments =
-            DataSegment::<Service>::max_number_of_segments(data_segment_type);
         let data_segment_type = DataSegmentType::new_from_allocation_strategy(
             client_factory.config.allocation_strategy,
         );
+        let max_number_of_segments =
+            DataSegment::<Service>::max_number_of_segments(data_segment_type);
 
         let sample_layout = static_config
             .request_message_type_details
diff --git a/iceoryx2/src/port/server.rs b/iceoryx2/src/port/server.rs
@@ -49,6 +49,7 @@ use iceoryx2_bb_posix::unique_system_id::UniqueSystemId;
 use iceoryx2_cal::dynamic_storage::DynamicStorage;
 
 use crate::service::naming_scheme::data_segment_name;
+use crate::service::port_factory::server::LocalServerConfig;
 use crate::{
     active_request::ActiveRequest,
     prelude::PortFactory,
@@ -80,6 +81,7 @@ const REQUEST_CHANNEL_ID: ChannelId = ChannelId::new(0);
 
 #[derive(Debug)]
 pub(crate) struct SharedServerState<Service: service::Service> {
+    pub(crate) config: LocalServerConfig,
     pub(crate) response_sender: Sender<Service>,
     server_handle: UnsafeCell<Option<ContainerHandle>>,
     request_receiver: Receiver<Service>,
@@ -247,24 +249,45 @@ impl<
         };
 
         let global_config = service.__internal_state().shared_node.config();
-        let data_segment_type = DataSegmentType::Static;
+        let data_segment_type = DataSegmentType::new_from_allocation_strategy(
+            server_factory.config.allocation_strategy,
+        );
         let segment_name = data_segment_name(server_port_id.value());
         let max_number_of_segments =
             DataSegment::<Service>::max_number_of_segments(data_segment_type);
-        let data_segment = DataSegment::<Service>::create_static_segment(
-            &segment_name,
-            static_config.response_message_type_details.sample_layout(1),
-            global_config,
-            number_of_responses,
-        );
+        let sample_layout = static_config
+            .response_message_type_details
+            .sample_layout(server_factory.config.initial_max_slice_len);
+        let data_segment = match data_segment_type {
+            DataSegmentType::Static => DataSegment::<Service>::create_static_segment(
+                &segment_name,
+                static_config.response_message_type_details.sample_layout(1),
+                global_config,
+                number_of_responses,
+            ),
+            DataSegmentType::Dynamic => DataSegment::<Service>::create_dynamic_segment(
+                &segment_name,
+                sample_layout,
+                global_config,
+                number_of_responses,
+                server_factory.config.allocation_strategy,
+            ),
+        };
 
         let data_segment = fail!(from origin,
             when data_segment,
             with ServerCreateError::UnableToCreateDataSegment,
             "{} since the server data segment could not be created.", msg);
 
         let response_sender = Sender {
-            segment_states: vec![SegmentState::new(number_of_responses)],
+            segment_states: {
+                let mut v =
+                    std::vec::Vec::<SegmentState>::with_capacity(max_number_of_segments as usize);
+                for _ in 0..max_number_of_segments {
+                    v.push(SegmentState::new(number_of_responses))
+                }
+                v
+            },
             data_segment,
             connections: (0..client_list.capacity())
                 .map(|_| UnsafeCell::new(None))
@@ -284,7 +307,7 @@ impl<
             service_state: service.__internal_state().clone(),
             tagger: CyclicTagger::new(),
             loan_counter: IoxAtomicUsize::new(0),
-            unable_to_deliver_strategy: server_factory.unable_to_deliver_strategy,
+            unable_to_deliver_strategy: server_factory.config.unable_to_deliver_strategy,
             message_type_details: static_config.response_message_type_details.clone(),
             number_of_channels: number_of_requests_per_client,
         };
@@ -297,6 +320,7 @@ impl<
                 .request_response()
                 .enable_fire_and_forget_requests,
             shared_state: Arc::new(SharedServerState {
+                config: server_factory.config,
                 request_receiver,
                 client_list_state: UnsafeCell::new(unsafe { client_list.get_state() }),
                 server_handle: UnsafeCell::new(None),
@@ -327,6 +351,9 @@ impl<
                     server_port_id,
                     request_buffer_size: static_config.max_active_requests_per_client,
                     number_of_responses,
+                    max_slice_len: server_factory.config.initial_max_slice_len,
+                    data_segment_type,
+                    max_number_of_segments,
                 }) {
                 Some(v) => Some(v),
                 None => {
diff --git a/iceoryx2/src/service/dynamic_config/request_response.rs b/iceoryx2/src/service/dynamic_config/request_response.rs
@@ -33,6 +33,9 @@ pub struct ServerDetails {
     pub server_port_id: UniqueServerId,
     pub request_buffer_size: usize,
     pub number_of_responses: usize,
+    pub max_slice_len: usize,
+    pub data_segment_type: DataSegmentType,
+    pub max_number_of_segments: u8,
 }
 
 #[doc(hidden)]
diff --git a/iceoryx2/src/service/port_factory/server.rs b/iceoryx2/src/service/port_factory/server.rs
@@ -40,6 +40,13 @@ use iceoryx2_bb_elementary::zero_copy_send::ZeroCopySend;
 use iceoryx2_bb_log::{fail, warn};
 use iceoryx2_cal::shm_allocator::AllocationStrategy;
 
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct LocalServerConfig {
+    pub(crate) unable_to_deliver_strategy: UnableToDeliverStrategy,
+    pub(crate) initial_max_slice_len: usize,
+    pub(crate) allocation_strategy: AllocationStrategy,
+}
+
 /// Defines a failure that can occur when a [`Server`] is created with
 /// [`crate::service::port_factory::server::PortFactoryServer`].
 #[derive(Debug, PartialEq, Eq, Copy, Clone)]
@@ -79,8 +86,8 @@ pub struct PortFactoryServer<
         ResponseHeader,
     >,
 
+    pub(crate) config: LocalServerConfig,
     pub(crate) max_loaned_responses_per_request: usize,
-    pub(crate) unable_to_deliver_strategy: UnableToDeliverStrategy,
     pub(crate) request_degradation_callback: Option<DegradationCallback<'static>>,
     pub(crate) response_degradation_callback: Option<DegradationCallback<'static>>,
 }
@@ -122,7 +129,11 @@ impl<
         Self {
             factory,
             max_loaned_responses_per_request: defs.server_max_loaned_responses_per_request,
-            unable_to_deliver_strategy: defs.server_unable_to_deliver_strategy,
+            config: LocalServerConfig {
+                unable_to_deliver_strategy: defs.server_unable_to_deliver_strategy,
+                initial_max_slice_len: 1,
+                allocation_strategy: AllocationStrategy::Static,
+            },
             request_degradation_callback: None,
             response_degradation_callback: None,
         }
@@ -133,7 +144,7 @@ impl<
     /// [`Response`](crate::response::Response) since
     /// its internal buffer is full.
     pub fn unable_to_deliver_strategy(mut self, value: UnableToDeliverStrategy) -> Self {
-        self.unable_to_deliver_strategy = value;
+        self.config.unable_to_deliver_strategy = value;
         self
     }
 
@@ -218,10 +229,12 @@ impl<
     >
 {
     pub fn initial_max_slice_len(mut self, value: usize) -> Self {
-        todo!()
+        self.config.initial_max_slice_len = value;
+        self
     }
 
     pub fn allocation_strategy(mut self, value: AllocationStrategy) -> Self {
-        todo!()
+        self.config.allocation_strategy = value;
+        self
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -33,6 +33,9 @@ pub struct ServerDetails {`
`33`	`33`	`pub server_port_id: UniqueServerId,`
`34`	`34`	`pub request_buffer_size: usize,`
`35`	`35`	`pub number_of_responses: usize,`
	`36`	`+ pub max_slice_len: usize,`
	`37`	`+ pub data_segment_type: DataSegmentType,`
	`38`	`+ pub max_number_of_segments: u8,`
`36`	`39`	`}`
`37`	`40`
`38`	`41`	`#[doc(hidden)]`