rsynko_manager/
attention.rs

1use crate::*;
2use alux_ext::ext;
3use rsynko_session::Wanted;
4use std::path::PathBuf;
5use std::time::Duration;
6
7/// States how much of a failure a reader is told about at a glance.
8const SUMMARY_COLUMNS: usize = 160;
9
10/// States how much of a failure is kept for whoever goes looking.
11const DETAIL_COLUMNS: usize = 500;
12
13/// Derives what beginning, running, refusing, and finishing one run states to the collection.
14///
15/// A run is attended to by what the request wants of it, and a request wanting nothing is one
16/// that was removed, or one whose reader is leaving. Which kind of run a request asks for is
17/// something the request already states, so nobody attending to it has to carry that around.
18#[ext(name = ManagerAttentionExt)]
19pub impl<This> This
20where
21    This: QueueCatalogAlg + SafeExitAlg + TransferStateAlg + RehearsalStateAlg,
22    This::Entry: QueueEntryAlg,
23    This::Change: PlannedChangeAlg,
24    This::Id: Copy + Eq,
25{
26    /// States every request waiting for an interpreter to do what it asks, in collection order.
27    fn wanting_work(&self) -> Vec<This::Id> {
28        self.queue_ids().filter(|id| self.entry_phase(*id) == Some(TransferPhase::Waiting)).collect()
29    }
30
31    /// States what one request wants of the run working on its behalf.
32    fn wanted(&self, id: This::Id) -> Wanted {
33        // Leaving is wanting nothing: every run ends the way a removed request's run ends.
34        if self.exit_requested() {
35            return Wanted::Unwanted;
36        }
37        match self.entry_phase(id) {
38            None => Wanted::Unwanted,
39            Some(TransferPhase::Paused) => Wanted::Held,
40            Some(_) => Wanted::Running,
41        }
42    }
43
44    /// States whether one request asks for a rehearsal rather than for the thing itself.
45    fn rehearses(&self, id: This::Id) -> bool {
46        self.queue_entry(id).and_then(QueueEntryAlg::dry_run) == Some(true)
47    }
48
49    /// States that a run has begun, as the kind of run the request asks for.
50    ///
51    /// A rehearsal moves nothing and is over when it has said what it would do, so it is never
52    /// told that it could be held still.
53    fn begun(&mut self, id: This::Id, holdable: bool) {
54        if self.rehearses(id) {
55            self.apply_rehearsal_event(id, RehearsalObservationOp::Started {});
56            return;
57        }
58        self.apply_transfer_event(id, TransferObservationOp::Started {});
59        self.apply_transfer_event(id, TransferObservationOp::PauseCapability { supported: holdable });
60    }
61
62    /// States how long one run has been running, when the request counts that.
63    ///
64    /// A rehearsal moves nothing, so how long it took states nothing about a transfer.
65    fn ran_for(&mut self, id: This::Id, elapsed: Duration) {
66        if !self.rehearses(id) {
67            self.apply_transfer_event(id, TransferObservationOp::Elapsed { elapsed });
68        }
69    }
70
71    /// States that one run refused, as the kind of run the request asks for.
72    fn refused(&mut self, id: This::Id, detail: &str) {
73        if self.rehearses(id) {
74            self.apply_rehearsal_event(id, RehearsalObservationOp::Failed { message: summary_line(detail) });
75            return;
76        }
77        self.apply_transfer_event(
78            id,
79            TransferObservationOp::ProgramFailed { summary: summary_line(detail), detail: detail_text(detail) },
80        );
81    }
82
83    /// States what one finished run did, as the kind of run the request asks for.
84    ///
85    /// A rehearsal states what it would have changed. A transfer states that it is done, and how
86    /// much it moved, which is what its changes add up to.
87    fn performed(&mut self, id: This::Id, destination: PathBuf, changes: Vec<This::Change>) {
88        if self.rehearses(id) {
89            self.apply_rehearsal_event(id, RehearsalObservationOp::Reported { changes });
90            return;
91        }
92        let bytes = changes.iter().filter_map(PlannedChangeAlg::change_size).fold(0_u64, u64::saturating_add);
93        self.apply_transfer_event(id, TransferObservationOp::Completed { destination, bytes });
94    }
95
96    /// States whether another request is already writing where this one would write.
97    ///
98    /// Two runs writing one path would each publish over the other, and which of them a reader
99    /// ends up with is decided by whichever finished last. The collection knows both, so it is
100    /// the collection that states the clash rather than whoever happens to be running them.
101    fn destination_is_taken(&self, id: This::Id) -> bool {
102        let Some(destination) = self.queue_entry(id).and_then(QueueEntryAlg::output) else {
103            return false;
104        };
105        self.queue_ids()
106            .filter(|other| *other != id)
107            .filter(|other| self.entry_phase(*other).is_some_and(TransferPhase::is_running))
108            .any(|other| self.queue_entry(other).and_then(QueueEntryAlg::output) == Some(destination))
109    }
110
111    /// Observes the phase one request has reached, while the collection still holds it.
112    fn entry_phase(&self, id: This::Id) -> Option<TransferPhase> {
113        self.queue_entry(id).map(QueueEntryAlg::phase)
114    }
115}
116
117/// States the one line a reader is shown about a failure.
118pub fn summary_line(detail: &str) -> String {
119    detail.lines().next().unwrap_or_default().chars().take(SUMMARY_COLUMNS).collect()
120}
121
122/// States what is kept about a failure for whoever goes looking for it.
123pub fn detail_text(detail: &str) -> String {
124    detail.chars().take(DETAIL_COLUMNS).collect()
125}
rsynko_manager/attention.rs

rsynko_manager/
attention.rs
