carpool 0.1.0

Deduplicate and batch concurrent async requests through a collection window
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
122

use std::hint::black_box;
use std::time::Duration;

use criterion::{BatchSize, Criterion, criterion_group, criterion_main};
use tokio::sync::oneshot;

use carpool::queue::Queue;

type Tx = oneshot::Sender<u64>;
type Rx = oneshot::Receiver<u64>;

const FILL: usize = 1024;
const BATCH: usize = 128;

// Callers' channel halves, prebuilt so enqueue benches measure the queue alone.
fn channels(n: usize) -> (Vec<(u64, Tx)>, Vec<Rx>) {
    (0..n as u64)
        .map(|i| {
            let (tx, rx) = oneshot::channel();
            ((i, tx), rx)
        })
        .unzip()
}

fn filled_queue() -> (Queue<u64, u64>, Vec<Rx>) {
    let (pairs, rxs) = channels(FILL);
    let queue = Queue::default();
    for (input, tx) in pairs {
        queue.push(input, tx);
    }
    (queue, rxs)
}

fn fill(c: &mut Criterion) {
    c.bench_function("fill_1024", |b| {
        b.iter_batched(
            || channels(FILL),
            |(pairs, rxs)| {
                let queue: Queue<u64, u64> = Queue::default();
                for (input, tx) in pairs {
                    queue.push(input, tx);
                }
                (queue, rxs)
            },
            BatchSize::SmallInput,
        )
    });
}

// Every caller cancels; one full take scan buries the dead.
fn cancel_all(c: &mut Criterion) {
    c.bench_function("cancel_all_1024", |b| {
        b.iter_batched(
            filled_queue,
            |(queue, rxs)| {
                drop(rxs);
                black_box(queue.take(usize::MAX));
                queue
            },
            BatchSize::SmallInput,
        )
    });
}

fn slice(c: &mut Criterion) {
    c.bench_function("slice_take_128_of_1024", |b| {
        b.iter_batched(
            filled_queue,
            |(queue, rxs)| {
                let batch = queue.take(BATCH);
                (queue, batch, rxs)
            },
            BatchSize::SmallInput,
        )
    });
    c.bench_function("slice_take_if_128_of_1024", |b| {
        b.iter_batched(
            filled_queue,
            |(queue, rxs)| {
                let batch = queue.take_if(BATCH).expect("all entries are live");
                (queue, batch, rxs)
            },
            BatchSize::SmallInput,
        )
    });
}

// One consumer window end to end: fill, a quarter cancels, slice a batch, deliver it.
// The leftover tail survives into the next window, so its teardown is not timed.
fn window_cycle(c: &mut Criterion) {
    const N: usize = 512;
    c.bench_function("window_cycle_512", |b| {
        b.iter_with_large_drop(|| {
            let queue: Queue<u64, u64> = Queue::default();
            let mut rxs = Vec::with_capacity(N);
            for i in 0..N as u64 {
                let (tx, rx) = oneshot::channel();
                queue.push(i, tx);
                rxs.push(Some(rx));
            }
            for i in (0..N).step_by(4) {
                rxs[i] = None;
            }
            let batch = queue
                .take_if(BATCH)
                .expect("384 live entries cover the batch");
            for (input, tx) in batch {
                tx.send(input * 2).expect("receiver is alive");
            }
            (queue, rxs)
        })
    });
}

criterion_group! {
    name = benches;
    config = Criterion::default()
        .warm_up_time(Duration::from_secs(1))
        .measurement_time(Duration::from_secs(3));
    targets = fill, cancel_all, slice, window_cycle
}
criterion_main!(benches);