disk_backed_queue 0.1.3

A robust, crash-resistant queue implementation that persists all data to disk using SQLite
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

use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
use disk_backed_queue::DiskBackedQueue;
use tempfile::NamedTempFile;
use tokio::runtime::Runtime;

fn benchmark_throughput(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();
    let mut group = c.benchmark_group("throughput");

    // Define message sizes to benchmark
    let msg_sizes = [64, 1024]; // 64 bytes, 1KB

    for size in msg_sizes {
        let msg = vec![0u8; size];

        // Benchmark individual send
        group.throughput(Throughput::Bytes(size as u64));
        group.bench_with_input(BenchmarkId::new("send", size), &size, |b, &_size| {
            b.to_async(&rt).iter_with_setup(
                || {
                    let tmp = NamedTempFile::new().unwrap();
                    let path = tmp.path().to_owned();
                    let queue = rt
                        .block_on(DiskBackedQueue::<Vec<u8>>::new(
                            &path,
                            "bench_send".to_string(),
                            None,
                        ))
                        .unwrap();
                    (queue, tmp, msg.clone())
                },
                |(queue, _tmp, msg)| async move {
                    queue.send(msg).await.unwrap();
                },
            );
        });

        // Benchmark batch send (batch size 100)
        let batch_size = 100;
        let total_bytes = (size * batch_size) as u64;
        group.throughput(Throughput::Bytes(total_bytes));
        group.bench_with_input(
            BenchmarkId::new("send_batch_100", size),
            &size,
            |b, &_size| {
                b.to_async(&rt).iter_with_setup(
                    || {
                        let tmp = NamedTempFile::new().unwrap();
                        let path = tmp.path().to_owned();
                        let queue = rt
                            .block_on(DiskBackedQueue::<Vec<u8>>::new(
                                &path,
                                "bench_batch".to_string(),
                                None,
                            ))
                            .unwrap();
                        let batch = vec![msg.clone(); batch_size];
                        (queue, tmp, batch)
                    },
                    |(queue, _tmp, batch)| async move {
                        queue.send_batch(batch).await.unwrap();
                    },
                );
            },
        );
    }

    group.finish();
}

criterion_group!(benches, benchmark_throughput);
criterion_main!(benches);