Shm allocators (#2117)

* Implicit SHM optimization added

* rustfmt

* fix transport performance by optimizing it's layout

* optimize TransmissionPipelineProducer layout (that also optimizes layouts for transports)

* code format

* significantly improve TransmissionPipelineProducer layout (add 600k msgs\sec)

* add comment

* - fix bug with always-enabled SHM

* Add collection of shm allocators with various strong sides

* polish code

* spelling

* fix merge issues

* make tests for implicit SHM optimization

* review fixes

* go back to Duration in Waits

* Update Cargo.toml

* fallback to buddy_system_allocator 0.10.0 as Rust 1.75-friendly

* clippy fix

* taplo fix

* Fix port number in tests

* fix port conflict in tests

* Update unicast_shm.rs

* review fixes

Author: yellowhatter

Cargo.lock

@@ -89,6 +89,7 @@ async-std = { version = "1.6.5", features = ["tokio1"] }
 async-trait = "0.1.82"
 base64 = "0.22.1"
 bincode = "1.3.3"
+buddy_system_allocator = "=0.10.0"
 bytes = "1.7.1"
 clap = { version = "4.5.17", features = ["derive"] }
 console-subscriber = "0.4.0"
@@ -170,6 +171,7 @@ static_assertions = "1.1.0"
 static_init = "1.0.3"
 stop-token = "0.7.0"
 syn = "2.0"
+talc = { version = "4.4.3", default-features = false }
 test-case = "3.3.1"
 tide = "0.16.0"
 time = "0.3.36"

Cargo.toml

@@ -33,6 +33,7 @@ test = ["num_cpus"]
 
 [dependencies]
 async-trait = { workspace = true }
+buddy_system_allocator = { workspace = true }
 crossbeam-channel = { workspace = true }
 crossbeam-queue = { workspace = true }
 num-traits = { workspace = true }
@@ -41,6 +42,7 @@ rand = { workspace = true, features = ["std", "std_rng"] }
 stabby = { workspace = true }
 static_assertions = { workspace = true }
 static_init = { workspace = true }
+talc = { workspace = true }
 thread-priority = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }

commons/zenoh-shm/Cargo.toml

@@ -13,7 +13,12 @@
 //
 
 pub mod posix_shm_client;
+
 pub mod posix_shm_provider_backend;
+pub mod posix_shm_provider_backend_binary_heap;
+pub mod posix_shm_provider_backend_buddy;
+pub mod posix_shm_provider_backend_talc;
+
 pub mod protocol_id;
 
 pub(crate) mod posix_shm_segment;

commons/zenoh-shm/src/api/protocol_implementations/posix/mod.rs

@@ -12,289 +12,10 @@
 //   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
 //
 
