how-far 0.1.0

Cancellation, progress, and weighted phases through one interface for library authors
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

//! A reference, box or `Arc` of a pulse is a pulse, so code that takes a pulse
//! by value or owns one can hold any pulse, `NoPulse` included.

use how_far::{
    Child, Execution, NoPulse, PhaseSpec, PlanError, ProgressExt, Pulse, Report, SharedPulse, Stop,
    StopReason, Total,
};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};

/// Stops on request, counts, plans inert children, and shares itself.
#[derive(Clone, Default)]
struct Flagged {
    stopped: Arc<AtomicBool>,
    units: Arc<AtomicU64>,
}
struct Flag(Arc<AtomicBool>);
impl Stop for Flag {
    fn check(&self) -> Result<(), StopReason> {
        if self.0.load(Ordering::Relaxed) {
            Err(StopReason::Cancelled)
        } else {
            Ok(())
        }
    }
}
impl Stop for Flagged {
    fn check(&self) -> Result<(), StopReason> {
        Flag(self.stopped.clone()).check()
    }
}
impl Report for Flagged {
    fn advance(&self, completed: u64) {
        self.units.fetch_add(completed, Ordering::Relaxed);
    }
}
impl Pulse for Flagged {
    fn split(
        &self,
        execution: Execution,
        parts: &[PhaseSpec<'_>],
    ) -> Result<Vec<Child<'_>>, PlanError> {
        NoPulse.split(execution, parts)
    }
    fn share(&self) -> Result<SharedPulse, PlanError> {
        Ok(SharedPulse::new(self.clone()))
    }
}

/// Generic code that takes its pulse by value.
fn two_stages<P: Pulse>(pulse: P) -> Result<usize, PlanError> {
    let parts = [
        PhaseSpec::new("a", 1, Total::Exact(1)),
        PhaseSpec::new("b", 1, Total::Exact(1)),
    ];
    Ok(pulse.split(Execution::Sequence, &parts)?.len())
}

#[test]
fn generic_code_takes_any_pulse_by_value() {
    assert_eq!(two_stages(&NoPulse), Ok(2));
    assert_eq!(two_stages(Box::new(&NoPulse)), Ok(2));
    assert_eq!(two_stages(Arc::new(&NoPulse)), Ok(2));
    let mut flagged = Flagged::default();
    assert_eq!(two_stages(&flagged), Ok(2));
    assert_eq!(two_stages(&mut flagged), Ok(2));
    assert_eq!(two_stages(Box::new(flagged)), Ok(2));
}

#[test]
fn owned_pulses_can_default_to_no_pulse() {
    struct Options {
        pulse: Box<dyn Pulse>,
    }
    impl Default for Options {
        fn default() -> Self {
            Self {
                pulse: Box::new(&NoPulse),
            }
        }
    }
    let options = Options::default();
    options.pulse.step(3).unwrap();
    assert!(!options.pulse.share().unwrap().may_stop());
    assert_eq!(
        options.pulse.split(Execution::Sequence, &[]).err(),
        Some(PlanError::EmptyOrZeroWeight)
    );
    let shared: Arc<dyn Pulse> = Arc::new(&NoPulse);
    std::thread::scope(|scope| {
        scope.spawn(|| shared.step(1).unwrap());
    });
}

#[test]
fn wrappers_forward_stops_reports_splits_and_shares() {
    let flagged = Arc::new(Flagged::default());
    let wrapped: [Box<dyn Pulse>; 3] = [
        Box::new(Arc::clone(&flagged)),
        Box::new(Box::new(Arc::clone(&flagged))),
        Box::new(Arc::new(Arc::clone(&flagged))),
    ];
    for pulse in &wrapped {
        pulse.step(2).unwrap();
        assert_eq!(two_stages(&**pulse), Ok(2));
        let shared = pulse.share().unwrap();
        shared.advance(1);
        assert!(shared.may_stop());
    }
    assert_eq!(flagged.units.load(Ordering::Relaxed), 9);
    flagged.stopped.store(true, Ordering::Relaxed);
    for pulse in &wrapped {
        assert_eq!(pulse.check(), Err(StopReason::Cancelled));
        assert_eq!(pulse.share().unwrap().check(), Err(StopReason::Cancelled));
    }
}