RFC: IpcSharedMemoryVec an appendable data structure which can be shared via a reader

Cargo.toml

@@ -41,7 +41,7 @@ fnv = "1.0.3"
 futures-channel = { version = "0.3.31", optional = true }
 futures-core = { version = "0.3.31", optional = true }
 libc = "0.2.162"
-serde_core = "1.0"
+serde = { version = "1.0", features = ["derive"] }
 uuid = { version = "1", features = ["v4"] }
 
 [target.'cfg(any(target_os = "linux", target_os = "openbsd", target_os = "freebsd", target_os = "illumos"))'.dependencies]

src/ipc.rs

@@ -7,13 +7,16 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
-use crate::platform::{self, OsIpcChannel, OsIpcReceiver, OsIpcReceiverSet, OsIpcSender};
+use crate::platform::{
+    self, OsIpcChannel, OsIpcReceiver, OsIpcReceiverSet, OsIpcSender, OsIpcSharedMemoryIndex,
+    OsIpcSharedMemoryVec,
+};
 use crate::platform::{
     OsIpcOneShotServer, OsIpcSelectionResult, OsIpcSharedMemory, OsOpaqueIpcChannel,
 };
 
 use bincode;
-use serde_core::{de::Error, Deserialize, Deserializer, Serialize, Serializer};
+use serde::{de::Error, Deserialize, Deserializer, Serialize, Serializer};
 use std::cell::RefCell;
 use std::cmp::min;
 use std::error::Error as StdError;
@@ -33,6 +36,12 @@
     static OS_IPC_CHANNELS_FOR_SERIALIZATION: RefCell<Vec<OsIpcChannel>> = const { RefCell::new(Vec::new()) };
 
     static OS_IPC_SHARED_MEMORY_REGIONS_FOR_SERIALIZATION: RefCell<Vec<OsIpcSharedMemory>> =
+        const { RefCell::new(Vec::new()) };
+
+    static OS_IPC_SHARED_MEMORY_VEC_REGIONS_FOR_DESERIALIZATION:
+        RefCell<Vec<Option<OsIpcSharedMemoryVec>>> = const { RefCell::new(Vec::new()) };
+
+            static OS_IPC_SHARED_MEMORY_VEC_REGIONS_FOR_SERIALIZATION: RefCell<Vec<OsIpcSharedMemoryVec>> =
         const { RefCell::new(Vec::new()) }
 }
 
@@ -368,15 +377,17 @@
         OS_IPC_CHANNELS_FOR_SERIALIZATION.with(|os_ipc_channels_for_serialization| {
             OS_IPC_SHARED_MEMORY_REGIONS_FOR_SERIALIZATION.with(
                 |os_ipc_shared_memory_regions_for_serialization| {
-                    bincode::serialize_into(&mut bytes, &data)?;
-                    let os_ipc_channels = os_ipc_channels_for_serialization.take();
-                    let os_ipc_shared_memory_regions =
-                        os_ipc_shared_memory_regions_for_serialization.take();
-                    Ok(self.os_sender.send(
-                        &bytes[..],
-                        os_ipc_channels,
-                        os_ipc_shared_memory_regions,
-                    )?)
+                    OS_IPC_SHARED_MEMORY_VEC_REGIONS_FOR_SERIALIZATION.with(
+                        |os_ipc_shared_memory_vec_regions_for_serialization| {
+                            bincode::serialize_into(&mut bytes, &data)?;
+                            Ok(self.os_sender.send(
+                                &bytes[..],
+                                os_ipc_channels_for_serialization.take(),
+                                os_ipc_shared_memory_regions_for_serialization.take(),
+                                os_ipc_shared_memory_vec_regions_for_serialization.take(),
+                            )?)
+                        },
+                    )
                 },
             )
         })
@@ -631,6 +642,117 @@
     }
 }
 