-use std::{
-    cmp,
-    collections::BinaryHeap,
-    num::NonZeroUsize,
-    sync::{
-        atomic::{AtomicUsize, Ordering},
-        Arc, Mutex,
-    },
+use crate::api::protocol_implementations::posix::posix_shm_provider_backend_talc::{
+    PosixShmProviderBackendTalc, PosixShmProviderBackendTalcBuilder,
 };
 
-use zenoh_core::{zlock, Resolvable, Wait};
-use zenoh_result::ZResult;
+pub type PosixShmProviderBackendBuilder<What> = PosixShmProviderBackendTalcBuilder<What>;
 
-use super::posix_shm_segment::PosixShmSegment;
-use crate::api::{
-    common::{
-        types::{ChunkID, ProtocolID, PtrInSegment},
-        with_id::WithProtocolID,
-    },
-    protocol_implementations::posix::protocol_id::POSIX_PROTOCOL_ID,
-    provider::{
-        chunk::{AllocatedChunk, ChunkDescriptor},
-        memory_layout::{MemLayout, MemoryLayout, StaticLayout, TryIntoMemoryLayout},
-        shm_provider_backend::ShmProviderBackend,
-        types::{AllocAlignment, ChunkAllocResult, ZAllocError, ZLayoutError},
-    },
-};
-
-#[derive(Eq, Copy, Clone, Debug)]
-struct Chunk {
-    offset: ChunkID,
-    size: NonZeroUsize,
-}
-
-impl Ord for Chunk {
-    fn cmp(&self, other: &Self) -> cmp::Ordering {
-        self.size.cmp(&other.size)
-    }
-}
-
-impl PartialOrd for Chunk {
-    fn partial_cmp(&self, other: &Self) -> Option<cmp::Ordering> {
-        Some(self.cmp(other))
-    }
-}
-
-impl PartialEq for Chunk {
-    fn eq(&self, other: &Self) -> bool {
-        self.size == other.size
-    }
-}
-
-/// Builder to create posix SHM provider
-#[zenoh_macros::unstable_doc]
-pub struct PosixShmProviderBackendBuilder<Layout: MemLayout> {
-    layout: Layout,
-}
-
-#[zenoh_macros::unstable_doc]
-impl<Layout: TryIntoMemoryLayout> Resolvable for PosixShmProviderBackendBuilder<Layout> {
-    type To = ZResult<PosixShmProviderBackend>;
-}
-
-#[zenoh_macros::unstable_doc]
-impl<Layout: TryIntoMemoryLayout> Wait for PosixShmProviderBackendBuilder<Layout> {
-    fn wait(self) -> <Self as Resolvable>::To {
-        let layout: MemoryLayout = self.layout.try_into()?;
-        PosixShmProviderBackend::new(&layout)
-    }
-}
-
-#[zenoh_macros::unstable_doc]
-impl Resolvable for PosixShmProviderBackendBuilder<&MemoryLayout> {
-    type To = ZResult<PosixShmProviderBackend>;
-}
-
-#[zenoh_macros::unstable_doc]
-impl Wait for PosixShmProviderBackendBuilder<&MemoryLayout> {
-    fn wait(self) -> <Self as Resolvable>::To {
-        PosixShmProviderBackend::new(self.layout)
-    }
-}
-
-#[zenoh_macros::unstable_doc]
-impl<T> Resolvable for PosixShmProviderBackendBuilder<StaticLayout<T>> {
-    type To = ZResult<PosixShmProviderBackend>;
-}
-
-#[zenoh_macros::unstable_doc]
-impl<T> Wait for PosixShmProviderBackendBuilder<StaticLayout<T>> {
-    fn wait(self) -> <Self as Resolvable>::To {
-        PosixShmProviderBackend::new(&self.layout.into())
-    }
-}
-
-/// A backend for ShmProvider based on POSIX shared memory.
-/// This is the default general-purpose backed shipped with Zenoh.
-#[zenoh_macros::unstable_doc]
-pub struct PosixShmProviderBackend {
-    available: AtomicUsize,
-    segment: Arc<PosixShmSegment>,
-    free_list: Mutex<BinaryHeap<Chunk>>,
-    alignment: AllocAlignment,
-}
-
-impl PosixShmProviderBackend {
-    /// Get the builder to construct a new instance
-    #[zenoh_macros::unstable_doc]
-    pub fn builder<Layout: MemLayout>(layout: Layout) -> PosixShmProviderBackendBuilder<Layout> {
-        PosixShmProviderBackendBuilder { layout }
-    }
-
-    fn new(layout: &MemoryLayout) -> ZResult<Self> {
-        let segment = Arc::new(PosixShmSegment::create(layout.size())?);
-
-        // because of platform specific, our shm segment is >= requested size, so in order to utilize
-        // additional memory we re-layout the size
-        let real_size = segment.segment.elem_count().get();
-        let aligned_size = (real_size
-            - (real_size % layout.alignment().get_alignment_value().get()))
-        .try_into()?;
-
-        let mut free_list = BinaryHeap::new();
-        let root_chunk = Chunk {
-            offset: 0,
-            size: aligned_size,
-        };
-        free_list.push(root_chunk);
-
-        tracing::trace!(
-            "Created PosixShmProviderBackend id {}, layout {:?}, aligned size {aligned_size}",
-            segment.segment.id(),
-            layout
-        );
-
-        Ok(Self {
-            available: AtomicUsize::new(aligned_size.get()),
-            segment,
-            free_list: Mutex::new(free_list),
-            alignment: layout.alignment(),
-        })
-    }
-}
-
-impl WithProtocolID for PosixShmProviderBackend {
-    fn id(&self) -> ProtocolID {
-        POSIX_PROTOCOL_ID
-    }
-}
-
-impl ShmProviderBackend for PosixShmProviderBackend {
-    fn alloc(&self, layout: &MemoryLayout) -> ChunkAllocResult {
-        tracing::trace!("PosixShmProviderBackend::alloc({:?})", layout);
-
-        let required_len = layout.size();
-
-        if self.available.load(Ordering::Relaxed) < required_len.get() {
-            tracing::trace!( "PosixShmProviderBackend does not have sufficient free memory to allocate {:?}, try de-fragmenting!", layout);
-            return Err(ZAllocError::OutOfMemory);
-        }
-
-        let mut guard = zlock!(self.free_list);
-        // The strategy taken is the same for some Unix System V implementations -- as described in the
-        // famous Bach's book --  in essence keep an ordered list of free slot and always look for the
-        // biggest as that will give the biggest left-over.
-        match guard.pop() {
-            Some(mut chunk) if chunk.size >= required_len => {
-                // NOTE: don't loose any chunks here, as it will lead to memory leak
-                tracing::trace!("Allocator selected Chunk ({:?})", &chunk);
-
-                if chunk.size > required_len {
-                    let free_chunk = Chunk {
-                        offset: chunk.offset + required_len.get() as ChunkID,
-                        // SAFETY: this is safe because we always operate on a leftover, which is checked above!
-                        size: unsafe {
-                            NonZeroUsize::new_unchecked(chunk.size.get() - required_len.get())
-                        },
-                    };
-                    tracing::trace!("The allocation will leave a Free Chunk: {:?}", &free_chunk);
-                    guard.push(free_chunk);
-                    chunk.size = required_len;
-                }
-
-                self.available
-                    .fetch_sub(chunk.size.get(), Ordering::Relaxed);
-
-                let descriptor =
-                    ChunkDescriptor::new(self.segment.segment.id(), chunk.offset, chunk.size);
-
-                let data = PtrInSegment::new(
-                    unsafe { self.segment.segment.elem_mut(chunk.offset) },
-                    self.segment.clone(),
-                );
-
-                Ok(AllocatedChunk { descriptor, data })
-            }
-            Some(c) => {
-                tracing::trace!("PosixShmProviderBackend::alloc({:?}) cannot find any big enough chunk\nShmManager::free_list = {:?}", layout, self.free_list);
-                guard.push(c);
-                Err(ZAllocError::NeedDefragment)
-            }
-            None => {
-                // NOTE: that should never happen! If this happens - there is a critical bug somewhere around!
-                let err = format!("PosixShmProviderBackend::alloc({:?}) cannot find any available chunk\nShmManager::free_list = {:?}", layout, self.free_list);
-                #[cfg(feature = "test")]
-                panic!("{err}");
-                #[cfg(not(feature = "test"))]
-                {
-                    tracing::error!("{err}");
-                    Err(ZAllocError::OutOfMemory)
-                }
-            }
-        }
-    }
-
-    fn free(&self, chunk: &ChunkDescriptor) {
-        let free_chunk = Chunk {
-            offset: chunk.chunk,
-            size: chunk.len,
-        };
-        self.available
-            .fetch_add(free_chunk.size.get(), Ordering::Relaxed);
-        zlock!(self.free_list).push(free_chunk);
-    }
-
-    fn defragment(&self) -> usize {
-        fn try_merge_adjacent_chunks(a: &Chunk, b: &Chunk) -> Option<Chunk> {
-            let end_offset = a.offset as usize + a.size.get();
-            if end_offset == b.offset as usize {
-                Some(Chunk {
-                    // SAFETY: this is safe because we operate on non-zero sizes and it will never overflow
-                    size: unsafe { NonZeroUsize::new_unchecked(a.size.get() + b.size.get()) },
-                    offset: a.offset,
-                })
-            } else {
-                None
-            }
-        }
-
-        let mut largest = 0usize;
-
-        // TODO: optimize this!
-        // this is an old legacy algo for merging adjacent chunks
-        // we extract chunks to separate container, sort them by offset and then check each chunk for
-        // adjacence with neighbour. Adjacent chunks are joined and returned back to temporary container.
-        // If chunk is not adjacent with it's neighbour, it is placed back to self.free_list
-        let mut guard = zlock!(self.free_list);
-        if guard.len() > 1 {
-            let mut fbs: Vec<Chunk> = guard.drain().collect();
-            fbs.sort_by(|x, y| x.offset.cmp(&y.offset));
-            let mut current = fbs.remove(0);
-            let mut i = 0;
-            let n = fbs.len();
-            for chunk in fbs.iter() {
-                i += 1;
-                let next = *chunk;
-                match try_merge_adjacent_chunks(&current, &next) {
-                    Some(c) => {
-                        current = c;
-                        largest = largest.max(current.size.get());
-                        if i == n {
-                            guard.push(current)
-                        }
-                    }
-                    None => {
-                        guard.push(current);
-                        if i == n {
-                            guard.push(next);
-                        } else {
-                            current = next;
-                        }
-                    }
-                }
-            }
-        }
-        largest
-    }
-
-    fn available(&self) -> usize {
-        self.available.load(Ordering::Relaxed)
-    }
-
-    fn layout_for(&self, layout: MemoryLayout) -> Result<MemoryLayout, ZLayoutError> {
-        layout.extend(self.alignment)
-    }
-}
+pub type PosixShmProviderBackend = PosixShmProviderBackendTalc;