taskmill 0.7.1

Adaptive priority work scheduler with IO-aware concurrency and SQLite persistence
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

/// One-shot timing breakdown of dep_chain_submit to identify where time is spent.
/// Run with: cargo run --release --example profile_dep_chain
use std::time::{Duration, Instant};

use serde::{Deserialize, Serialize};
use taskmill::{
    Domain, DomainKey, DomainTaskContext, Scheduler, TaskError, TaskStore, TaskSubmission,
    TypedExecutor, TypedTask,
};

struct BenchDomain;
impl DomainKey for BenchDomain {
    const NAME: &'static str = "bench";
}

#[derive(Serialize, Deserialize)]
struct BenchTask;
impl TypedTask for BenchTask {
    type Domain = BenchDomain;
    const TASK_TYPE: &'static str = "test";
}

struct NoopExecutor;
impl TypedExecutor<BenchTask> for NoopExecutor {
    async fn execute<'a>(
        &'a self,
        _payload: BenchTask,
        _ctx: DomainTaskContext<'a, BenchDomain>,
    ) -> Result<(), TaskError> {
        Ok(())
    }
}

async fn build_scheduler() -> Scheduler {
    Scheduler::builder()
        .store(TaskStore::open_memory().await.unwrap())
        .domain(Domain::<BenchDomain>::new().task::<BenchTask>(NoopExecutor))
        .max_concurrency(8)
        .poll_interval(Duration::from_millis(10))
        .build()
        .await
        .unwrap()
}

async fn run(depth: usize, iters: u32) {
    // Warm up the allocator / SQLite page cache with a throw-away run.
    let _ = build_scheduler().await;

    let mut build_times = Vec::with_capacity(iters as usize);
    let mut first_submit_times = Vec::with_capacity(iters as usize);
    let mut chain_submit_times = Vec::with_capacity(iters as usize);

    for _ in 0..iters {
        let t0 = Instant::now();
        let sched = build_scheduler().await;
        build_times.push(t0.elapsed());

        let t1 = Instant::now();
        let first = sched
            .submit(&TaskSubmission::new("bench::test").key("d-0"))
            .await
            .unwrap()
            .id()
            .unwrap();
        first_submit_times.push(t1.elapsed());

        let t2 = Instant::now();
        let mut prev = first;
        for i in 1..depth {
            prev = sched
                .submit(
                    &TaskSubmission::new("bench::test")
                        .key(format!("d-{i}"))
                        .depends_on(prev),
                )
                .await
                .unwrap()
                .id()
                .unwrap();
        }
        chain_submit_times.push(t2.elapsed());
    }

    let avg = |v: &[Duration]| -> Duration { v.iter().sum::<Duration>() / v.len() as u32 };
    let med = |v: &mut Vec<Duration>| -> Duration {
        v.sort();
        v[v.len() / 2]
    };

    let total_avg = avg(&build_times) + avg(&first_submit_times) + avg(&chain_submit_times);
    let per_chained_submit_avg = avg(&chain_submit_times)
        .checked_div((depth - 1) as u32)
        .unwrap_or_default();

    println!("depth={depth}  iters={iters}");
    println!(
        "  build_scheduler  avg={:>8.3?}  med={:>8.3?}",
        avg(&build_times),
        med(&mut build_times)
    );
    println!(
        "  first submit     avg={:>8.3?}  med={:>8.3?}",
        avg(&first_submit_times),
        med(&mut first_submit_times)
    );
    println!(
        "  chain submits    avg={:>8.3?}  med={:>8.3?}  ({} calls, {:.3?}/call)",
        avg(&chain_submit_times),
        med(&mut chain_submit_times),
        depth - 1,
        per_chained_submit_avg,
    );
    println!("  total            avg={total_avg:>8.3?}");
    println!();
}

#[tokio::main]
async fn main() {
    for depth in [10, 50, 200] {
        run(depth, 20).await;
    }
}