datum-agent 0.10.6

Embeddable Datum job registry and lifecycle supervisor
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

use std::{
    sync::atomic::{AtomicU64, Ordering},
    time::Duration,
};

use criterion::{Criterion, criterion_group, criterion_main};
use datum::{Keep, Sink, Source};
use datum_agent::{Agent, AgentConfig, JobMat, JobSpec};

static NEXT_JOB: AtomicU64 = AtomicU64::new(1);

fn empty_job(name: String) -> JobSpec {
    JobSpec::new(name, |context| {
        let control = context.control();
        Ok(Source::<u64>::empty()
            .via_mat(context.drain_flow(), Keep::right)
            .to_mat(Sink::ignore(), move |_switch, completion| {
                JobMat::new(completion, control.clone())
            }))
    })
}

fn submit_and_list() -> usize {
    let agent = Agent::start_with_config(AgentConfig {
        poll_interval: Duration::from_millis(10),
        event_buffer: 64,
        ..AgentConfig::default()
    })
    .expect("agent starts");
    let registry = agent.registry();
    let id = NEXT_JOB.fetch_add(1, Ordering::Relaxed);
    registry
        .submit(empty_job(format!("bench-{id}")))
        .expect("job submits");
    let count = registry.list().expect("jobs list").len();
    registry.shutdown().expect("registry shuts down");
    count
}

fn bench_agent_registry(c: &mut Criterion) {
    c.bench_function("agent_submit_and_list", |b| {
        b.iter(submit_and_list);
    });
}

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