rill_core/queues/
ring.rs

1//! # Кольцевая очередь с произвольным доступом
2//!
3//! [`RingQueue`](crate::queues::ring::RingQueue) — гибрид между кольцевым буфером и очередью,
4//! позволяющий читать данные с произвольной задержкой.
5
6use super::QueueStats;
7use crate::buffer::AtomicCell;
8use std::sync::atomic::{AtomicUsize, Ordering};
9
10/// Кольцевая очередь с произвольным доступом
11///
12/// Позволяет читать данные не только из головы, но и с произвольной
13/// задержкой. Полезно для эффектов задержки и реверберации.
14#[repr(C, align(64))]
15pub struct RingQueue<T: Copy, const CAP: usize> {
16    /// Данные
17    data: [AtomicCell<T>; CAP],
18    /// Индекс записи
19    write_pos: AtomicUsize,
20    /// Маска для быстрого вычисления
21    mask: usize,
22    /// Статистика
23    stats: QueueStats,
24}
25
26impl<T: Copy + Default, const CAP: usize> Default for RingQueue<T, CAP> {
27    fn default() -> Self { Self::new() }
28}
29
30impl<T: Copy + Default, const CAP: usize> RingQueue<T, CAP> {
31    /// Создать новую кольцевую очередь
32    pub fn new() -> Self {
33        assert!(CAP.is_power_of_two(), "CAP must be a power of two");
34
35        let data = std::array::from_fn(|_| AtomicCell::new(T::default()));
36
37        Self {
38            data,
39            write_pos: AtomicUsize::new(0),
40            mask: CAP - 1,
41            stats: QueueStats::new(),
42        }
43    }
44
45    /// Записать элемент (всегда успешно)
46    pub fn push(&self, value: T) {
47        let pos = self.write_pos.load(Ordering::Relaxed);
48        self.data[pos].store(value);
49        self.write_pos
50            .store((pos + 1) & self.mask, Ordering::Release);
51        self.stats.record_push(self.len());
52    }
53
54    /// Прочитать элемент с задержкой
55    ///
56    /// # Arguments
57    /// * `delay` - задержка в семплах (0 = последний записанный)
58    pub fn read_delayed(&self, delay: usize) -> T {
59        assert!(delay < CAP, "Delay must be less than CAP");
60
61        let write_pos = self.write_pos.load(Ordering::Acquire);
62        let read_pos = (write_pos + CAP - delay - 1) & self.mask;
63
64        self.data[read_pos].load()
65    }
66
67    /// Прочитать элемент с плавающей задержкой (линейная интерполяция)
68    pub fn read_interpolated(&self, delay_frac: f64) -> T
69    where
70        T: From<f64> + Into<f64>,
71    {
72        let delay_int = delay_frac.floor() as usize;
73        let frac = delay_frac.fract();
74
75        let s1: f64 = self.read_delayed(delay_int).into();
76        let s2: f64 = self.read_delayed(delay_int + 1).into();
77
78        T::from(s1 * (1.0 - frac) + s2 * frac)
79    }
80
81    /// Прочитать элемент по абсолютному индексу
82    pub fn read_at(&self, index: usize) -> T {
83        let write_pos = self.write_pos.load(Ordering::Acquire);
84        let read_pos = (write_pos + CAP - index - 1) & self.mask;
85        self.data[read_pos].load()
86    }
87
88    /// Записать массив данных
89    pub fn push_slice(&self, slice: &[T]) {
90        for &value in slice {
91            self.push(value);
92        }
93    }
94
95    /// Прочитать срез данных с задержкой
96    pub fn read_slice_delayed(&self, delay: usize, output: &mut [T]) {
97        for (i, out) in output.iter_mut().enumerate() {
98            *out = self.read_delayed(delay + i);
99        }
100    }
101
102    /// Текущая позиция записи
103    pub fn write_pos(&self) -> usize {
104        self.write_pos.load(Ordering::Acquire)
105    }
106
107    /// Ёмкость
108    pub const fn capacity(&self) -> usize {
109        CAP
110    }
111
112    /// Количество записанных элементов (не больше CAP)
113    pub fn len(&self) -> usize {
114        CAP
115    }
116
117    /// Returns `true` if no elements have been written.
118    pub fn is_empty(&self) -> bool {
119        self.len() == 0
120    }
121
122    /// Сбросить позицию записи
123    pub fn reset(&self) {
124        self.write_pos.store(0, Ordering::Release);
125    }
126
127    /// Получить статистику
128    pub fn stats(&self) -> &QueueStats {
129        &self.stats
130    }
131}
132
133#[allow(unsafe_code)]
134unsafe impl<T: Copy + Send, const CAP: usize> Send for RingQueue<T, CAP> {}
135#[allow(unsafe_code)]
136unsafe impl<T: Copy + Sync, const CAP: usize> Sync for RingQueue<T, CAP> {}
137
138#[cfg(test)]
139mod tests {
140    use super::*;
141
142    #[test]
143    fn test_ring_queue_basic() {
144        let queue = RingQueue::<i32, 4>::new();
145
146        queue.push(1);
147        queue.push(2);
148        queue.push(3);
149        queue.push(4);
150
151        assert_eq!(queue.read_delayed(0), 4);
152        assert_eq!(queue.read_delayed(1), 3);
153        assert_eq!(queue.read_delayed(2), 2);
154        assert_eq!(queue.read_delayed(3), 1);
155    }
156
157    #[test]
158    fn test_ring_queue_wraparound() {
159        let queue = RingQueue::<i32, 4>::new();
160
161        for i in 0..10 {
162            queue.push(i);
163        }
164
165        // После переполнения должны быть последние 4 значения
166        assert_eq!(queue.read_delayed(0), 9);
167        assert_eq!(queue.read_delayed(1), 8);
168        assert_eq!(queue.read_delayed(2), 7);
169        assert_eq!(queue.read_delayed(3), 6);
170    }
171
172    #[test]
173    fn test_ring_queue_interpolated() {
174        let queue = RingQueue::<f64, 4>::new();
175
176        queue.push(1.0);
177        queue.push(2.0);
178        queue.push(3.0);
179        queue.push(4.0);
180
181        let val = queue.read_interpolated(1.5);
182        assert!((val - 2.5).abs() < 0.001);
183    }
184}
rill_core/queues/ring.rs

rill_core/queues/
ring.rs
