loam_app/session/
commands.rs

1use std::sync::{Arc, Mutex, MutexGuard};
2
3use loam_runtime::{
4    AppCommand, Command, Dispatch, Outcome, Rejection, RequestId, RestoreError, Session, Stores,
5};
6
7pub(crate) const MAX_PENDING_COMMANDS: usize = 1024;
8const MAX_REPORTED_COMMANDS: usize = 256;
9
10struct Queued<A> {
11    name: &'static str,
12    command: Command<A>,
13    report: bool,
14}
15
16struct Submitted {
17    request: RequestId,
18    name: &'static str,
19    report: bool,
20    recovery: bool,
21}
22
23pub(crate) struct Reported {
24    pub name: &'static str,
25    pub outcome: Result<Outcome, Rejection>,
26}
27
28struct State<A> {
29    queued: Vec<Queued<A>>,
30    submitted: Vec<Submitted>,
31    reported: Vec<Reported>,
32    dropped_reports: usize,
33    in_flight: usize,
34    recovery_in_flight: bool,
35}
36
37impl<A> Default for State<A> {
38    fn default() -> Self {
39        Self {
40            queued: Vec::with_capacity(MAX_PENDING_COMMANDS),
41            submitted: Vec::with_capacity(MAX_PENDING_COMMANDS),
42            reported: Vec::with_capacity(MAX_REPORTED_COMMANDS),
43            dropped_reports: 0,
44            in_flight: 0,
45            recovery_in_flight: false,
46        }
47    }
48}
49
50type Shared<A> = Arc<Mutex<State<A>>>;
51
52fn lock<A>(shared: &Shared<A>) -> MutexGuard<'_, State<A>> {
53    shared.lock().unwrap_or_else(|error| error.into_inner())
54}
55
56pub struct CommandSender<A: Stores> {
57    shared: Shared<A>,
58    report: bool,
59}
60
61impl<A: Stores> Clone for CommandSender<A> {
62    fn clone(&self) -> Self {
63        Self {
64            shared: self.shared.clone(),
65            report: self.report,
66        }
67    }
68}
69
70impl<A: Stores> CommandSender<A> {
71    pub fn submit(&self, command: Command<A>) {
72        let name = command.name();
73        let recovery = matches!(&command, Command::Reset);
74        let mut shared = lock(&self.shared);
75        let pending = shared.queued.len() + shared.submitted.len() + shared.in_flight;
76        let rejected = if recovery {
77            pending >= MAX_PENDING_COMMANDS
78                || shared.recovery_in_flight
79                || shared
80                    .queued
81                    .iter()
82                    .any(|queued| matches!(&queued.command, Command::Reset))
83                || shared.submitted.iter().any(|submitted| submitted.recovery)
84        } else {
85            pending >= MAX_PENDING_COMMANDS - 1
86        };
87        if rejected {
88            deliver(&mut shared, name, self.report, Err(Rejection::Capacity));
89            return;
90        }
91        shared.queued.push(Queued {
92            name,
93            command,
94            report: self.report,
95        });
96    }
97
98    pub fn app(&self, command: impl AppCommand<A>) {
99        self.submit(Command::App(Box::new(command)));
100    }
101
102    pub fn app_fn(
103        &self,
104        name: &'static str,
105        mut apply: impl FnMut(&mut Dispatch<'_, A>) + Send + 'static,
106    ) {
107        self.submit(Command::try_app_fn(name, move |dispatch| {
108            apply(dispatch);
109            Ok(Outcome::Done)
110        }));
111    }
112
113    pub fn try_app_fn(
114        &self,
115        name: &'static str,
116        apply: impl FnMut(&mut Dispatch<'_, A>) -> Result<Outcome, Rejection> + Send + 'static,
117    ) {
118        self.submit(Command::try_app_fn(name, apply));
119    }
120
121    pub fn reset(&self) {
122        self.submit(Command::Reset);
123    }
124
125    pub(crate) fn reported(&self) -> Self {
126        Self {
127            shared: self.shared.clone(),
128            report: true,
129        }
130    }
131
132    pub(crate) fn take_reported(&self, into: &mut Vec<Reported>) -> usize {
133        let mut shared = lock(&self.shared);
134        into.append(&mut shared.reported);
135        std::mem::take(&mut shared.dropped_reports)
136    }
137}
138
139fn deliver<A>(
140    shared: &mut State<A>,
141    name: &'static str,
142    report: bool,
143    outcome: Result<Outcome, Rejection>,
144) {
145    if !report {
146        match &outcome {
147            Ok(_) => return,
148            Err(rejection) => tracing::warn!("{name}: rejected, {rejection}"),
149        }
150    }
151    if shared.reported.len() < MAX_REPORTED_COMMANDS {
152        shared.reported.push(Reported { name, outcome });
153    } else {
154        shared.dropped_reports = shared.dropped_reports.saturating_add(1);
155    }
156}
157
158pub(crate) struct CommandInbox<A: Stores> {
159    shared: Shared<A>,
160}
161
162impl<A: Stores> Default for CommandInbox<A> {
163    fn default() -> Self {
164        Self {
165            shared: Shared::default(),
166        }
167    }
168}
169
170impl<A: Stores> CommandInbox<A> {
171    pub fn sender(&self) -> CommandSender<A> {
172        CommandSender {
173            shared: self.shared.clone(),
174            report: false,
175        }
176    }
177
178    pub fn recover(&self, session: &mut Session<A>) -> Result<(), RestoreError> {
179        if session.faulted_phase().is_none() {
180            return Ok(());
181        }
182        let (mut batch, reset) = {
183            let mut shared = lock(&self.shared);
184            let Some(reset) = shared
185                .queued
186                .iter()
187                .position(|queued| matches!(&queued.command, Command::Reset))
188            else {
189                return Ok(());
190            };
191            let batch =
192                std::mem::replace(&mut shared.queued, Vec::with_capacity(MAX_PENDING_COMMANDS));
193            shared.in_flight += batch.len();
194            shared.recovery_in_flight = true;
195            (batch, reset)
196        };
197        let outcome = session.reset();
198        self.collect(session);
199        let mut shared = lock(&self.shared);
200        shared.in_flight -= batch.len();
201        shared.recovery_in_flight = false;
202        for (index, queued) in batch.drain(..).enumerate() {
203            let result = if index == reset {
204                outcome.map(|()| Outcome::Done).map_err(Rejection::Restore)
205            } else {
206                Err(Rejection::Cancelled)
207            };
208            deliver(&mut shared, queued.name, queued.report, result);
209        }
210        outcome
211    }
212
213    pub fn drain(&self, session: &mut Session<A>) {
214        let mut shared = lock(&self.shared);
215        let State {
216            queued, submitted, ..
217        } = &mut *shared;
218        for queued in queued.drain(..) {
219            let recovery = matches!(&queued.command, Command::Reset);
220            submitted.push(Submitted {
221                request: session.submit(queued.command),
222                name: queued.name,
223                report: queued.report,
224                recovery,
225            });
226        }
227    }
228
229    pub fn collect(&self, session: &Session<A>) {
230        let mut shared = lock(&self.shared);
231        for result in session.results() {
232            let Some(index) = shared
233                .submitted
234                .iter()
235                .position(|submitted| submitted.request == result.request)
236            else {
237                if result.outcome.is_err() {
238                    deliver(&mut shared, result.name, false, result.outcome);
239                }
240                continue;
241            };
242            let submitted = shared.submitted.swap_remove(index);
243            deliver(
244                &mut shared,
245                submitted.name,
246                submitted.report,
247                result.outcome,
248            );
249        }
250    }
251}
loam_app/session/commands.rs

loam_app/session/
commands.rs
