zsui/
dispatch_queue.rs

1//! Bounded admission for application actions. Accepted actions retain FIFO order.
2use std::{collections::VecDeque, fmt};
3
4/// Default number of outstanding actions in each window queue.
5pub const DEFAULT_DISPATCH_QUEUE_CAPACITY: usize = 4096;
6/// Maximum actions of one kind dispatched in one event-loop turn.
7pub const DISPATCH_BATCH_SIZE: usize = 32;
8
9/// Admission failed; the action was not accepted and may be retried later.
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11pub struct QueueFull {
12    /// Queue that exhausted its remaining capacity.
13    pub queue: &'static str,
14    /// Remaining admission allowance for this update or queue.
15    pub capacity: usize,
16}
17impl fmt::Display for QueueFull {
18    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
19        write!(
20            f,
21            "{} queue is full (allowance {})",
22            self.queue, self.capacity
23        )
24    }
25}
26impl std::error::Error for QueueFull {}
27
28#[derive(Debug, Clone, Copy)]
29pub(crate) struct DispatchAllowance {
30    pub commands: usize,
31    pub ui_commands: usize,
32    pub effects: usize,
33}
34impl Default for DispatchAllowance {
35    fn default() -> Self {
36        Self {
37            commands: DEFAULT_DISPATCH_QUEUE_CAPACITY,
38            ui_commands: DEFAULT_DISPATCH_QUEUE_CAPACITY,
39            effects: DEFAULT_DISPATCH_QUEUE_CAPACITY,
40        }
41    }
42}
43
44#[derive(Debug, Clone)]
45pub(crate) struct DeferredQueue<T> {
46    pending: VecDeque<T>,
47    rejected: usize,
48}
49impl<T> Default for DeferredQueue<T> {
50    fn default() -> Self {
51        Self {
52            pending: VecDeque::new(),
53            rejected: 0,
54        }
55    }
56}
57impl<T> DeferredQueue<T> {
58    pub fn len(&self) -> usize {
59        self.pending.len()
60    }
61    pub fn is_empty(&self) -> bool {
62        self.pending.is_empty()
63    }
64    pub fn remaining(&self) -> usize {
65        DEFAULT_DISPATCH_QUEUE_CAPACITY.saturating_sub(self.len())
66    }
67    pub fn push(&mut self, value: T) {
68        if self.remaining() == 0 {
69            self.rejected = self.rejected.saturating_add(1);
70        } else {
71            self.pending.push_back(value);
72        }
73    }
74    pub fn extend(&mut self, values: impl IntoIterator<Item = T>) {
75        for value in values {
76            self.push(value);
77        }
78    }
79    pub fn pop(&mut self) -> Option<T> {
80        self.pending.pop_front()
81    }
82    pub fn take_batch(&mut self) -> Vec<T> {
83        self.pending
84            .drain(..self.len().min(DISPATCH_BATCH_SIZE))
85            .collect()
86    }
87    pub fn take_rejected(&mut self) -> usize {
88        std::mem::take(&mut self.rejected)
89    }
90    pub fn clear(&mut self) {
91        self.pending.clear();
92    }
93}
94
95/// A turn checks its deadline before removing the next request. Modal calls
96/// already in progress cannot be preempted; the remainder stays in the queue.
97#[allow(dead_code)] // Used by the optional native desktop hosts.
98pub(crate) struct DispatchTurnBudget {
99    start: std::time::Instant,
100    count: usize,
101}
102#[allow(dead_code)]
103impl DispatchTurnBudget {
104    pub fn new() -> Self {
105        Self {
106            start: std::time::Instant::now(),
107            count: 0,
108        }
109    }
110    pub fn take(&mut self) -> bool {
111        if self.count >= DISPATCH_BATCH_SIZE
112            || self.start.elapsed() >= std::time::Duration::from_millis(4)
113        {
114            return false;
115        }
116        self.count += 1;
117        true
118    }
119}
120
121#[cfg(test)]
122mod tests {
123    use super::*;
124    #[test]
125    fn deferred_dispatch_deadline_expires_without_popping_a_request() {
126        let mut queue = DeferredQueue::default();
127        queue.push(1);
128        let mut budget = DispatchTurnBudget {
129            start: std::time::Instant::now() - std::time::Duration::from_millis(10),
130            count: 0,
131        };
132        assert!(!budget.take());
133        assert_eq!(queue.pop(), Some(1));
134    }
135    #[test]
136    fn bounded_command_queue_reports_admission_failure_without_changing_fifo() {
137        let mut queue = crate::CommandQueue::with_capacity(2);
138        let command = |id| crate::UiCommand::window(crate::CommandId(id));
139        assert!(queue.try_push(command("first")).is_ok());
140        queue.push(command("second"));
141        assert!(queue.try_push(command("third")).is_err());
142        assert_eq!(queue.rejected_count(), 1);
143        assert_eq!(queue.pop().unwrap().id, crate::CommandId("first"));
144        assert_eq!(queue.pop().unwrap().id, crate::CommandId("second"));
145        assert!(queue.pop().is_none());
146    }
147}
zsui/dispatch_queue.rs

zsui/
dispatch_queue.rs
