ailloli_ui_runtime/app/runtime_inbox.rs
1//! Bounded cross-thread inbox for invalidations and runtime service work.
2
3use std::collections::HashSet;
4use std::fmt;
5use std::num::NonZeroUsize;
6use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
7use std::sync::mpsc::{self, Receiver, SyncSender, TryRecvError, TrySendError};
8use std::sync::{Arc, Mutex};
9
10use ailloli_ui_core::LogicalWindowId;
11
12use super::ui_inbox::{UiWake, UiWakeError};
13use super::RuntimeHandle;
14
15/// Maximum number of mailbox messages applied during one host callback.
16///
17/// A bounded drain prevents unbounded producer traffic from monopolizing the UI
18/// thread. The 257th observed message is prefetched and retained for the next
19/// callback.
20///
21/// # Examples
22///
23/// ```
24/// use ailloli_ui_runtime::app::RUNTIME_INBOX_DRAIN_BUDGET;
25/// assert_eq!(RUNTIME_INBOX_DRAIN_BUDGET, 256);
26/// ```
27pub const RUNTIME_INBOX_DRAIN_BUDGET: usize = 256;
28
29/// Result of attempting to enqueue one runtime message.
30///
31/// `EnqueuedButWakeFailed` means the message is already owned by the inbox; do
32/// not resend it as recovery because that can duplicate actions.
33///
34/// # Examples
35///
36/// ```
37/// use ailloli_ui_runtime::app::{RuntimeSendError, UiWakeError};
38/// assert!(matches!(RuntimeSendError::EnqueuedButWakeFailed(UiWakeError::TemporarilyUnavailable), RuntimeSendError::EnqueuedButWakeFailed(_)));
39/// ```
40#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
41pub enum RuntimeSendError {
42 /// Bounded queue is full; the message was not enqueued.
43 #[error("the runtime inbox is full")]
44 Full,
45 /// Consumer was dropped; the message was not enqueued.
46 #[error("the runtime inbox is closed")]
47 Closed,
48 /// Message was queued, but host notification failed.
49 #[error("the message was enqueued, but waking the UI host failed: {0}")]
50 EnqueuedButWakeFailed(UiWakeError),
51}
52
53/// Internal provider-neutral message variants crossing the thread boundary.
54#[derive(Debug)]
55enum RuntimeMessage<A> {
56 /// Provider-neutral application action delivered in channel order.
57 Action(A),
58 /// Coalescible request to redraw every logical window.
59 RedrawAll,
60 /// Coalescible request to redraw one logical window.
61 RedrawWindow(LogicalWindowId),
62}
63
64/// Coalesced host-notification state.
65#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
66enum WakeStatus {
67 #[default]
68 /// No queued work currently requires a host notification.
69 Idle,
70 /// Work is pending and the next producer or installer must notify the host.
71 NeedsWake,
72 /// Callback invocation `u64` is executing outside the state mutex.
73 InFlight(u64),
74 /// A callback succeeded for the currently pending work.
75 Signaled,
76}
77
78/// Installed wake callback and wrapping attempt sequence.
79struct WakeState {
80 /// Optional thread-safe host notification callback.
81 wake: Option<Arc<dyn UiWake>>,
82 /// Coalesced notification lifecycle for queued work.
83 status: WakeStatus,
84 /// Wrapping nonzero-preferred identity reserved for the next invocation.
85 next_attempt_id: u64,
86}
87
88/// Implements the Default contract for WakeState.
89impl Default for WakeState {
90 /// Constructs the documented default value.
91 fn default() -> Self {
92 Self {
93 wake: None,
94 status: WakeStatus::Idle,
95 next_attempt_id: 1,
96 }
97 }
98}
99
100/// Provides the operations defined for WakeState.
101impl WakeState {
102 /// Reserves a wrapping attempt ID when a callback exists.
103 fn start_attempt(&mut self) -> Option<(u64, Arc<dyn UiWake>)> {
104 let wake = self.wake.clone()?;
105 let attempt_id = self.next_attempt_id;
106 self.next_attempt_id = self.next_attempt_id.wrapping_add(1);
107 self.status = WakeStatus::InFlight(attempt_id);
108 Some((attempt_id, wake))
109 }
110}
111
112/// Atomic mailbox counters shared across producers and the UI consumer.
113#[derive(Default)]
114struct MailboxStats {
115 /// Successfully enqueued action messages.
116 enqueued: AtomicU64,
117 /// Action messages returned to the UI consumer.
118 drained: AtomicU64,
119 /// Redraw requests merged into an already pending request.
120 coalesced: AtomicU64,
121 /// Action sends rejected because the bounded channel was full.
122 overflow: AtomicU64,
123 /// Sends or receives that observed the opposite channel endpoint closed.
124 disconnected: AtomicU64,
125 /// Host wake callback invocations attempted.
126 wake_calls: AtomicU64,
127 /// Host wake callbacks that returned an error.
128 wake_failures: AtomicU64,
129 /// Best-effort queued or prefetched message depth.
130 current_depth: AtomicUsize,
131 /// Maximum best-effort depth observed since channel creation.
132 max_depth: AtomicUsize,
133}
134
135/// Read-only instrumentation snapshot for a runtime mailbox.
136///
137/// Counts are cumulative wrapping atomic counters and fields are loaded
138/// independently, so concurrent snapshots are not transactional. For action
139/// messages, queue publication currently precedes depth accounting; a consumer
140/// racing in that interval can make `current_depth`/`max_depth` wrap and they
141/// should therefore be treated as diagnostics, not protocol state.
142///
143/// # Examples
144///
145/// ```
146/// use ailloli_ui_runtime::app::RuntimeInboxStats;
147/// assert_eq!(RuntimeInboxStats::default().coalesced, 0);
148/// ```
149#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
150pub struct RuntimeInboxStats {
151 /// Successfully queued non-coalesced messages.
152 pub enqueued: u64,
153 /// Messages applied to a runtime.
154 pub drained: u64,
155 /// Duplicate outstanding redraw requests that did not add a message.
156 pub coalesced: u64,
157 /// Sends rejected because the bounded queue was full.
158 pub overflow: u64,
159 /// First observed channel disconnection (zero or one).
160 pub disconnected: u64,
161 /// Wake callback invocations.
162 pub wake_calls: u64,
163 /// Wake callback errors.
164 pub wake_failures: u64,
165 /// Best-effort queued/pending depth; see the type-level concurrency caveat.
166 pub current_depth: usize,
167 /// Maximum best-effort observed depth.
168 pub max_depth: usize,
169}
170
171/// Shared wake, redraw-coalescing, disconnection, and counter state.
172struct MailboxShared {
173 /// Installed callback and coalesced wake lifecycle protected from producers.
174 wake: Mutex<WakeState>,
175 /// Whether an all-window redraw is waiting for the consumer.
176 pending_redraw_all: Mutex<bool>,
177 /// Logical windows with a pending targeted redraw request.
178 pending_redraw_windows: Mutex<HashSet<LogicalWindowId>>,
179 /// Whether any operation has observed channel disconnection.
180 disconnected_observed: AtomicBool,
181 /// Shared wrapping instrumentation counters.
182 stats: MailboxStats,
183}
184
185/// Provides the operations defined for MailboxShared.
186impl MailboxShared {
187 /// Creates idle shared state with empty redraw sets and zero counters.
188 fn new() -> Self {
189 Self {
190 wake: Mutex::new(WakeState::default()),
191 pending_redraw_all: Mutex::new(false),
192 pending_redraw_windows: Mutex::new(HashSet::new()),
193 disconnected_observed: AtomicBool::new(false),
194 stats: MailboxStats::default(),
195 }
196 }
197
198 /// Records a successful enqueue and best-effort depth high-water mark.
199 fn message_enqueued(&self) {
200 self.stats.enqueued.fetch_add(1, Ordering::Relaxed);
201 let depth = self.stats.current_depth.fetch_add(1, Ordering::AcqRel) + 1;
202 let mut current_max = self.stats.max_depth.load(Ordering::Relaxed);
203 while depth > current_max {
204 match self.stats.max_depth.compare_exchange_weak(
205 current_max,
206 depth,
207 Ordering::Relaxed,
208 Ordering::Relaxed,
209 ) {
210 Ok(_) => break,
211 Err(observed) => current_max = observed,
212 }
213 }
214 }
215
216 /// Records one applied message and decrements best-effort depth.
217 fn message_drained(&self) {
218 self.stats.drained.fetch_add(1, Ordering::Relaxed);
219 self.stats.current_depth.fetch_sub(1, Ordering::AcqRel);
220 }
221
222 /// Records channel disconnection at most once.
223 fn observe_disconnected(&self) {
224 if !self.disconnected_observed.swap(true, Ordering::AcqRel) {
225 self.stats.disconnected.fetch_add(1, Ordering::Relaxed);
226 }
227 }
228
229 /// Coalesces or starts a host wake attempt.
230 ///
231 /// # Errors
232 ///
233 /// Returns the installed [`UiWake`] callback's error when this call starts
234 /// an attempt. Coalesced requests and requests without a callback succeed.
235 fn request_wake(&self) -> Result<(), UiWakeError> {
236 let attempt = {
237 let mut state = self.wake.lock().unwrap_or_else(|error| error.into_inner());
238 match state.status {
239 WakeStatus::Signaled | WakeStatus::InFlight(_) => return Ok(()),
240 WakeStatus::Idle | WakeStatus::NeedsWake => {
241 let attempt = state.start_attempt();
242 if attempt.is_none() {
243 state.status = WakeStatus::NeedsWake;
244 }
245 attempt
246 }
247 }
248 };
249 match attempt {
250 Some((attempt_id, wake)) => self.invoke_wake_attempt(attempt_id, wake),
251 None => Ok(()),
252 }
253 }
254
255 /// Invokes a wake outside the mutex and commits only a matching attempt.
256 ///
257 /// # Errors
258 ///
259 /// Returns the exact error produced by [`UiWake::wake`]. The wake remains
260 /// required so a later callback installation or send can retry it.
261 fn invoke_wake_attempt(
262 &self,
263 attempt_id: u64,
264 wake: Arc<dyn UiWake>,
265 ) -> Result<(), UiWakeError> {
266 self.stats.wake_calls.fetch_add(1, Ordering::Relaxed);
267 let result = wake.wake();
268 if result.is_err() {
269 self.stats.wake_failures.fetch_add(1, Ordering::Relaxed);
270 }
271 let mut state = self.wake.lock().unwrap_or_else(|error| error.into_inner());
272 if state.status == WakeStatus::InFlight(attempt_id) {
273 state.status = if result.is_ok() {
274 WakeStatus::Signaled
275 } else {
276 WakeStatus::NeedsWake
277 };
278 }
279 result
280 }
281
282 /// Reopens the wake slot before inspecting the receive queue.
283 fn begin_drain(&self) {
284 // Clear the delivered (or in-flight) signal before looking at the
285 // queue. An enqueue racing with the last `try_recv` can therefore
286 // reserve and deliver a fresh wake. An older in-flight result is
287 // ignored because its attempt id no longer matches the state.
288 self.wake
289 .lock()
290 .unwrap_or_else(|error| error.into_inner())
291 .status = WakeStatus::Idle;
292 }
293
294 /// Replaces the callback and immediately retries any outstanding wake.
295 ///
296 /// # Errors
297 ///
298 /// Returns the newly installed callback's [`UiWake::wake`] error when an
299 /// outstanding notification triggers an immediate attempt.
300 fn install_wake(&self, wake: Arc<dyn UiWake>) -> Result<(), UiWakeError> {
301 let attempt = {
302 let mut state = self.wake.lock().unwrap_or_else(|error| error.into_inner());
303 state.wake = Some(wake);
304 match state.status {
305 WakeStatus::Idle => None,
306 WakeStatus::NeedsWake | WakeStatus::InFlight(_) | WakeStatus::Signaled => {
307 state.start_attempt()
308 }
309 }
310 };
311 if let Some((attempt_id, wake)) = attempt {
312 self.invoke_wake_attempt(attempt_id, wake)?;
313 }
314 Ok(())
315 }
316
317 /// Removes the callback while preserving outstanding notification state.
318 fn detach_wake(&self) {
319 let mut state = self.wake.lock().unwrap_or_else(|error| error.into_inner());
320 state.wake = None;
321 if state.status != WakeStatus::Idle {
322 state.status = WakeStatus::NeedsWake;
323 }
324 }
325
326 /// Loads a non-resetting, non-transactional counter snapshot.
327 fn stats(&self) -> RuntimeInboxStats {
328 RuntimeInboxStats {
329 enqueued: self.stats.enqueued.load(Ordering::Relaxed),
330 drained: self.stats.drained.load(Ordering::Relaxed),
331 coalesced: self.stats.coalesced.load(Ordering::Relaxed),
332 overflow: self.stats.overflow.load(Ordering::Relaxed),
333 disconnected: self.stats.disconnected.load(Ordering::Relaxed),
334 wake_calls: self.stats.wake_calls.load(Ordering::Relaxed),
335 wake_failures: self.stats.wake_failures.load(Ordering::Relaxed),
336 current_depth: self.stats.current_depth.load(Ordering::Relaxed),
337 max_depth: self.stats.max_depth.load(Ordering::Relaxed),
338 }
339 }
340}
341
342/// Cloneable producer for a bounded runtime mailbox.
343///
344/// Clones share queue, redraw-coalescing state, wake state, and counters.
345/// `RuntimeSender<A>` is suitable across threads when `A: Send`; it never
346/// accesses the UI-local `RuntimeHandle` from producers.
347///
348/// # Examples
349///
350/// ```
351/// use std::num::NonZeroUsize;
352/// use ailloli_ui_runtime::app::RuntimeInbox;
353/// let (sender, _inbox) = RuntimeInbox::<String>::channel(NonZeroUsize::new(4).unwrap());
354/// assert_eq!(sender.stats().enqueued, 0);
355/// ```
356pub struct RuntimeSender<A> {
357 /// Bounded MPSC producer for non-coalesced action messages.
358 sender: SyncSender<RuntimeMessage<A>>,
359 /// Shared redraw, wake, disconnect, and instrumentation state.
360 shared: Arc<MailboxShared>,
361}
362
363/// Implements the `Clone` contract for `RuntimeSender<A>`.
364impl<A> Clone for RuntimeSender<A> {
365 /// Produces the clone required by the standard cloning contract.
366 fn clone(&self) -> Self {
367 Self {
368 sender: self.sender.clone(),
369 shared: self.shared.clone(),
370 }
371 }
372}
373
374/// Implements the `fmt::Debug` contract for `RuntimeSender<A>`.
375impl<A> fmt::Debug for RuntimeSender<A> {
376 /// Formats the value for human-readable diagnostics.
377 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
378 formatter
379 .debug_struct("RuntimeSender")
380 .field("stats", &self.stats())
381 .finish_non_exhaustive()
382 }
383}
384
385/// Provides the operations defined for `RuntimeSender<A>`.
386impl<A> RuntimeSender<A> {
387 /// Enqueues one application action without accessing [`RuntimeHandle`]
388 /// across threads.
389 ///
390 /// The operation is nonblocking. `EnqueuedButWakeFailed` leaves the action
391 /// queued; `Full` and `Closed` do not. Actions are not coalesced.
392 ///
393 /// # Errors
394 ///
395 /// Returns [`RuntimeSendError::Full`] or [`RuntimeSendError::Closed`] when
396 /// the action is not enqueued. [`RuntimeSendError::EnqueuedButWakeFailed`]
397 /// means the action remains queued but the host wake failed.
398 ///
399 /// # Examples
400 ///
401 /// ```
402 /// use std::num::NonZeroUsize;
403 /// use ailloli_ui_runtime::app::{RuntimeInbox, RuntimeSendError};
404 /// let (sender, _inbox) = RuntimeInbox::channel(NonZeroUsize::new(1).unwrap());
405 /// sender.dispatch(1).unwrap();
406 /// assert_eq!(sender.dispatch(2), Err(RuntimeSendError::Full));
407 /// ```
408 pub fn dispatch(&self, action: A) -> Result<(), RuntimeSendError> {
409 self.send(RuntimeMessage::Action(action))
410 }
411
412 /// Requests a redraw of every logical window.
413 ///
414 /// At most one outstanding `RedrawAll` message is queued. Repeated calls
415 /// increment `coalesced` and retry host wake without consuming capacity.
416 /// Per-window redraw messages remain independent.
417 ///
418 /// # Errors
419 ///
420 /// Returns [`RuntimeSendError::Full`] or [`RuntimeSendError::Closed`] when a
421 /// new redraw message cannot be enqueued. A coalesced or newly enqueued
422 /// request returns [`RuntimeSendError::EnqueuedButWakeFailed`] if waking the
423 /// host fails; the pending redraw marker remains set in that case.
424 ///
425 /// # Examples
426 ///
427 /// ```
428 /// use std::num::NonZeroUsize;
429 /// use ailloli_ui_runtime::app::RuntimeInbox;
430 /// let (sender, _inbox) = RuntimeInbox::<()>::channel(NonZeroUsize::new(2).unwrap());
431 /// sender.request_redraw().unwrap();
432 /// sender.request_redraw().unwrap();
433 /// assert_eq!((sender.stats().enqueued, sender.stats().coalesced), (1, 1));
434 /// ```
435 pub fn request_redraw(&self) -> Result<(), RuntimeSendError> {
436 let mut pending = self
437 .shared
438 .pending_redraw_all
439 .lock()
440 .unwrap_or_else(|error| error.into_inner());
441 if *pending {
442 self.shared.stats.coalesced.fetch_add(1, Ordering::Relaxed);
443 drop(pending);
444 return self.request_host_wake();
445 }
446 *pending = true;
447 let result = self.try_enqueue(RuntimeMessage::RedrawAll);
448 if result.is_err() {
449 *pending = false;
450 }
451 drop(pending);
452 result?;
453 self.request_host_wake()
454 }
455
456 /// Requests a redraw of one stable logical window.
457 ///
458 /// At most one outstanding message per exact logical ID is queued. Duplicate
459 /// requests increment `coalesced`; different windows consume separate
460 /// capacity. Failed enqueue removes the pending marker so retry can enqueue.
461 ///
462 /// # Errors
463 ///
464 /// Returns [`RuntimeSendError::Full`] or [`RuntimeSendError::Closed`] when a
465 /// new window redraw cannot be enqueued. A coalesced or newly enqueued
466 /// request returns [`RuntimeSendError::EnqueuedButWakeFailed`] if the host
467 /// wake fails; the pending marker remains set after that wake-only failure.
468 ///
469 /// # Examples
470 ///
471 /// ```
472 /// use std::num::NonZeroUsize;
473 /// use ailloli_ui_runtime::app::RuntimeInbox;
474 /// let (sender, _inbox) = RuntimeInbox::<()>::channel(NonZeroUsize::new(2).unwrap());
475 /// sender.request_window_redraw("main").unwrap();
476 /// sender.request_window_redraw("main").unwrap();
477 /// assert_eq!(sender.stats().coalesced, 1);
478 /// ```
479 pub fn request_window_redraw(
480 &self,
481 logical_window_id: impl Into<LogicalWindowId>,
482 ) -> Result<(), RuntimeSendError> {
483 let logical_window_id = logical_window_id.into();
484 let mut pending = self
485 .shared
486 .pending_redraw_windows
487 .lock()
488 .unwrap_or_else(|error| error.into_inner());
489 if !pending.insert(logical_window_id.clone()) {
490 self.shared.stats.coalesced.fetch_add(1, Ordering::Relaxed);
491 drop(pending);
492 return self.request_host_wake();
493 }
494 let result = self.try_enqueue(RuntimeMessage::RedrawWindow(logical_window_id.clone()));
495 if result.is_err() {
496 pending.remove(&logical_window_id);
497 }
498 drop(pending);
499 result?;
500 self.request_host_wake()
501 }
502
503 /// Returns a non-resetting shared instrumentation snapshot.
504 ///
505 /// # Examples
506 ///
507 /// ```
508 /// use std::num::NonZeroUsize;
509 /// use ailloli_ui_runtime::app::RuntimeInbox;
510 /// let (sender, _inbox) = RuntimeInbox::channel(NonZeroUsize::new(1).unwrap());
511 /// sender.dispatch(3).unwrap();
512 /// assert_eq!(sender.stats().enqueued, 1);
513 /// ```
514 pub fn stats(&self) -> RuntimeInboxStats {
515 self.shared.stats()
516 }
517
518 /// Publishes a non-coalesced message then requests a host wake.
519 ///
520 /// # Errors
521 ///
522 /// Returns [`RuntimeSendError::Full`] or [`RuntimeSendError::Closed`] when
523 /// enqueueing fails, or [`RuntimeSendError::EnqueuedButWakeFailed`] after a
524 /// successful enqueue whose host wake fails.
525 fn send(&self, message: RuntimeMessage<A>) -> Result<(), RuntimeSendError> {
526 self.try_enqueue(message)?;
527 self.request_host_wake()
528 }
529
530 /// Maps a host wake failure without changing queue ownership.
531 ///
532 /// # Errors
533 ///
534 /// Returns [`RuntimeSendError::EnqueuedButWakeFailed`] with the underlying
535 /// [`UiWakeError`] when the host callback rejects the notification.
536 fn request_host_wake(&self) -> Result<(), RuntimeSendError> {
537 self.shared
538 .request_wake()
539 .map_err(RuntimeSendError::EnqueuedButWakeFailed)
540 }
541
542 /// Attempts bounded publication and updates send-side diagnostics.
543 ///
544 /// # Errors
545 ///
546 /// Returns [`RuntimeSendError::Full`] when the bounded queue is at capacity,
547 /// or [`RuntimeSendError::Closed`] after the receiver disconnects.
548 fn try_enqueue(&self, message: RuntimeMessage<A>) -> Result<(), RuntimeSendError> {
549 match self.sender.try_send(message) {
550 Ok(()) => {
551 self.shared.message_enqueued();
552 Ok(())
553 }
554 Err(TrySendError::Full(_)) => {
555 self.shared.stats.overflow.fetch_add(1, Ordering::Relaxed);
556 Err(RuntimeSendError::Full)
557 }
558 Err(TrySendError::Disconnected(_)) => {
559 self.shared.observe_disconnected();
560 Err(RuntimeSendError::Closed)
561 }
562 }
563 }
564}
565
566/// Result of applying one bounded mailbox batch to a UI-local runtime.
567///
568/// Actions are already appended to the runtime when returned. Redraw flags are
569/// host intents, not runtime invalidations. `remaining` indicates one prefetched
570/// message is retained and another wake was requested.
571///
572/// # Examples
573///
574/// ```
575/// use ailloli_ui_runtime::app::RuntimeDrain;
576/// let drain = RuntimeDrain { drained_messages: 2, dispatched_actions: 1,
577/// redraw_all: true, redraw_windows: vec![], remaining: false };
578/// assert_eq!(drain.dispatched_actions, 1);
579/// ```
580#[derive(Debug, Clone, Default, PartialEq, Eq)]
581pub struct RuntimeDrain {
582 /// Total messages applied in this batch, at most 256.
583 pub drained_messages: usize,
584 /// Action messages appended to the runtime action queue.
585 pub dispatched_actions: usize,
586 /// Whether at least one global-redraw message was applied.
587 pub redraw_all: bool,
588 /// Logical windows requested for redraw, in applied message order.
589 pub redraw_windows: Vec<LogicalWindowId>,
590 /// Whether a prefetched 257th message remains pending.
591 pub remaining: bool,
592}
593
594/// UI-thread consumer for a bounded runtime mailbox.
595///
596/// The receiver and target [`RuntimeHandle`] remain UI-local. Dropping the
597/// inbox closes all sender clones and drops buffered actions.
598///
599/// # Examples
600///
601/// ```
602/// use std::num::NonZeroUsize;
603/// use ailloli_ui_runtime::app::RuntimeInbox;
604/// let (_sender, inbox) = RuntimeInbox::<()>::channel(NonZeroUsize::new(2).unwrap());
605/// assert_eq!(inbox.stats().drained, 0);
606/// ```
607pub struct RuntimeInbox<A> {
608 /// Single consumer for ordered action messages.
609 receiver: Receiver<RuntimeMessage<A>>,
610 /// Shared redraw, wake, disconnect, and instrumentation state.
611 shared: Arc<MailboxShared>,
612 /// One prefetched message retained when a bounded drain stops early.
613 pending: Option<RuntimeMessage<A>>,
614}
615
616/// Implements the `fmt::Debug` contract for `RuntimeInbox<A>`.
617impl<A> fmt::Debug for RuntimeInbox<A> {
618 /// Formats the value for human-readable diagnostics.
619 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
620 formatter
621 .debug_struct("RuntimeInbox")
622 .field("has_pending_message", &self.pending.is_some())
623 .field("stats", &self.stats())
624 .finish_non_exhaustive()
625 }
626}
627
628/// Provides the operations defined for `RuntimeInbox<A>`.
629impl<A> RuntimeInbox<A> {
630 /// Creates a bounded mailbox with exact nonzero message capacity.
631 ///
632 /// No wake callback is initially installed; messages sent before late
633 /// installation retain a pending wake requirement.
634 ///
635 /// # Examples
636 ///
637 /// ```
638 /// use std::num::NonZeroUsize;
639 /// use ailloli_ui_runtime::app::{RuntimeHandle, RuntimeInbox};
640 /// let (sender, mut inbox) = RuntimeInbox::channel(NonZeroUsize::new(2).unwrap());
641 /// sender.dispatch(8).unwrap();
642 /// let runtime = RuntimeHandle::new();
643 /// assert_eq!(inbox.drain(&runtime).unwrap().dispatched_actions, 1);
644 /// assert_eq!(runtime.take_actions(), [8]);
645 /// ```
646 pub fn channel(capacity: NonZeroUsize) -> (RuntimeSender<A>, Self) {
647 let (sender, receiver) = mpsc::sync_channel(capacity.get());
648 let shared = Arc::new(MailboxShared::new());
649 (
650 RuntimeSender {
651 sender,
652 shared: shared.clone(),
653 },
654 Self {
655 receiver,
656 shared,
657 pending: None,
658 },
659 )
660 }
661
662 /// Installs or replaces the host wake callback. Messages queued before this
663 /// call cause the newly installed callback to run immediately.
664 ///
665 /// Callback invocation is synchronous and outside the wake mutex. On error,
666 /// the callback remains installed and the wake remains retryable.
667 ///
668 /// # Errors
669 ///
670 /// Returns the callback's [`UiWakeError`] when delivering an already
671 /// outstanding notification fails.
672 ///
673 /// # Examples
674 ///
675 /// ```
676 /// use std::num::NonZeroUsize; use std::sync::Arc;
677 /// use ailloli_ui_runtime::app::{RuntimeInbox, UiWake, UiWakeError};
678 /// struct Wake; impl UiWake for Wake { fn wake(&self)->Result<(),UiWakeError>{Ok(())} }
679 /// let (_sender, inbox) = RuntimeInbox::<()>::channel(NonZeroUsize::new(1).unwrap());
680 /// assert!(inbox.install_wake(Arc::new(Wake)).is_ok());
681 /// ```
682 pub fn install_wake(&self, wake: Arc<dyn UiWake>) -> Result<(), UiWakeError> {
683 self.shared.install_wake(wake)
684 }
685
686 /// Detaches the current host callback while retaining any outstanding wake
687 /// requirement for the next installed callback.
688 ///
689 /// # Examples
690 ///
691 /// ```
692 /// use std::num::NonZeroUsize;
693 /// use ailloli_ui_runtime::app::RuntimeInbox;
694 /// let (_sender, inbox) = RuntimeInbox::<()>::channel(NonZeroUsize::new(1).unwrap());
695 /// inbox.detach_wake();
696 /// assert_eq!(inbox.stats().wake_calls, 0);
697 /// ```
698 pub fn detach_wake(&self) {
699 self.shared.detach_wake();
700 }
701
702 /// Returns a non-resetting shared instrumentation snapshot.
703 ///
704 /// # Examples
705 ///
706 /// ```
707 /// use std::num::NonZeroUsize;
708 /// use ailloli_ui_runtime::app::RuntimeInbox;
709 /// let (_sender, inbox) = RuntimeInbox::<()>::channel(NonZeroUsize::new(1).unwrap());
710 /// assert_eq!(inbox.stats(), Default::default());
711 /// ```
712 pub fn stats(&self) -> RuntimeInboxStats {
713 self.shared.stats()
714 }
715
716 /// Applies at most [`RUNTIME_INBOX_DRAIN_BUDGET`] messages to the UI-local
717 /// runtime. When work remains, another host wake is requested.
718 ///
719 /// Actions preserve FIFO order. Redraw pending markers are cleared as their
720 /// messages are applied. At the budget, one extra message is prefetched to
721 /// set `remaining`; it stays counted in depth until the next drain.
722 ///
723 /// # Errors
724 ///
725 /// Returns a wake error only when a remainder needs another host signal.
726 /// By then the first 256 messages have already been applied and their result
727 /// is not returned, so hosts should treat this failure as requiring prompt
728 /// retry/fatal recovery rather than repeat the messages.
729 ///
730 /// # Examples
731 ///
732 /// ```
733 /// use std::num::NonZeroUsize;
734 /// use ailloli_ui_runtime::app::{RuntimeHandle, RuntimeInbox};
735 /// let (sender, mut inbox) = RuntimeInbox::channel(NonZeroUsize::new(2).unwrap());
736 /// sender.dispatch("one").unwrap(); sender.request_redraw().unwrap();
737 /// let runtime = RuntimeHandle::new();
738 /// let result = inbox.drain(&runtime).unwrap();
739 /// assert_eq!((result.drained_messages, result.dispatched_actions, result.redraw_all), (2, 1, true));
740 /// ```
741 pub fn drain(&mut self, runtime: &RuntimeHandle<A>) -> Result<RuntimeDrain, UiWakeError> {
742 self.shared.begin_drain();
743 let mut result = RuntimeDrain::default();
744
745 while result.drained_messages < RUNTIME_INBOX_DRAIN_BUDGET {
746 let message = match self.pending.take() {
747 Some(message) => message,
748 None => match self.receiver.try_recv() {
749 Ok(message) => message,
750 Err(TryRecvError::Empty) => break,
751 Err(TryRecvError::Disconnected) => {
752 self.shared.observe_disconnected();
753 break;
754 }
755 },
756 };
757 self.apply_message(runtime, message, &mut result);
758 }
759
760 if result.drained_messages == RUNTIME_INBOX_DRAIN_BUDGET {
761 match self.receiver.try_recv() {
762 Ok(message) => {
763 self.pending = Some(message);
764 result.remaining = true;
765 }
766 Err(TryRecvError::Empty) => {}
767 Err(TryRecvError::Disconnected) => self.shared.observe_disconnected(),
768 }
769 }
770
771 if result.remaining {
772 self.shared.request_wake()?;
773 }
774 Ok(result)
775 }
776
777 /// Applies one message and updates output plus depth diagnostics.
778 fn apply_message(
779 &self,
780 runtime: &RuntimeHandle<A>,
781 message: RuntimeMessage<A>,
782 result: &mut RuntimeDrain,
783 ) {
784 match message {
785 RuntimeMessage::Action(action) => {
786 runtime.dispatch(action);
787 result.dispatched_actions += 1;
788 }
789 RuntimeMessage::RedrawAll => {
790 *self
791 .shared
792 .pending_redraw_all
793 .lock()
794 .unwrap_or_else(|error| error.into_inner()) = false;
795 result.redraw_all = true;
796 }
797 RuntimeMessage::RedrawWindow(logical_window_id) => {
798 self.shared
799 .pending_redraw_windows
800 .lock()
801 .unwrap_or_else(|error| error.into_inner())
802 .remove(&logical_window_id);
803 result.redraw_windows.push(logical_window_id);
804 }
805 }
806 result.drained_messages += 1;
807 self.shared.message_drained();
808 }
809}
810
811#[cfg(test)]
812/// Tests implementation details.
813mod tests {
814 use super::*;
815 use std::sync::atomic::{AtomicUsize, Ordering};
816
817 #[derive(Default)]
818 /// Test support type for CountingWake scenarios.
819 struct CountingWake(AtomicUsize);
820
821 /// Implements the UiWake contract for CountingWake.
822 impl UiWake for CountingWake {
823 /// Notifies the test or host wake target.
824 ///
825 /// # Errors
826 ///
827 /// This counting test callback never returns an error.
828 fn wake(&self) -> Result<(), UiWakeError> {
829 self.0.fetch_add(1, Ordering::Relaxed);
830 Ok(())
831 }
832 }
833
834 #[test]
835 /// Verifies that sender is send and sync when action is send.
836 fn sender_is_send_and_sync_when_action_is_send() {
837 /// Compile-time assertion helper for send sync.
838 fn assert_send_sync<T: Send + Sync>() {}
839 assert_send_sync::<RuntimeSender<String>>();
840 }
841
842 #[test]
843 /// Implements the wake_is_late_bound_and_redraw_is_coalesced helper used by this module.
844 fn wake_is_late_bound_and_redraw_is_coalesced() {
845 let (sender, mut inbox) = RuntimeInbox::<()>::channel(NonZeroUsize::new(4).unwrap());
846 sender.request_redraw().unwrap();
847 sender.request_redraw().unwrap();
848
849 let wake = Arc::new(CountingWake::default());
850 inbox.install_wake(wake.clone()).unwrap();
851 assert_eq!(wake.0.load(Ordering::Relaxed), 1);
852
853 let runtime = RuntimeHandle::new();
854 let drained = inbox.drain(&runtime).unwrap();
855 assert!(drained.redraw_all);
856 assert_eq!(drained.drained_messages, 1);
857 assert_eq!(sender.stats().coalesced, 1);
858 }
859
860 #[test]
861 /// Verifies that drain budget rearms wake without losing actions.
862 fn drain_budget_rearms_wake_without_losing_actions() {
863 let capacity = NonZeroUsize::new(RUNTIME_INBOX_DRAIN_BUDGET + 1).unwrap();
864 let (sender, mut inbox) = RuntimeInbox::channel(capacity);
865 for action in 0..=RUNTIME_INBOX_DRAIN_BUDGET {
866 sender.dispatch(action).unwrap();
867 }
868 let wake = Arc::new(CountingWake::default());
869 inbox.install_wake(wake.clone()).unwrap();
870
871 let runtime = RuntimeHandle::new();
872 let first = inbox.drain(&runtime).unwrap();
873 assert_eq!(first.drained_messages, RUNTIME_INBOX_DRAIN_BUDGET);
874 assert!(first.remaining);
875 let second = inbox.drain(&runtime).unwrap();
876 assert_eq!(second.drained_messages, 1);
877 assert!(!second.remaining);
878
879 let actions = runtime.take_actions();
880 assert_eq!(actions.len(), RUNTIME_INBOX_DRAIN_BUDGET + 1);
881 assert_eq!(actions[0], 0);
882 assert_eq!(
883 actions[RUNTIME_INBOX_DRAIN_BUDGET],
884 RUNTIME_INBOX_DRAIN_BUDGET
885 );
886 assert_eq!(sender.stats().current_depth, 0);
887 }
888
889 #[test]
890 /// Verifies that begin drain reopens the wake slot before examining the queue.
891 fn begin_drain_reopens_the_wake_slot_before_examining_the_queue() {
892 let (sender, mut inbox) = RuntimeInbox::channel(NonZeroUsize::new(4).unwrap());
893 let wake = Arc::new(CountingWake::default());
894 inbox.install_wake(wake.clone()).unwrap();
895
896 sender.dispatch(1).unwrap();
897 assert_eq!(wake.0.load(Ordering::Relaxed), 1);
898
899 // Model the host consuming the first wake immediately before another
900 // producer races with the drain's final queue observation.
901 inbox.shared.begin_drain();
902 sender.dispatch(2).unwrap();
903 assert_eq!(wake.0.load(Ordering::Relaxed), 2);
904
905 let runtime = RuntimeHandle::new();
906 assert_eq!(inbox.drain(&runtime).unwrap().drained_messages, 2);
907 assert_eq!(runtime.take_actions(), [1, 2]);
908 }
909}