threadteam 0.2.0

Scoped CPU thread-team placement token: a Pool and a non-Clone Exec with fork, for_ranges, external and broadcast; no Rayon type in the public API.
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

//! Ledger: team participants count as live tasks, never more than the budget.
#![cfg(all(feature = "parallel", feature = "ledger"))]

use std::time::Duration;
use threadteam::{ledger, Exec, Pool};

/// A hang is a failure: abort the whole process if the test body stalls.
fn watchdog() -> impl Drop {
    struct Stop(std::sync::mpsc::Sender<()>);
    impl Drop for Stop {
        fn drop(&mut self) {
            let _ = self.0.send(());
        }
    }
    let (tx, rx) = std::sync::mpsc::channel::<()>();
    std::thread::spawn(move || {
        if rx.recv_timeout(Duration::from_secs(45)).is_err() {
            eprintln!("watchdog: test stalled");
            std::process::exit(101);
        }
    });
    Stop(tx)
}

fn pool(n: usize) -> Pool {
    Pool::builder()
        .threads(n)
        .idle_spin(Duration::ZERO)
        .acquire_timeout(Duration::from_secs(5))
        .build()
        .unwrap()
}

fn nap() {
    std::thread::sleep(Duration::from_millis(2));
}

#[test]
fn team_sections_and_nested_children_never_exceed_the_root_budget() {
    let _w = watchdog();
    let pool = pool(8);
    for budget in [2, 4, 8] {
        let mut exec = Exec::new(&pool, budget).unwrap();
        let root = exec.ledger_root().unwrap();
        exec.team(|mut team| {
            team.reborrow().run(|_, _| nap());
            team.reborrow().run_with(|p| {
                p.barrier().unwrap();
                nap();
            });
            team.fork(2, |_, child| {
                child.fork(2, |_, grand| {
                    grand.run(|_, _| nap());
                });
            });
        })
        .unwrap();
        assert!(ledger::max_live(root) <= budget, "budget {budget}");
        assert!(ledger::max_live(root) >= 1);
        assert_eq!(ledger::live(root), 0);
    }
}

#[test]
fn a_team_inside_a_fork_child_stays_within_the_budget() {
    let _w = watchdog();
    let pool = pool(8);
    let exec = Exec::new(&pool, 8).unwrap();
    let root = exec.ledger_root().unwrap();
    exec.fork(2, |_, mut child| {
        // Each child holds 4 slots: it leads a team of 4 or falls back.
        let _ = child.team(|team| team.run(|_, _| nap()));
    });
    assert!(ledger::max_live(root) <= 8);
    assert_eq!(ledger::live(root), 0);
}

#[test]
fn team_tokens_with_nested_forks_stay_within_the_budget() {
    let _w = watchdog();
    let pool = pool(8);
    let team = pool.into_team(8).unwrap();
    for budget in [1, 2, 3, 5, 8] {
        let exec = team.exec(budget);
        let root = exec.ledger_root().unwrap();
        exec.fork(5, |_, l1| {
            l1.fork(4, |_, l2| {
                l2.fork(3, |_, mut l3| {
                    nap();
                    l3.reborrow()
                        .for_ranges(10, std::num::NonZeroUsize::MIN, |_| nap());
                    let _ = l3.reborrow().broadcast(|_| nap());
                    l3.fork(2, |_, _| nap());
                });
            });
        });
        assert!(ledger::max_live(root) <= budget, "budget {budget}");
        assert!(ledger::max_live(root) >= 1);
        assert_eq!(ledger::live(root), 0);
    }
}