Struct ringbuf::ring_buffer::SharedRb

source ·

pub struct SharedRb<T, C: Container<T>> { /* private fields */ }

Expand description

Ring buffer that could be shared between threads.

Implements Sync if T implements Send. And therefore its Producer and Consumer implement Send.

Note that there is no explicit requirement of T: Send. Instead SharedRb will work just fine even with T: !Send until you try to send its Producer or Consumer to another thread.

use std::{thread, vec::Vec};
use ringbuf::SharedRb;

let (mut prod, mut cons) = SharedRb::<i32, Vec<_>>::new(256).split();
thread::spawn(move || {
    prod.push(123).unwrap();
})
.join();
thread::spawn(move || {
    assert_eq!(cons.pop().unwrap(), 123);
})
.join();

Implementations§

source §

impl<T> SharedRb<T, Vec<MaybeUninit<T>>>

source

pub fn new(capacity: usize) -> Self

Creates a new instance of a ring buffer.

Panics if allocation failed or capacity is zero.

Examples found in repository ?

examples/overwrite.rs (line 4)

fn main() {
    let mut rb = HeapRb::<i32>::new(2);

    assert_eq!(rb.push_overwrite(0), None);
    assert_eq!(rb.push_overwrite(1), None);
    assert_eq!(rb.push_overwrite(2), Some(0));

    assert_eq!(rb.pop(), Some(1));
    assert_eq!(rb.pop(), Some(2));
    assert_eq!(rb.pop(), None);
}

More examples

Hide additional examples

examples/simple.rs (line 4)

fn main() {
    let rb = HeapRb::<i32>::new(2);
    let (mut prod, mut cons) = rb.split();

    prod.push(0).unwrap();
    prod.push(1).unwrap();
    assert_eq!(prod.push(2), Err(2));

    assert_eq!(cons.pop().unwrap(), 0);

    prod.push(2).unwrap();

    assert_eq!(cons.pop().unwrap(), 1);
    assert_eq!(cons.pop().unwrap(), 2);
    assert_eq!(cons.pop(), None);
}

examples/message.rs (line 6)

fn main() {
    let buf = HeapRb::<u8>::new(10);
    let (mut prod, mut cons) = buf.split();

    let smsg = "The quick brown fox jumps over the lazy dog";

    let pjh = thread::spawn(move || {
        println!("-> sending message: '{}'", smsg);

        let zero = [0];
        let mut bytes = smsg.as_bytes().chain(&zero[..]);
        loop {
            if prod.is_full() {
                println!("-> buffer is full, waiting");
                thread::sleep(Duration::from_millis(1));
            } else {
                let n = prod.read_from(&mut bytes, None).unwrap();
                if n == 0 {
                    break;
                }
                println!("-> {} bytes sent", n);
            }
        }

        println!("-> message sent");
    });

    let cjh = thread::spawn(move || {
        println!("<- receiving message");

        let mut bytes = Vec::<u8>::new();
        loop {
            if cons.is_empty() {
                if bytes.ends_with(&[0]) {
                    break;
                } else {
                    println!("<- buffer is empty, waiting");
                    thread::sleep(Duration::from_millis(1));
                }
            } else {
                let n = cons.write_into(&mut bytes, None).unwrap();
                println!("<- {} bytes received", n);
            }
        }

        assert_eq!(bytes.pop().unwrap(), 0);
        let msg = String::from_utf8(bytes).unwrap();
        println!("<- message received: '{}'", msg);

        msg
    });

    pjh.join().unwrap();
    let rmsg = cjh.join().unwrap();

    assert_eq!(smsg, rmsg);
}

source

pub fn try_new(capacity: usize) -> Result<Self, TryReserveError>

Creates a new instance of a ring buffer returning an error if allocation failed.

Panics if capacity is zero.

source §

impl<T, C: Container<T>> SharedRb<T, C>

source

pub unsafe fn from_raw_parts(container: C, head: usize, tail: usize) -> Self

Constructs ring buffer from container and counters.

Safety

The items in container inside head..tail range must be initialized, items outside this range must be uninitialized. head and tail values must be valid (see RbBase).

source

pub unsafe fn into_raw_parts(self) -> (C, usize, usize)

Destructures ring buffer into underlying container and head and tail counters.

Safety

Initialized contents of the container must be properly dropped.

source

pub fn split(self) -> (Producer<T, Arc<Self>>, Consumer<T, Arc<Self>>)where Self: Sized,

Splits ring buffer into producer and consumer.

This method consumes the ring buffer and puts it on heap in Arc. If you don’t want to use heap the see Self::split_ref.

Examples found in repository ?

examples/simple.rs (line 5)

fn main() {
    let rb = HeapRb::<i32>::new(2);
    let (mut prod, mut cons) = rb.split();

    prod.push(0).unwrap();
    prod.push(1).unwrap();
    assert_eq!(prod.push(2), Err(2));

    assert_eq!(cons.pop().unwrap(), 0);

    prod.push(2).unwrap();

    assert_eq!(cons.pop().unwrap(), 1);
    assert_eq!(cons.pop().unwrap(), 2);
    assert_eq!(cons.pop(), None);
}

More examples

Hide additional examples

examples/message.rs (line 7)

fn main() {
    let buf = HeapRb::<u8>::new(10);
    let (mut prod, mut cons) = buf.split();

    let smsg = "The quick brown fox jumps over the lazy dog";

    let pjh = thread::spawn(move || {
        println!("-> sending message: '{}'", smsg);

        let zero = [0];
        let mut bytes = smsg.as_bytes().chain(&zero[..]);
        loop {
            if prod.is_full() {
                println!("-> buffer is full, waiting");
                thread::sleep(Duration::from_millis(1));
            } else {
                let n = prod.read_from(&mut bytes, None).unwrap();
                if n == 0 {
                    break;
                }
                println!("-> {} bytes sent", n);
            }
        }

        println!("-> message sent");
    });

    let cjh = thread::spawn(move || {
        println!("<- receiving message");

        let mut bytes = Vec::<u8>::new();
        loop {
            if cons.is_empty() {
                if bytes.ends_with(&[0]) {
                    break;
                } else {
                    println!("<- buffer is empty, waiting");
                    thread::sleep(Duration::from_millis(1));
                }
            } else {
                let n = cons.write_into(&mut bytes, None).unwrap();
                println!("<- {} bytes received", n);
            }
        }

        assert_eq!(bytes.pop().unwrap(), 0);
        let msg = String::from_utf8(bytes).unwrap();
        println!("<- message received: '{}'", msg);

        msg
    });

    pjh.join().unwrap();
    let rmsg = cjh.join().unwrap();

    assert_eq!(smsg, rmsg);
}

source

pub fn split_ref(&mut self) -> (Producer<T, &Self>, Consumer<T, &Self>)where Self: Sized,

Splits ring buffer into producer and consumer without using the heap.

In this case producer and consumer stores a reference to the ring buffer, so you also need to store the buffer somewhere.

Examples found in repository ?

examples/static.rs (line 8)

fn main() {
    const RB_SIZE: usize = 1;
    let mut rb = StaticRb::<i32, RB_SIZE>::default();
    let (mut prod, mut cons) = rb.split_ref();

    assert_eq!(prod.push(123), Ok(()));
    assert_eq!(prod.push(321), Err(321));

    assert_eq!(cons.pop(), Some(123));
    assert_eq!(cons.pop(), None);
}