zenoh_shm/api/provider/
shm_provider.rs

1//
2// Copyright (c) 2023 ZettaScale Technology
3//
4// This program and the accompanying materials are made available under the
5// terms of the Eclipse Public License 2.0 which is available at
6// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
7// which is available at https://www.apache.org/licenses/LICENSE-2.0.
8//
9// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
10//
11// Contributors:
12//   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
13//
14
15use std::{
16    error::Error,
17    future::{Future, IntoFuture},
18    marker::PhantomData,
19    mem::MaybeUninit,
20    num::NonZeroUsize,
21    pin::Pin,
22    sync::{atomic::Ordering, Mutex},
23    time::Duration,
24};
25
26use async_trait::async_trait;
27use zenoh_core::{zlock, zresult::ZResult, Resolvable, Wait};
28
29use super::{
30    chunk::{AllocatedChunk, ChunkDescriptor},
31    shm_provider_backend::ShmProviderBackend,
32    types::{ChunkAllocResult, ZAllocError, ZLayoutAllocError, ZLayoutError},
33};
34use crate::{
35    api::{
36        buffer::{typed::Typed, zshmmut::ZShmMut},
37        protocol_implementations::posix::posix_shm_provider_backend::{
38            PosixShmProviderBackend, PosixShmProviderBackendBuilder,
39        },
40        provider::memory_layout::{MemoryLayout, TypedLayout},
41    },
42    metadata::{
43        allocated_descriptor::AllocatedMetadataDescriptor, descriptor::MetadataDescriptor,
44        storage::GLOBAL_METADATA_STORAGE,
45    },
46    watchdog::{
47        confirmator::{ConfirmedDescriptor, GLOBAL_CONFIRMATOR},
48        validator::GLOBAL_VALIDATOR,
49    },
50    ShmBufInfo, ShmBufInner,
51};
52
53/// The type of allocation.
54pub trait AllocLayout {
55    /// The buffer returned by the allocation.
56    type Buffer;
57    /// Returns the memory layouts of the allocation.
58    fn memory_layout(self) -> Result<MemoryLayout, ZLayoutError>;
59    /// Wraps the raw allocated buffer.
60    ///
61    /// # Safety
62    ///
63    /// The buffer must have the layout returned by [`Self::memory_layout`]
64    unsafe fn wrap_buffer(buffer: ZShmMut) -> Self::Buffer;
65}
66
67impl<T: TryInto<MemoryLayout>> AllocLayout for T
68where
69    T::Error: Into<ZLayoutError>,
70{
71    type Buffer = ZShmMut;
72
73    fn memory_layout(self) -> Result<MemoryLayout, ZLayoutError> {
74        self.try_into().map_err(Into::into)
75    }
76
77    /// # Safety
78    ///
79    /// The buffer must have the layout returned by `memory_layout`
80    unsafe fn wrap_buffer(buffer: ZShmMut) -> Self::Buffer {
81        buffer
82    }
83}
84
85impl<T> AllocLayout for TypedLayout<T> {
86    type Buffer = Typed<MaybeUninit<T>, ZShmMut>;
87
88    fn memory_layout(self) -> Result<MemoryLayout, ZLayoutError> {
89        Ok(MemoryLayout::for_type::<T>())
90    }
91
92    /// # Safety
93    ///
94    /// The buffer must have the layout returned by `memory_layout`
95    unsafe fn wrap_buffer(buffer: ZShmMut) -> Self::Buffer {
96        // SAFETY: same precondition
97        unsafe { Typed::new_unchecked(buffer) }
98    }
99}
100
101/// A layout for allocations.
102///
103/// This is a pre-calculated layout suitable for making series of similar allocations
104/// adopted for particular ShmProvider
105#[zenoh_macros::unstable_doc]
106#[derive(Debug)]
107pub struct PrecomputedLayout<'a, Backend, Layout> {
108    size: NonZeroUsize,
109    provider_layout: MemoryLayout,
110    provider: &'a ShmProvider<Backend>,
111    _phantom: PhantomData<Layout>,
112}
113
114impl<'a, Backend, Layout> PrecomputedLayout<'a, Backend, Layout>
115where
116    Backend: ShmProviderBackend,
117{
118    /// Allocate the new buffer with this layout
119    #[zenoh_macros::unstable_doc]
120    pub fn alloc<'b>(&'b self) -> PrecomputedAllocBuilder<'b, 'a, Backend, Layout> {
121        PrecomputedAllocBuilder {
122            layout: self,
123            policy: JustAlloc,
124        }
125    }
126}
127
128impl<'a, Backend, Layout> PrecomputedLayout<'a, Backend, Layout>
129where
130    Backend: ShmProviderBackend,
131{
132    fn new(
133        provider: &'a ShmProvider<Backend>,
134        required_layout: MemoryLayout,
135    ) -> Result<Self, ZLayoutError> {
136        // NOTE: Depending on internal implementation, provider's backend might relayout
137        // the allocations for bigger alignment (ex. 4-byte aligned allocation to 8-bytes aligned)
138
139        // calculate required layout size
140        let size = required_layout.size();
141
142        // Obtain provider's layout for our layout
143        let provider_layout = provider
144            .backend
145            .layout_for(required_layout)
146            .map_err(|_| ZLayoutError::ProviderIncompatibleLayout)?;
147
148        Ok(Self {
149            size,
150            provider_layout,
151            provider,
152            _phantom: PhantomData,
153        })
154    }
155}
156
157/// Trait for allocation policies
158#[zenoh_macros::unstable_doc]
159pub trait AllocPolicy<Backend> {
160    fn alloc(&self, layout: &MemoryLayout, provider: &ShmProvider<Backend>) -> ChunkAllocResult;
161}
162
163/// Trait for async allocation policies
164#[zenoh_macros::unstable_doc]
165#[async_trait]
166pub trait AsyncAllocPolicy<Backend>: Send {
167    async fn alloc_async(
168        &self,
169        layout: &MemoryLayout,
170        provider: &ShmProvider<Backend>,
171    ) -> ChunkAllocResult;
172}
173
174/// Marker trait for policies that are safe to use
175///
176/// # Safety
177///
178/// Policies implementing this trait must be safe to use. For example, [`Deallocate`] is not safe
179/// as it may deallocate and reuse a buffer that is currently in use. Users of unsafe policy must
180/// enforce the safety of their code by other means.
181pub unsafe trait SafePolicy {}
182
183/// Marker trait for compile-time allocation policies
184#[zenoh_macros::unstable_doc]
185pub trait ConstPolicy {
186    const NEW: Self;
187}
188
189/// A policy value that can be either constant or runtime provided.
190pub trait PolicyValue<T> {
191    fn get(&self) -> T;
192}
193
194impl<T: Copy> PolicyValue<T> for T {
195    fn get(&self) -> T {
196        *self
197    }
198}
199/// A const `bool`.
200#[derive(Debug)]
201pub struct ConstBool<const VALUE: bool>;
202impl<const VALUE: bool> ConstPolicy for ConstBool<VALUE> {
203    const NEW: Self = Self;
204}
205impl<const VALUE: bool> PolicyValue<bool> for ConstBool<VALUE> {
206    fn get(&self) -> bool {
207        VALUE
208    }
209}
210/// A const `usize`.
211#[derive(Debug)]
212pub struct ConstUsize<const VALUE: usize>;
213impl<const VALUE: usize> ConstPolicy for ConstUsize<VALUE> {
214    const NEW: Self = Self;
215}
216impl<const VALUE: usize> PolicyValue<usize> for ConstUsize<VALUE> {
217    fn get(&self) -> usize {
218        VALUE
219    }
220}
221
222/// Just try to allocate
223#[zenoh_macros::unstable_doc]
224#[derive(Debug, Clone, Copy)]
225pub struct JustAlloc;
226impl<Backend> AllocPolicy<Backend> for JustAlloc
227where
228    Backend: ShmProviderBackend,
229{
230    fn alloc(&self, layout: &MemoryLayout, provider: &ShmProvider<Backend>) -> ChunkAllocResult {
231        provider.backend.alloc(layout)
232    }
233}
234unsafe impl SafePolicy for JustAlloc {}
235impl ConstPolicy for JustAlloc {
236    const NEW: Self = Self;
237}
238
239/// Garbage collection policy.
240///
241/// Try to reclaim old buffers if allocation failed and allocate again
242/// if the largest reclaimed chuk is not smaller than the one required.
243///
244/// If `Safe` is set to false, the policy may lead to reallocate memory currently in-use.
245#[zenoh_macros::unstable_doc]
246#[derive(Debug, Clone, Copy)]
247pub struct GarbageCollect<InnerPolicy = JustAlloc, AltPolicy = JustAlloc, Safe = ConstBool<true>> {
248    inner_policy: InnerPolicy,
249    alt_policy: AltPolicy,
250    safe: Safe,
251}
252impl<InnerPolicy, AltPolicy, Safe> GarbageCollect<InnerPolicy, AltPolicy, Safe> {
253    /// Creates a new `GarbageCollect` policy.
254    pub fn new(inner_policy: InnerPolicy, alt_policy: AltPolicy, safe: Safe) -> Self {
255        Self {
256            inner_policy,
257            alt_policy,
258            safe,
259        }
260    }
261}
262impl<Backend, InnerPolicy, AltPolicy, Safe> AllocPolicy<Backend>
263    for GarbageCollect<InnerPolicy, AltPolicy, Safe>
264where
265    Backend: ShmProviderBackend,
266    InnerPolicy: AllocPolicy<Backend>,
267    AltPolicy: AllocPolicy<Backend>,
268    Safe: PolicyValue<bool>,
269{
270    fn alloc(&self, layout: &MemoryLayout, provider: &ShmProvider<Backend>) -> ChunkAllocResult {
271        let result = self.inner_policy.alloc(layout, provider);
272        if result.is_err() {
273            // try to alloc again only if GC managed to reclaim big enough chunk
274            let collected = if self.safe.get() {
275                provider.garbage_collect()
276            } else {
277                // SAFETY: the policy doesn't implement `SafePolicy` in this case, so the user
278                // must enforce the safety himself
279                unsafe { provider.garbage_collect_unsafe() }
280            };
281            if collected >= layout.size().get() {
282                return self.alt_policy.alloc(layout, provider);
283            }
284        }
285        result
286    }
287}
288unsafe impl<InnerPolicy: SafePolicy, AltPolicy: SafePolicy> SafePolicy
289    for GarbageCollect<InnerPolicy, AltPolicy, ConstBool<true>>
290{
291}
292impl<InnerPolicy: ConstPolicy, AltPolicy: ConstPolicy, Safe: ConstPolicy> ConstPolicy
293    for GarbageCollect<InnerPolicy, AltPolicy, Safe>
294{
295    const NEW: Self = Self {
296        inner_policy: InnerPolicy::NEW,
297        alt_policy: AltPolicy::NEW,
298        safe: Safe::NEW,
299    };
300}
301
302/// Defragmenting policy.
303///
304/// Try to defragment if allocation failed and allocate again
305/// if the largest defragmented chuk is not smaller than the one required
306#[zenoh_macros::unstable_doc]
307#[derive(Debug, Clone, Copy)]
308pub struct Defragment<InnerPolicy = JustAlloc, AltPolicy = JustAlloc> {
309    inner_policy: InnerPolicy,
310    alt_policy: AltPolicy,
311}
312impl<InnerPolicy, AltPolicy> Defragment<InnerPolicy, AltPolicy> {
313    /// Creates a new `Defragment` policy.
314    pub fn new(inner_policy: InnerPolicy, alt_policy: AltPolicy) -> Self {
315        Self {
316            inner_policy,
317            alt_policy,
318        }
319    }
320}
321impl<Backend, InnerPolicy, AltPolicy> AllocPolicy<Backend> for Defragment<InnerPolicy, AltPolicy>
322where
323    Backend: ShmProviderBackend,
324    InnerPolicy: AllocPolicy<Backend>,
325    AltPolicy: AllocPolicy<Backend>,
326{
327    fn alloc(&self, layout: &MemoryLayout, provider: &ShmProvider<Backend>) -> ChunkAllocResult {
328        let result = self.inner_policy.alloc(layout, provider);
329        if let Err(ZAllocError::NeedDefragment) = result {
330            // try to alloc again only if big enough chunk was defragmented
331            if provider.defragment() >= layout.size().get() {
332                return self.alt_policy.alloc(layout, provider);
333            }
334        }
335        result
336    }
337}
338unsafe impl<InnerPolicy: SafePolicy, AltPolicy: SafePolicy> SafePolicy
339    for Defragment<InnerPolicy, AltPolicy>
340{
341}
342impl<InnerPolicy: ConstPolicy, AltPolicy: ConstPolicy> ConstPolicy
343    for Defragment<InnerPolicy, AltPolicy>
344{
345    const NEW: Self = Self {
346        inner_policy: InnerPolicy::NEW,
347        alt_policy: AltPolicy::NEW,
348    };
349}
350
351/// Deallocating policy.
352///
353/// Forcibly deallocate up to N buffers until allocation succeeds.
354///
355/// This policy is unsafe as it may lead to reallocate memory currently in-use.
356#[zenoh_macros::unstable_doc]
357#[derive(Debug, Clone, Copy)]
358pub struct Deallocate<Limit, InnerPolicy = JustAlloc, AltPolicy = InnerPolicy> {
359    limit: Limit,
360    inner_policy: InnerPolicy,
361    alt_policy: AltPolicy,
362}
363impl<Limit, InnerPolicy, AltPolicy> Deallocate<Limit, InnerPolicy, AltPolicy> {
364    /// Creates a new `Deallocate` policy.
365    pub fn new(limit: Limit, inner_policy: InnerPolicy, alt_policy: AltPolicy) -> Self {
366        Self {
367            limit,
368            inner_policy,
369            alt_policy,
370        }
371    }
372}
373impl<Backend, Limit, InnerPolicy, AltPolicy> AllocPolicy<Backend>
374    for Deallocate<Limit, InnerPolicy, AltPolicy>
375where
376    Backend: ShmProviderBackend,
377    Limit: PolicyValue<usize>,
378    InnerPolicy: AllocPolicy<Backend>,
379    AltPolicy: AllocPolicy<Backend>,
380{
381    fn alloc(&self, layout: &MemoryLayout, provider: &ShmProvider<Backend>) -> ChunkAllocResult {
382        for _ in 0..self.limit.get() {
383            match self.inner_policy.alloc(layout, provider) {
384                res @ (Err(ZAllocError::NeedDefragment) | Err(ZAllocError::OutOfMemory)) => {
385                    let Some(chunk) = zlock!(provider.busy_list).pop() else {
386                        return res;
387                    };
388                    provider.backend.free(&chunk.descriptor());
389                }
390                res => return res,
391            }
392        }
393        self.alt_policy.alloc(layout, provider)
394    }
395}
396impl<Limit: ConstPolicy, InnerPolicy: ConstPolicy, AltPolicy: ConstPolicy> ConstPolicy
397    for Deallocate<Limit, InnerPolicy, AltPolicy>
398{
399    const NEW: Self = Self {
400        limit: Limit::NEW,
401        inner_policy: InnerPolicy::NEW,
402        alt_policy: AltPolicy::NEW,
403    };
404}
405
406/// Blocking allocation policy.
407///
408/// This policy will block until the allocation succeeds.
409/// Both sync and async modes available.
410#[zenoh_macros::unstable_doc]
411#[derive(Debug, Clone, Copy)]
412pub struct BlockOn<InnerPolicy = JustAlloc> {
413    inner_policy: InnerPolicy,
414}
415impl<InnerPolicy> BlockOn<InnerPolicy> {
416    /// Creates a new `BlockOn` policy.
417    pub fn new(inner_policy: InnerPolicy) -> Self {
418        Self { inner_policy }
419    }
420}
421#[async_trait]
422impl<Backend, InnerPolicy> AsyncAllocPolicy<Backend> for BlockOn<InnerPolicy>
423where
424    Backend: ShmProviderBackend + Sync,
425    InnerPolicy: AllocPolicy<Backend> + Send + Sync,
426{
427    async fn alloc_async(
428        &self,
429        layout: &MemoryLayout,
430        provider: &ShmProvider<Backend>,
431    ) -> ChunkAllocResult {
432        loop {
433            match self.inner_policy.alloc(layout, provider) {
434                Err(ZAllocError::NeedDefragment) | Err(ZAllocError::OutOfMemory) => {
435                    // TODO: implement provider's async signalling instead of this!
436                    tokio::time::sleep(Duration::from_millis(1)).await;
437                }
438                other_result => {
439                    return other_result;
440                }
441            }
442        }
443    }
444}
445unsafe impl<InnerPolicy: SafePolicy> SafePolicy for BlockOn<InnerPolicy> {}
446impl<Backend, InnerPolicy> AllocPolicy<Backend> for BlockOn<InnerPolicy>
447where
448    Backend: ShmProviderBackend,
449    InnerPolicy: AllocPolicy<Backend>,
450{
451    fn alloc(&self, layout: &MemoryLayout, provider: &ShmProvider<Backend>) -> ChunkAllocResult {
452        loop {
453            match self.inner_policy.alloc(layout, provider) {
454                Err(ZAllocError::NeedDefragment) | Err(ZAllocError::OutOfMemory) => {
455                    // TODO: implement provider's async signalling instead of this!
456                    std::thread::sleep(Duration::from_millis(1));
457                }
458                other_result => {
459                    return other_result;
460                }
461            }
462        }
463    }
464}
465impl<InnerPolicy: ConstPolicy> ConstPolicy for BlockOn<InnerPolicy> {
466    const NEW: Self = Self {
467        inner_policy: InnerPolicy::NEW,
468    };
469}
470
471#[zenoh_macros::unstable_doc]
472pub struct AllocBuilder<'a, Backend, Layout, Policy = JustAlloc> {
473    provider: &'a ShmProvider<Backend>,
474    layout: Layout,
475    policy: Policy,
476}
477
478impl<Backend, Layout, Policy> std::fmt::Debug for AllocBuilder<'_, Backend, Layout, Policy> {
479    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
480        f.debug_struct("AllocBuilder")
481            .field("provider", &"..")
482            .field("layout", &"..")
483            .field("policy", &"..")
484            .finish()
485    }
486}
487
488// Generic impl
489impl<'a, Backend, Layout, Policy> AllocBuilder<'a, Backend, Layout, Policy> {
490    /// Set the allocation policy
491    #[zenoh_macros::unstable_doc]
492    pub fn with_policy<OtherPolicy: SafePolicy + ConstPolicy>(
493        self,
494    ) -> AllocBuilder<'a, Backend, Layout, OtherPolicy> {
495        // SAFETY: the policy is safe
496        unsafe { self.with_runtime_policy(OtherPolicy::NEW) }
497    }
498
499    /// Set the unsafe allocation policy
500    ///
501    /// # Safety
502    ///
503    /// The user must ensure its use of allocated buffers are safe taking into account the policy.
504    #[zenoh_macros::unstable_doc]
505    pub unsafe fn with_unsafe_policy<OtherPolicy: ConstPolicy>(
506        self,
507    ) -> AllocBuilder<'a, Backend, Layout, OtherPolicy> {
508        // SAFETY: same function contract
509        unsafe { self.with_runtime_policy(OtherPolicy::NEW) }
510    }
511
512    /// Set the unsafe allocation policy at runtime
513    ///
514    /// # Safety
515    ///
516    /// The user must ensure its use of allocated buffers are safe taking into account the policy.
517    #[zenoh_macros::unstable_doc]
518    pub unsafe fn with_runtime_policy<OtherPolicy>(
519        self,
520        policy: OtherPolicy,
521    ) -> AllocBuilder<'a, Backend, Layout, OtherPolicy> {
522        AllocBuilder {
523            provider: self.provider,
524            layout: self.layout,
525            policy,
526        }
527    }
528
529    fn layout_policy(self) -> Result<(PrecomputedLayout<'a, Backend, Layout>, Policy), ZLayoutError>
530    where
531        Backend: ShmProviderBackend,
532        Layout: AllocLayout,
533    {
534        let layout = PrecomputedLayout::new(self.provider, self.layout.memory_layout()?)?;
535        Ok((layout, self.policy))
536    }
537}
538
539impl<'a, Backend: ShmProviderBackend, Layout: TryInto<MemoryLayout>>
540    AllocBuilder<'a, Backend, Layout>
541where
542    Layout::Error: Into<ZLayoutError>,
543{
544    /// Try to build an allocation layout
545    #[deprecated(
546        since = "1.5.3",
547        note = "Please use `ShmProvider::alloc_layout` method instead"
548    )]
549    #[zenoh_macros::unstable_doc]
550    pub fn into_layout(self) -> Result<PrecomputedLayout<'a, Backend, Layout>, ZLayoutError> {
551        PrecomputedLayout::new(self.provider, self.layout.try_into().map_err(Into::into)?)
552    }
553}
554
555impl<Backend, Layout: AllocLayout, Policy> Resolvable
556    for AllocBuilder<'_, Backend, Layout, Policy>
557{
558    type To = Result<Layout::Buffer, ZLayoutAllocError>;
559}
560
561impl<Backend: ShmProviderBackend, Layout: AllocLayout, Policy: AllocPolicy<Backend>> Wait
562    for AllocBuilder<'_, Backend, Layout, Policy>
563{
564    fn wait(self) -> <Self as Resolvable>::To {
565        let (layout, policy) = self.layout_policy()?;
566        // SAFETY: same contract used to set the policy
567        Ok(unsafe { layout.alloc().with_runtime_policy(policy) }.wait()?)
568    }
569}
570
571impl<
572        'a,
573        Backend: ShmProviderBackend + Sync,
574        Layout: AllocLayout + Send + Sync + 'a,
575        Policy: AsyncAllocPolicy<Backend> + Sync + 'a,
576    > IntoFuture for AllocBuilder<'a, Backend, Layout, Policy>
577{
578    type Output = <Self as Resolvable>::To;
579    type IntoFuture = Pin<Box<dyn Future<Output = <Self as IntoFuture>::Output> + 'a + Send>>;
580
581    fn into_future(self) -> Self::IntoFuture {
582        Box::pin(async move {
583            let (layout, policy) = self.layout_policy()?;
584            // SAFETY: same contract used to set the policy
585            Ok(unsafe { layout.alloc().with_runtime_policy(policy) }.await?)
586        })
587    }
588}
589
590/// Builder for making allocations with instant layout calculation
591#[zenoh_macros::unstable_doc]
592pub struct PrecomputedAllocBuilder<'b, 'a: 'b, Backend, Layout, Policy = JustAlloc> {
593    layout: &'b PrecomputedLayout<'a, Backend, Layout>,
594    policy: Policy,
595}
596
597impl<Backend, Layout, Policy> std::fmt::Debug
598    for PrecomputedAllocBuilder<'_, '_, Backend, Layout, Policy>
599{
600    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
601        f.debug_struct("PrecomputedAllocBuilder")
602            .field("layout", &"..")
603            .field("policy", &"..")
604            .finish()
605    }
606}
607
608// Generic impl
609impl<'b, 'a: 'b, Backend, Layout, Policy> PrecomputedAllocBuilder<'b, 'a, Backend, Layout, Policy> {
610    /// Set the allocation policy
611    #[zenoh_macros::unstable_doc]
612    pub fn with_policy<OtherPolicy: ConstPolicy>(
613        self,
614    ) -> PrecomputedAllocBuilder<'b, 'a, Backend, Layout, OtherPolicy> {
615        // SAFETY: the policy is safe
616        unsafe { self.with_runtime_policy(OtherPolicy::NEW) }
617    }
618
619    /// Set the unsafe allocation policy
620    ///
621    /// # Safety
622    ///
623    /// The user must ensure its use of allocated buffers are safe taking into account the policy.
624    #[zenoh_macros::unstable_doc]
625    pub unsafe fn with_unsafe_policy<OtherPolicy: ConstPolicy>(
626        self,
627    ) -> PrecomputedAllocBuilder<'b, 'a, Backend, Layout, OtherPolicy> {
628        // SAFETY: same function contract
629        unsafe { self.with_runtime_policy(OtherPolicy::NEW) }
630    }
631
632    /// Set the unsafe allocation policy at runtime
633    ///
634    /// # Safety
635    ///
636    /// The user must ensure its use of allocated buffers are safe taking into account the policy.
637    #[zenoh_macros::unstable_doc]
638    pub unsafe fn with_runtime_policy<OtherPolicy>(
639        self,
640        policy: OtherPolicy,
641    ) -> PrecomputedAllocBuilder<'b, 'a, Backend, Layout, OtherPolicy> {
642        PrecomputedAllocBuilder {
643            layout: self.layout,
644            policy,
645        }
646    }
647}
648
649impl<Backend, Layout: AllocLayout, Policy> Resolvable
650    for PrecomputedAllocBuilder<'_, '_, Backend, Layout, Policy>
651{
652    type To = Result<Layout::Buffer, ZAllocError>;
653}
654
655impl<Backend: ShmProviderBackend, Layout: AllocLayout, Policy: AllocPolicy<Backend>> Wait
656    for PrecomputedAllocBuilder<'_, '_, Backend, Layout, Policy>
657{
658    fn wait(self) -> <Self as Resolvable>::To {
659        let buffer = self.layout.provider.alloc_inner(
660            self.layout.size,
661            &self.layout.provider_layout,
662            &self.policy,
663        )?;
664        // SAFETY: the buffer has been allocated with the requested layout
665        Ok(unsafe { Layout::wrap_buffer(buffer) })
666    }
667}
668
669impl<
670        'b,
671        'a: 'b,
672        Backend: ShmProviderBackend + Sync,
673        Layout: AllocLayout + Sync,
674        Policy: AsyncAllocPolicy<Backend> + Sync + 'b,
675    > IntoFuture for PrecomputedAllocBuilder<'b, 'a, Backend, Layout, Policy>
676{
677    type Output = <Self as Resolvable>::To;
678    type IntoFuture = Pin<Box<dyn Future<Output = <Self as IntoFuture>::Output> + 'b + Send>>;
679
680    fn into_future(self) -> Self::IntoFuture {
681        Box::pin(async move {
682            let buffer = self
683                .layout
684                .provider
685                .alloc_inner_async(self.layout.size, &self.layout.provider_layout, &self.policy)
686                .await?;
687            // SAFETY: the buffer has been allocated with the requested layout
688            Ok(unsafe { Layout::wrap_buffer(buffer) })
689        })
690    }
691}
692
693#[zenoh_macros::unstable_doc]
694#[derive(Debug)]
695pub struct ShmProviderBuilder;
696impl ShmProviderBuilder {
697    /// Set the backend
698    #[zenoh_macros::unstable_doc]
699    pub fn backend<Backend: ShmProviderBackend>(
700        backend: Backend,
701    ) -> ShmProviderBuilderBackend<Backend> {
702        ShmProviderBuilderBackend { backend }
703    }
704
705    /// Set the default backend
706    #[zenoh_macros::unstable_doc]
707    pub fn default_backend<Layout>(layout: Layout) -> ShmProviderBuilderWithDefaultBackend<Layout> {
708        ShmProviderBuilderWithDefaultBackend { layout }
709    }
710}
711
712#[zenoh_macros::unstable_doc]
713pub struct ShmProviderBuilderBackend<Backend>
714where
715    Backend: ShmProviderBackend,
716{
717    backend: Backend,
718}
719
720impl<Backend> std::fmt::Debug for ShmProviderBuilderBackend<Backend>
721where
722    Backend: ShmProviderBackend,
723{
724    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
725        f.debug_struct("ShmProviderBuilderBackend")
726            .field("backend", &"..")
727            .finish()
728    }
729}
730#[zenoh_macros::unstable_doc]
731impl<Backend> Resolvable for ShmProviderBuilderBackend<Backend>
732where
733    Backend: ShmProviderBackend,
734{
735    type To = ShmProvider<Backend>;
736}
737
738#[zenoh_macros::unstable_doc]
739impl<Backend> Wait for ShmProviderBuilderBackend<Backend>
740where
741    Backend: ShmProviderBackend,
742{
743    /// build ShmProvider
744    fn wait(self) -> <Self as Resolvable>::To {
745        ShmProvider::new(self.backend)
746    }
747}
748
749#[zenoh_macros::unstable_doc]
750pub struct ShmProviderBuilderWithDefaultBackend<Layout> {
751    layout: Layout,
752}
753
754impl<Layout> std::fmt::Debug for ShmProviderBuilderWithDefaultBackend<Layout> {
755    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
756        f.debug_struct("ShmProviderBuilderWithDefaultBackend")
757            .field("layout", &"..")
758            .finish()
759    }
760}
761
762#[zenoh_macros::unstable_doc]
763impl<Layout> Resolvable for ShmProviderBuilderWithDefaultBackend<Layout> {
764    type To = ZResult<ShmProvider<PosixShmProviderBackend>>;
765}
766
767#[zenoh_macros::unstable_doc]
768impl<Layout: TryInto<MemoryLayout>> Wait for ShmProviderBuilderWithDefaultBackend<Layout>
769where
770    Layout::Error: Error,
771    PosixShmProviderBackendBuilder<Layout>:
772        Resolvable<To = ZResult<PosixShmProviderBackend>> + Wait,
773{
774    /// build ShmProvider
775    fn wait(self) -> <Self as Resolvable>::To {
776        let backend = PosixShmProviderBackend::builder(self.layout).wait()?;
777        Ok(ShmProvider::new(backend))
778    }
779}
780
781#[derive(Debug)]
782struct BusyChunk {
783    metadata: AllocatedMetadataDescriptor,
784}
785
786impl BusyChunk {
787    fn new(metadata: AllocatedMetadataDescriptor) -> Self {
788        Self { metadata }
789    }
790
791    fn descriptor(&self) -> ChunkDescriptor {
792        self.metadata.header().data_descriptor()
793    }
794}
795
796/// A generalized interface for shared memory data sources
797#[zenoh_macros::unstable_doc]
798#[derive(Debug)]
799pub struct ShmProvider<Backend> {
800    backend: Backend,
801    busy_list: Mutex<Vec<BusyChunk>>,
802}
803
804impl<Backend: Default> Default for ShmProvider<Backend> {
805    fn default() -> Self {
806        Self::new(Backend::default())
807    }
808}
809
810impl<Backend> ShmProvider<Backend>
811where
812    Backend: ShmProviderBackend,
813{
814    /// Allocates a buffer with a given memory layout.
815    #[zenoh_macros::unstable_doc]
816    pub fn alloc<Layout>(&self, layout: Layout) -> AllocBuilder<'_, Backend, Layout> {
817        AllocBuilder {
818            provider: self,
819            layout,
820            policy: JustAlloc,
821        }
822    }
823
824    /// Precompute the actual layout for an allocation.
825    #[zenoh_macros::unstable_doc]
826    pub fn alloc_layout<Layout: TryInto<MemoryLayout>>(
827        &self,
828        layout: Layout,
829    ) -> Result<PrecomputedLayout<'_, Backend, Layout>, ZLayoutError>
830    where
831        Layout::Error: Into<ZLayoutError>,
832    {
833        PrecomputedLayout::new(self, layout.try_into().map_err(Into::into)?)
834    }
835    /// Defragment memory
836    #[zenoh_macros::unstable_doc]
837    pub fn defragment(&self) -> usize {
838        self.backend.defragment()
839    }
840
841    /// Map externally-allocated chunk into ZShmMut.
842    ///
843    /// This method is designed to be used with push data sources.
844    /// Remember that chunk's len may be >= len!
845    #[zenoh_macros::unstable_doc]
846    pub fn map(&self, chunk: AllocatedChunk, len: usize) -> Result<ZShmMut, ZAllocError> {
847        let len = len.try_into().map_err(|_| ZAllocError::Other)?;
848
849        // allocate resources for SHM buffer
850        let (allocated_metadata, confirmed_metadata) = Self::alloc_resources()?;
851
852        // wrap everything to ShmBufInner
853        let wrapped = self.wrap(chunk, len, allocated_metadata, confirmed_metadata);
854        // SAFETY: `wrapped` is guaranteed to be valid as it represents a mapped chunk.
855        Ok(unsafe { ZShmMut::new_unchecked(wrapped) })
856    }
857
858    #[zenoh_macros::unstable_doc]
859    fn garbage_collect_impl<const SAFE: bool>(&self) -> usize {
860        fn retain_unordered<T>(vec: &mut Vec<T>, mut f: impl FnMut(&T) -> bool) {
861            let mut i = 0;
862            while i < vec.len() {
863                if f(&vec[i]) {
864                    i += 1;
865                } else {
866                    vec.swap_remove(i); // move last into place of vec[i]
867                                        // don't increment i: need to test the swapped-in element
868                }
869            }
870        }
871
872        tracing::trace!("Running Garbage Collector");
873        let mut largest = 0usize;
874        retain_unordered(&mut zlock!(self.busy_list), |chunk| {
875            let header = chunk.metadata.header();
876            if header.refcount.load(Ordering::SeqCst) == 0
877                || (!SAFE && header.watchdog_invalidated.load(Ordering::SeqCst))
878            {
879                tracing::trace!("Garbage Collecting Chunk: {:?}", chunk);
880                let descriptor_to_free = chunk.descriptor();
881                self.backend.free(&descriptor_to_free);
882                largest = largest.max(descriptor_to_free.len.get());
883                return false;
884            }
885            true
886        });
887        largest
888    }
889
890    /// Try to collect free chunks.
891    ///
892    /// Returns the size of largest collected chunk
893    #[zenoh_macros::unstable_doc]
894    pub fn garbage_collect(&self) -> usize {
895        self.garbage_collect_impl::<true>()
896    }
897
898    /// Try to collect free chunks.
899    ///
900    /// Returns the size of largest collected chunk
901    ///
902    /// # Safety
903    ///
904    /// User must ensure there is no data races with collected chunks, as some of them may still be in-use.
905    #[zenoh_macros::unstable_doc]
906    pub unsafe fn garbage_collect_unsafe(&self) -> usize {
907        self.garbage_collect_impl::<false>()
908    }
909
910    /// Bytes available for use
911    #[zenoh_macros::unstable_doc]
912    pub fn available(&self) -> usize {
913        self.backend.available()
914    }
915}
916
917// PRIVATE impls
918impl<Backend> ShmProvider<Backend> {
919    fn new(backend: Backend) -> Self {
920        Self {
921            backend,
922            busy_list: Default::default(),
923        }
924    }
925}
926
927impl<Backend> ShmProvider<Backend>
928where
929    Backend: ShmProviderBackend,
930{
931    fn alloc_inner<Policy>(
932        &self,
933        size: NonZeroUsize,
934        layout: &MemoryLayout,
935        policy: &Policy,
936    ) -> Result<ZShmMut, ZAllocError>
937    where
938        Policy: AllocPolicy<Backend>,
939    {
940        // allocate resources for SHM buffer
941        let (allocated_metadata, confirmed_metadata) = Self::alloc_resources()?;
942
943        // allocate data chunk
944        // Perform actions depending on the Policy
945        // NOTE: it is necessary to properly map this chunk OR free it if mapping fails!
946        // Don't loose this chunk as it leads to memory leak at the backend side!
947        // NOTE: self.backend.alloc(len) returns chunk with len >= required len,
948        // and it is necessary to handle that properly and pass this len to corresponding free(...)
949        let chunk = policy.alloc(layout, self)?;
950
951        // wrap allocated chunk to ShmBufInner
952        let wrapped = self.wrap(chunk, size, allocated_metadata, confirmed_metadata);
953        // SAFETY: `wrapped` is guaranteed to be valid as it represents an allocated chunk.
954        Ok(unsafe { ZShmMut::new_unchecked(wrapped) })
955    }
956
957    fn alloc_resources() -> Result<(AllocatedMetadataDescriptor, ConfirmedDescriptor), ZAllocError>
958    {
959        // allocate metadata
960        let allocated_metadata = GLOBAL_METADATA_STORAGE.read().allocate()?;
961
962        // add watchdog to confirmator
963        let confirmed_metadata = GLOBAL_CONFIRMATOR.read().add(allocated_metadata.clone());
964
965        Ok((allocated_metadata, confirmed_metadata))
966    }
967
968    fn wrap(
969        &self,
970        chunk: AllocatedChunk,
971        len: NonZeroUsize,
972        allocated_metadata: AllocatedMetadataDescriptor,
973        confirmed_metadata: ConfirmedDescriptor,
974    ) -> ShmBufInner {
975        // write additional metadata
976        // chunk descriptor
977        allocated_metadata
978            .header()
979            .set_data_descriptor(&chunk.descriptor);
980        // protocol
981        allocated_metadata
982            .header()
983            .protocol
984            .store(self.backend.id(), Ordering::Relaxed);
985
986        // add watchdog to validator
987        GLOBAL_VALIDATOR
988            .read()
989            .add(confirmed_metadata.owned.clone());
990
991        // Create buffer's info
992        let info = ShmBufInfo::new(
993            len,
994            MetadataDescriptor::from(&confirmed_metadata.owned),
995            allocated_metadata
996                .header()
997                .generation
998                .load(Ordering::SeqCst),
999        );
1000
1001        // Create buffer
1002        let shmb = ShmBufInner {
1003            metadata: confirmed_metadata,
1004            buf: chunk.data,
1005            info,
1006        };
1007
1008        // Create and store busy chunk
1009        zlock!(self.busy_list).push(BusyChunk::new(allocated_metadata));
1010
1011        shmb
1012    }
1013}
1014
1015// PRIVATE impls for Sync backend
1016impl<Backend> ShmProvider<Backend>
1017where
1018    Backend: ShmProviderBackend + Sync,
1019{
1020    async fn alloc_inner_async<Policy>(
1021        &self,
1022        size: NonZeroUsize,
1023        backend_layout: &MemoryLayout,
1024        policy: &Policy,
1025    ) -> Result<ZShmMut, ZAllocError>
1026    where
1027        Policy: AsyncAllocPolicy<Backend>,
1028    {
1029        // allocate resources for SHM buffer
1030        let (allocated_metadata, confirmed_metadata) = Self::alloc_resources()?;
1031
1032        // allocate data chunk
1033        // Perform actions depending on the Policy
1034        // NOTE: it is necessary to properly map this chunk OR free it if mapping fails!
1035        // Don't loose this chunk as it leads to memory leak at the backend side!
1036        // NOTE: self.backend.alloc(len) returns chunk with len >= required len,
1037        // and it is necessary to handle that properly and pass this len to corresponding free(...)
1038        let chunk = policy.alloc_async(backend_layout, self).await?;
1039
1040        // wrap allocated chunk to ShmBufInner
1041        let wrapped = self.wrap(chunk, size, allocated_metadata, confirmed_metadata);
1042        // SAFETY: `wrapped` is guaranteed to be valid as it represents an allocated chunk.
1043        Ok(unsafe { ZShmMut::new_unchecked(wrapped) })
1044    }
1045}
zenoh_shm/api/provider/shm_provider.rs

zenoh_shm/api/provider/
shm_provider.rs
