bcast 0.0.29

Low latency broadcast (SPMC) buffer designed to work with shared memory.
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

use anyhow::anyhow;
use bcast::{RingBuffer, error::Error};
use rand::{Rng, thread_rng};
use std::mem::MaybeUninit;

/// Generate random message every 1 millisecond.
#[allow(dead_code)]
pub fn writer(bytes: &[u8]) {
    let writer = RingBuffer::new(bytes).into_writer();
    loop {
        let symbol = thread_rng().gen_range(b'A'..=b'Z');
        let msg_len = thread_rng().gen_range(1..20);
        let random_bytes: Vec<u8> = (0..msg_len).map(|_| symbol).collect();
        let mut claim = writer.claim_with_user_defined(msg_len, true, symbol as u32);
        claim.get_buffer_mut().copy_from_slice(&random_bytes);
        claim.commit();
        std::thread::sleep(std::time::Duration::from_millis(1));
    }
}

/// Consume messages produced by the writer. It will also perform payload validation (using user
/// defined field set by the writer) and sleep for 10 milliseconds in order to process messages
/// in a batch.
#[allow(dead_code)]
pub fn reader(bytes: &[u8]) -> anyhow::Result<()> {
    let reader = RingBuffer::new(bytes).into_reader();
    loop {
        #[cfg(debug_assertions)]
        let mut count = 0;
        if let Some(batch) = reader.read_batch() {
            for msg in batch {
                let msg = match msg {
                    Ok(msg) => msg,
                    Err(Error::Overrun(position)) => {
                        println!("overrun for {} bytes, resetting reader", position);
                        reader.reset();
                        break;
                    }
                    Err(e) => {
                        return Err(anyhow!(e));
                    }
                };
                debug_assert!(!msg.is_padding, "padding frames should be skipped");
                let mut payload = unsafe { MaybeUninit::new([0u8; 1024]).assume_init() };
                msg.read(&mut payload)?;
                #[cfg(debug_assertions)]
                {
                    count += 1;
                    let payload = &payload[..msg.payload_len];
                    assert!(payload.iter().all(|b| *b == msg.user_defined as u8));
                    println!("{}", String::from_utf8_lossy(payload));
                }
            }
        }
        #[cfg(debug_assertions)]
        {
            if count > 0 {
                println!("batch_size: {}", count);
            }
            // adding delay here to simulate impact of batching
            std::thread::sleep(std::time::Duration::from_millis(10));
        }
    }
}

/// Consume messages produced by the writer using the bulk API. It performs the same payload
/// validation as `reader`, but copies the full bulk window first and then iterates over it
/// off-ring.
#[allow(dead_code)]
pub fn bulk_reader(bytes: &[u8]) -> anyhow::Result<()> {
    let reader = RingBuffer::new(bytes).into_reader();
    let mut bulk_bytes = Vec::new();

    loop {
        #[cfg(debug_assertions)]
        let mut count = 0;

        if let Some(bulk) = reader.read_bulk() {
            let bulk = match bulk {
                Ok(bulk) => bulk,
                Err(Error::Overrun(position)) => {
                    println!("overrun for {} bytes, resetting reader", position);
                    reader.reset();
                    continue;
                }
                Err(e) => return Err(anyhow!(e)),
            };

            if bulk_bytes.len() < bulk.len() {
                bulk_bytes.resize(bulk.len(), 0);
            }

            let iter = match bulk.into_iter(&mut bulk_bytes) {
                Ok(iter) => iter,
                Err(Error::Overrun(position)) => {
                    println!("overrun for {} bytes, resetting reader", position);
                    reader.reset();
                    continue;
                }
                Err(e) => return Err(anyhow!(e)),
            };

            for msg in iter {
                #[cfg(debug_assertions)]
                {
                    count += 1;
                    assert!(msg.payload.iter().all(|b| *b == msg.user_defined as u8));
                    println!("{}", String::from_utf8_lossy(msg.payload));
                }
            }
        }

        #[cfg(debug_assertions)]
        {
            if count > 0 {
                println!("batch_size: {}", count);
            }
            std::thread::sleep(std::time::Duration::from_millis(10));
        }
    }
}