+/// An index to access `IpcSharedMemoryVec`
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct IpcSharedMemoryIndex(OsIpcSharedMemoryIndex);
+
+/// Shared memory vector that will can be made accessible to a receiver via the reader method
+/// of an IPC message that contains the discriptor.
+///
+/// # Examples
+/// ```
+/// # use ipc_channel::ipc::{self, IpcSharedMemoryVec};
+/// # let (tx, rx) = ipc::channel().unwrap();
+/// # let data = [0x76, 0x69, 0x6d, 0x00];
+/// let (shmem, index) = IpcSharedMemoryVec::from_bytes(&data);
+/// tx.send(shmem.reader()).unwrap();
+/// # let rx_shmem = rx.recv().unwrap();
+/// # assert_eq!(shmem.get(&index), rx_shmem.get(&index));
+/// ```
+#[derive(Clone)]
+pub struct IpcSharedMemoryVec {
+    os_shared_memory_vec: OsIpcSharedMemoryVec,
+}
+
+unsafe impl Send for IpcSharedMemoryVec {}
+unsafe impl Sync for IpcSharedMemoryVec {}
+
+impl Default for IpcSharedMemoryVec {
+    fn default() -> Self {
+        let (memory, _index) = OsIpcSharedMemoryVec::from_bytes(&[0xab]);
+        IpcSharedMemoryVec {
+            os_shared_memory_vec: memory,
+        }
+    }
+}
+
+impl IpcSharedMemoryVec {
+    pub fn from_bytes(bytes: &[u8]) -> (IpcSharedMemoryVec, IpcSharedMemoryIndex) {
+        let (memory, index) = OsIpcSharedMemoryVec::from_bytes(bytes);
+        (
+            IpcSharedMemoryVec {
+                os_shared_memory_vec: memory,
+            },
+            IpcSharedMemoryIndex(index),
+        )
+    }
+
+    pub fn push(&mut self, bytes: &[u8]) -> IpcSharedMemoryIndex {
+        IpcSharedMemoryIndex(self.os_shared_memory_vec.push(bytes))
+    }
+
+    pub fn get(&self, index: &IpcSharedMemoryIndex) -> Option<&[u8]> {
+        self.os_shared_memory_vec.get(&index.0)
+    }
+
+    /// Gives you a reader to access the already stored memory locations.
+    /// Notice this will not allow you to access memory added _after_ you
+    /// got this reader.
+    pub fn reader(&self) -> IpcSharedMemoryReader {
+        IpcSharedMemoryReader(self.os_shared_memory_vec.clone())
+    }
+}
+
+#[derive(Clone)]
+pub struct IpcSharedMemoryReader(OsIpcSharedMemoryVec);
+
+impl IpcSharedMemoryReader {
+    pub fn get(&self, index: &IpcSharedMemoryIndex) -> Option<&[u8]> {
+        self.0.get(&index.0)
+    }
+}
+
+impl<'de> Deserialize<'de> for IpcSharedMemoryReader {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let index: usize = Deserialize::deserialize(deserializer)?;
+
+        let os_shared_memory_vec = OS_IPC_SHARED_MEMORY_VEC_REGIONS_FOR_DESERIALIZATION.with(
+            |os_ipc_shared_memory_regions_for_deserialization| {
+                let mut regions =  os_ipc_shared_memory_regions_for_deserialization.borrow_mut();
+                let Some(region) = regions.get_mut(index) else {
+                    return Err(format!("Cannot consume shared memory region at index {index}, there are only {} regions available", regions.len()));
+                };
+
+                region.take().ok_or_else(|| format!("Shared memory region {index} has already been consumed"))
+            },
+        ).map_err(D::Error::custom)?;
+
+        Ok(IpcSharedMemoryReader(os_shared_memory_vec))
+    }
+}
+
+impl Serialize for IpcSharedMemoryReader {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        let index = OS_IPC_SHARED_MEMORY_VEC_REGIONS_FOR_SERIALIZATION.with(
+            |os_ipc_shared_memory_vec_regions_for_serialization| {
+                let mut os_ipc_shared_memory_vec_regions_for_serialization =
+                    os_ipc_shared_memory_vec_regions_for_serialization.borrow_mut();
+                let index = os_ipc_shared_memory_vec_regions_for_serialization.len();
+                os_ipc_shared_memory_vec_regions_for_serialization.push(self.0.clone());
+                index
+            },
+        );
+        debug_assert!(index < usize::MAX);
+        index.serialize(serializer)
+    }
+}
+
 /// Result for readable events returned from [IpcReceiverSet::select].
 ///
 /// [IpcReceiverSet::select]: struct.IpcReceiverSet.html#method.select
@@ -674,6 +796,7 @@
     pub(crate) data: Vec<u8>,
     pub(crate) os_ipc_channels: Vec<OsOpaqueIpcChannel>,
     pub(crate) os_ipc_shared_memory_regions: Vec<OsIpcSharedMemory>,
+    pub(crate) os_ipc_shared_memory_vec: Vec<OsIpcSharedMemoryVec>,
 }
 
 impl IpcMessage {
@@ -684,6 +807,7 @@
             data,
             os_ipc_channels: vec![],
             os_ipc_shared_memory_regions: vec![],
+            os_ipc_shared_memory_vec: vec![],
         }
     }
 }
