oneringbuf 0.7.0

A lock-free single-producer, single-consumer (SPSC) ring buffer with in-place mutability, asynchronous support, and virtual memory optimisation.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137

use crate::OneRB;
use crate::StorageComponent;
#[allow(unused_imports)]
use crate::iterators::WorkIter;
use crate::iterators::iterator_trait::ORBIterator;
use crate::iterators::util_macros::delegate;
use crate::iterators::util_macros::muncher;
#[cfg(feature = "async")]
use crate::{
    iterators::{AsyncDetached, async_iterators::AsyncIterator},
    iters_components::async_iters::AsyncIterComp,
};

#[doc = r##"
Detached iterator: does not update the atomic index when advancing.

This makes it possible to explore available data back and forth, putting other iterators on hold.

A typical use case of this structure is to search something amidst produced data, aligning the detached
iterator to a suitable index and then returning to a normal iterator.

This struct can only be created by [`detaching`](ORBIterator::detach) an iterator.

When done worker iterator can be re-obtained via [`Self::attach`].

Note that, in order to avoid buffer saturation, the global index can be synced with [`Self::sync_index`];
this synchronises indices making the consumer iterator able to move on.

<div class="warning">

As [`WorkIter`], this iterator returns mutable references to data stored within the buffer.
Thus, as stated in the docs written for the former, [`Self::advance`] has to be called when done with the mutation
in order to move the iterator.
</div>
"##]
#[repr(transparent)]
pub struct Detached<I: ORBIterator> {
    inner: I,
}

unsafe impl<I: ORBIterator> Send for Detached<I> {}

#[cfg(feature = "async")]
impl<'buf, I: ORBIterator<Buffer: OneRB<Iters: AsyncIterComp>>> Detached<I> {
    pub fn into_async<AS: AsyncIterator<'buf, I = I>>(self) -> AsyncDetached<'buf, AS> {
        AsyncDetached::from_iter(AsyncIterator::from_sync(self.inner))
    }
}

impl<T, I: ORBIterator<Item = T>> Detached<I> {
    /// Creates a [`Self`] from an iterator.
    #[inline]
    pub(crate) fn from_iter(iter: I) -> Detached<I> {
        Self { inner: iter }
    }

    /// Attaches and yields the iterator.
    #[inline]
    pub fn attach(self) -> I {
        self.sync_index();
        self.inner
    }

    #[inline]
    fn inner(&self) -> &I {
        &self.inner
    }
    #[inline]
    fn inner_mut(&mut self) -> &mut I {
        &mut self.inner
    }

    delegate!(ORBIterator (inline), pub fn available(&(mut) self) -> usize);
    delegate!(ORBIterator (inline), pub fn wait_for(&(mut) self, count: usize));
    delegate!(ORBIterator (inline), pub fn index(&self) -> usize);
    delegate!(ORBIterator (inline), pub fn buf_len(&self) -> usize);

    /// Sets the *local* index. To sync the atomic index, use [`Self::sync_index`].
    ///
    /// # Safety
    /// Index must always be between consumer and producer.
    #[inline]
    pub unsafe fn set_index(&mut self, index: usize) {
        self.inner.set_local_index(index);
    }

    /// Resets the *local* index of the iterator. I.e., moves the iterator to the location occupied by its successor.
    /// To sync the atomic index, use [`Self::sync_index`].
    #[inline]
    pub fn reset_index(&mut self) {
        let new_idx = self.inner.succ_index();
        self.inner.set_local_index(new_idx);
    }

    /// Advances the iterator as in [`ORBIterator::advance()`], but does not modify the atomic counter,
    /// making the change local.
    ///
    /// # Safety
    /// See [`ORBIterator::advance`].
    #[inline]
    pub unsafe fn advance(&mut self, count: usize) {
        unsafe { self.inner.advance_local(count) };
    }

    /// Goes back, wrapping if necessary.
    ///
    /// # Safety
    /// Index must always be between consumer and producer.
    pub unsafe fn go_back(&mut self, count: usize) {
        let idx = self.inner.index();

        self.inner.set_local_index(match idx < count {
            true => unsafe { self.inner.buf_len().unchecked_sub(count).unchecked_sub(idx) },
            false => unsafe { idx.unchecked_sub(count) },
        });

        let cached_avail = self.inner.cached_avail();
        self.inner
            .set_cached_avail(unsafe { cached_avail.unchecked_add(count) });
    }

    delegate!(ORBIterator (inline), pub fn prod_index(&self) -> usize);
    delegate!(ORBIterator (inline), pub fn work_index(&self) -> usize);
    delegate!(ORBIterator (inline), pub fn cons_index(&self) -> usize);

    delegate!(ORBIterator (inline), pub fn get_mut(&(mut) self) -> Option<&'_ mut T>);
    delegate!(ORBIterator (inline), pub fn get_mut_slice_exact(&(mut) self, count: usize) -> Option<<<I::Buffer as OneRB>::Storage as StorageComponent>::SliceOutputMut<'_>>);
    delegate!(ORBIterator (inline), pub fn get_mut_slice_avail(&(mut) self) -> Option<<<I::Buffer as OneRB>::Storage as StorageComponent>::SliceOutputMut<'_>>);
    delegate!(ORBIterator (inline), pub fn get_mut_slice_multiple_of(&(mut) self, rhs: usize) -> Option<<<I::Buffer as OneRB>::Storage as StorageComponent>::SliceOutputMut<'_>>);

    /// Synchronises the underlying atomic index with the local index. I.e. let the consumer iterator
    /// advance.
    #[inline]
    pub fn sync_index(&self) {
        self.inner.set_atomic_index(self.inner.index());
    }
}