zenoh_sync/
fifo_queue.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//
14use tokio::sync::Mutex;
15use zenoh_collections::RingBuffer;
16use zenoh_core::zasynclock;
17
18use crate::Condition;
19
20#[derive(Debug)]
21pub struct FifoQueue<T> {
22    not_empty: Condition,
23    not_full: Condition,
24    buffer: Mutex<RingBuffer<T>>,
25}
26
27impl<T> FifoQueue<T> {
28    pub fn new(capacity: usize) -> FifoQueue<T> {
29        FifoQueue {
30            not_empty: Condition::new(),
31            not_full: Condition::new(),
32            buffer: Mutex::new(RingBuffer::new(capacity)),
33        }
34    }
35
36    pub fn try_push(&self, x: T) -> Option<T> {
37        if let Ok(mut guard) = self.buffer.try_lock() {
38            let res = guard.push(x);
39            if res.is_none() {
40                drop(guard);
41                self.not_empty.notify_one();
42            }
43            return res;
44        }
45        Some(x)
46    }
47
48    pub async fn push(&self, x: T) {
49        loop {
50            let mut guard = zasynclock!(self.buffer);
51            if !guard.is_full() {
52                guard.push(x);
53                drop(guard);
54                self.not_empty.notify_one();
55                return;
56            }
57            self.not_full.wait(guard).await;
58        }
59    }
60
61    pub fn try_pull(&self) -> Option<T> {
62        if let Ok(mut guard) = self.buffer.try_lock() {
63            if let Some(e) = guard.pull() {
64                drop(guard);
65                self.not_full.notify_one();
66                return Some(e);
67            }
68        }
69        None
70    }
71
72    pub async fn pull(&self) -> T {
73        loop {
74            let mut guard = zasynclock!(self.buffer);
75            if let Some(e) = guard.pull() {
76                drop(guard);
77                self.not_full.notify_one();
78                return e;
79            }
80            self.not_empty.wait(guard).await;
81        }
82    }
83}
zenoh_sync/fifo_queue.rs

zenoh_sync/
fifo_queue.rs