@@ -702,11 +826,13 @@
         data: Vec<u8>,
         os_ipc_channels: Vec<OsOpaqueIpcChannel>,
         os_ipc_shared_memory_regions: Vec<OsIpcSharedMemory>,
+        os_ipc_shared_memory_vec: Vec<OsIpcSharedMemoryVec>,
     ) -> IpcMessage {
         IpcMessage {
             data,
             os_ipc_channels,
             os_ipc_shared_memory_regions,
+            os_ipc_shared_memory_vec,
         }
     }
 
@@ -718,23 +844,34 @@
         OS_IPC_CHANNELS_FOR_DESERIALIZATION.with(|os_ipc_channels_for_deserialization| {
             OS_IPC_SHARED_MEMORY_REGIONS_FOR_DESERIALIZATION.with(
                 |os_ipc_shared_memory_regions_for_deserialization| {
-                    // Setup the thread local memory for deserialization to take it.
-                    *os_ipc_channels_for_deserialization.borrow_mut() = self.os_ipc_channels;
-                    *os_ipc_shared_memory_regions_for_deserialization.borrow_mut() = self
-                        .os_ipc_shared_memory_regions
-                        .into_iter()
-                        .map(Some)
-                        .collect();
-
-                    let result = bincode::deserialize(&self.data[..]);
-
-                    // Clear the shared memory
-                    let _ = os_ipc_shared_memory_regions_for_deserialization.take();
-                    let _ = os_ipc_channels_for_deserialization.take();
-
-                    /* Error check comes after doing cleanup,
-                     * since we need the cleanup both in the success and the error cases. */
-                    result
+                    OS_IPC_SHARED_MEMORY_VEC_REGIONS_FOR_DESERIALIZATION.with(
+                        |os_ipc_shared_memory_vec_regions_for_deserialization| {
+                            // Setup the thread local memory for deserialization to take it.
+                            *os_ipc_channels_for_deserialization.borrow_mut() =
+                                self.os_ipc_channels;
+                            *os_ipc_shared_memory_regions_for_deserialization.borrow_mut() = self
+                                .os_ipc_shared_memory_regions
+                                .into_iter()
+                                .map(Some)
+                                .collect();
+                            *os_ipc_shared_memory_vec_regions_for_deserialization.borrow_mut() =
+                                self.os_ipc_shared_memory_vec
+                                    .into_iter()
+                                    .map(Some)
+                                    .collect();
+
+                            let result = bincode::deserialize(&self.data[..]);
+
+                            // Clear the shared memory
+                            let _ = os_ipc_shared_memory_regions_for_deserialization.take();
+                            let _ = os_ipc_channels_for_deserialization.take();
+                            let _ = os_ipc_shared_memory_vec_regions_for_deserialization.take();
+
+                            /* Error check comes after doing cleanup,
+                             * since we need the cleanup both in the success and the error cases. */
+                            result
+                        },
+                    )
                 },
             )
         })
@@ -960,7 +1097,7 @@
     #[inline]
     pub fn send(&self, data: &[u8]) -> Result<(), io::Error> {
         self.os_sender
-            .send(data, vec![], vec![])
+            .send(data, vec![], vec![], vec![])
             .map_err(io::Error::from)
     }
 }

src/platform/inprocess/mod.rs

@@ -153,9 +153,15 @@
         data: &[u8],
         ports: Vec<OsIpcChannel>,
         shared_memory_regions: Vec<OsIpcSharedMemory>,
+        shared_memory_vecs: Vec<OsIpcSharedMemoryVec>,
     ) -> Result<(), ChannelError> {
         let os_ipc_channels = ports.into_iter().map(OsOpaqueIpcChannel::new).collect();
-        let ipc_message = IpcMessage::new(data.to_vec(), os_ipc_channels, shared_memory_regions);
+        let ipc_message = IpcMessage::new(
+            data.to_vec(),
+            os_ipc_channels,
+            shared_memory_regions,
+            shared_memory_vecs,
+        );
         self.sender
             .send(ChannelMessage(ipc_message))
             .map_err(|_| ChannelError::BrokenPipeError)

src/platform/mod.rs

@@ -66,6 +66,7 @@
 pub use self::os::{channel, OsOpaqueIpcChannel};
 pub use self::os::{OsIpcChannel, OsIpcOneShotServer, OsIpcReceiver, OsIpcReceiverSet};
 pub use self::os::{OsIpcSelectionResult, OsIpcSender, OsIpcSharedMemory};
+pub use self::os::{OsIpcSharedMemoryIndex, OsIpcSharedMemoryVec};
 
 #[cfg(test)]
 mod test;