Struct AtomicQueue

Source

pub struct AtomicQueue<'a, T> { /* private fields */ }

Expand description

A thread safe, lock-free atomic queue. Objects are added to the queue via a closure and are owned by the queue as well This means objects in the queue will only be dropped when the queue’s backing buffer is also destroyed.

Implementations§

Source §

impl<'a, T> AtomicQueue<'a, T>


use rustpool::AtomicQueue;
use rustpool::RPQueue;
fn main() {
    // create a blocking queue with 3 elements set to value 0
    let q = AtomicQueue::<isize>::new(3, || 0);

    println!("simple blocking queue size {}", q.capacity());

    // take the first element
    let v1 = q.take().unwrap();

    println!("first element is {}", v1);

    // remove the remaining items
    let v2 = q.take().unwrap();
    let v3 = q.take().unwrap();

    // how many items available now?
    println!("items in the queue {}", q.available());

    // no more items to return but the atomic queue will never block
    let empty = q.take_wait(std::time::Duration::from_secs(1));

    assert_eq!(None, empty);

    // put everything back
    q.offer(v1);
    q.offer(v2);
    q.offer(v3);

    // how many items available now?
    println!("items in the queue {}", q.available());

    let new_value: isize = 1;

    // we can't add more items than the original capacity
    q.offer(&new_value);

    // we still have the same number of available items
    println!("items in the queue {}", q.available());
}

Source

pub fn new(size: usize, f: impl FnMut() -> T) -> Self

Create a new queue with fixed size size using custom create function

Examples found in repository ?

examples/atomic_queue.rs (line 7)

5fn main() {
6    // create a blocking queue with 3 elements set to value 0
7    let q = AtomicQueue::<isize>::new(3, || 0);
8
9    println!("simple blocking queue size {}", q.capacity());
10
11    // take the first element
12    let v1 = q.take().unwrap();
13
14    println!("first element is {}", v1);
15
16    // remove the remaining items
17    let v2 = q.take().unwrap();
18    let v3 = q.take().unwrap();
19
20    // how many items available now?
21    println!("items in the queue {}", q.available());
22
23    // no more items to return but the atomic queue will never block
24    let empty = q.take_wait(std::time::Duration::from_secs(1));
25
26    assert_eq!(None, empty);
27
28    // put everything back
29    q.offer(v1);
30    q.offer(v2);
31    q.offer(v3);
32
33    // how many items available now?
34    println!("items in the queue {}", q.available());
35
36    let new_value: isize = 1;
37
38    // we can't add more items than the original capacity
39    q.offer(&new_value);
40
41    // we still have the same number of available items
42    println!("items in the queue {}", q.available());
43}