steel_core/command/queue/
pending.rs

1//! Start-of-tick queue for suspended command executions.
2
3use std::{collections::VecDeque, mem};
4
5use rustc_hash::FxHashMap;
6
7use super::super::{
8    execution::{
9        CommandExecutionContext, CommandSuspensionOrder, ExecutionCommandSource, ExecutionStop,
10    },
11    sender::{CommandExecutionOwner, CommandSenderKey},
12};
13
14/// Maximum retained command executions polled before new command requests in one tick.
15pub(crate) const COMMAND_RESUMPTIONS_PER_TICK: usize = 128;
16
17/// Suspended command executions owned by the server tick.
18pub(crate) struct PendingCommandExecutionQueue<S>
19where
20    S: ExecutionCommandSource,
21{
22    queued: VecDeque<PendingCommandExecution<S>>,
23    blocked_sources: FxHashMap<CommandSenderKey, usize>,
24    global_barriers: usize,
25}
26
27struct PendingCommandExecution<S>
28where
29    S: ExecutionCommandSource,
30{
31    owner: CommandExecutionOwner,
32    order: CommandSuspensionOrder,
33    execution: CommandExecutionContext<S>,
34}
35
36impl<S> PendingCommandExecutionQueue<S>
37where
38    S: ExecutionCommandSource,
39{
40    pub(crate) fn new() -> Self {
41        Self {
42            queued: VecDeque::new(),
43            blocked_sources: FxHashMap::default(),
44            global_barriers: 0,
45        }
46    }
47
48    /// Retains an execution only when it is waiting on suspended work.
49    #[must_use]
50    pub(crate) fn push_suspended(
51        &mut self,
52        owner: CommandExecutionOwner,
53        execution: CommandExecutionContext<S>,
54    ) -> bool {
55        let Some(order) = execution.suspension_order() else {
56            return false;
57        };
58        let source = owner.key();
59        self.retain_barrier(source, order);
60        self.queued.push_back(PendingCommandExecution {
61            owner,
62            order,
63            execution,
64        });
65        true
66    }
67
68    /// Returns whether a later top-level command from `source` must wait.
69    pub(crate) fn blocks(&self, source: CommandSenderKey) -> bool {
70        self.global_barriers > 0 || self.blocked_sources.contains_key(&source)
71    }
72
73    /// Polls each execution selected for this tick at most once, preserving FIFO order.
74    pub(crate) fn tick(
75        &mut self,
76        limit: usize,
77        mut owner_is_current: impl FnMut(&CommandExecutionOwner) -> bool,
78    ) -> PendingCommandExecutionStats {
79        let scheduled = self.queued.len().min(limit);
80        let mut polled = 0;
81        let mut finished = 0;
82
83        for _ in 0..scheduled {
84            let Some(mut pending) = self.queued.pop_front() else {
85                break;
86            };
87            polled += 1;
88            if !owner_is_current(&pending.owner) {
89                pending.execution.cancel();
90                self.release_barrier(pending.owner.key(), pending.order);
91                finished += 1;
92                continue;
93            }
94            match pending.execution.poll_suspension() {
95                ExecutionStop::Suspended => {
96                    let Some(order) = pending.execution.suspension_order() else {
97                        tracing::error!("suspended command lost its active suspension");
98                        self.release_barrier(pending.owner.key(), pending.order);
99                        finished += 1;
100                        continue;
101                    };
102                    if order != pending.order {
103                        self.release_barrier(pending.owner.key(), pending.order);
104                        self.retain_barrier(pending.owner.key(), order);
105                        pending.order = order;
106                    }
107                    self.queued.push_back(pending);
108                }
109                ExecutionStop::Completed
110                | ExecutionStop::CommandLimit
111                | ExecutionStop::QueueOverflow => {
112                    self.release_barrier(pending.owner.key(), pending.order);
113                    finished += 1;
114                }
115            }
116        }
117
118        PendingCommandExecutionStats {
119            polled,
120            finished,
121            pending: self.queued.len(),
122        }
123    }
124
125    pub(crate) fn cancel_all(&mut self) {
126        let executions = mem::take(&mut self.queued);
127        self.blocked_sources.clear();
128        self.global_barriers = 0;
129        for mut pending in executions {
130            pending.execution.cancel();
131        }
132    }
133
134    fn retain_barrier(&mut self, source: CommandSenderKey, order: CommandSuspensionOrder) {
135        *self.blocked_sources.entry(source).or_default() += 1;
136        if order == CommandSuspensionOrder::Global {
137            self.global_barriers += 1;
138        }
139    }
140
141    fn release_barrier(&mut self, source: CommandSenderKey, order: CommandSuspensionOrder) {
142        if let Some(count) = self.blocked_sources.get_mut(&source) {
143            *count -= 1;
144            if *count == 0 {
145                self.blocked_sources.remove(&source);
146            }
147        } else {
148            tracing::error!(?source, "pending command source barrier was not retained");
149        }
150        if order == CommandSuspensionOrder::Global {
151            self.global_barriers = self.global_barriers.saturating_sub(1);
152        }
153    }
154
155    #[cfg(test)]
156    pub(crate) fn len(&self) -> usize {
157        self.queued.len()
158    }
159}
160
161impl<S> Default for PendingCommandExecutionQueue<S>
162where
163    S: ExecutionCommandSource,
164{
165    fn default() -> Self {
166        Self::new()
167    }
168}
169
170#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
171pub(crate) struct PendingCommandExecutionStats {
172    pub(crate) polled: usize,
173    pub(crate) finished: usize,
174    pub(crate) pending: usize,
175}
steel_core/command/queue/pending.rs

steel_core/command/queue/
pending.rs
