uart_fpl/
window.rs

1// Copyright 2026 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5//! Sliding-window flow control math, sequence validation, and cumulative ACK tracking.
6
7use futures::future::poll_fn;
8use futures::task::AtomicWaker;
9use std::sync::Arc;
10use std::sync::atomic::{AtomicU64, Ordering};
11use std::task::{Context, Poll};
12
13const HAS_ACK_FLAG: u64 = 1 << 63;
14
15/// Evaluates whether an 8-bit sequence number `seq` lies within the active sliding window `[base, next_seq)`.
16///
17/// Handles 8-bit modular wrapping (e.g. `base = 250`, `next_seq = 5`).
18/// When `base == next_seq`, the window is empty and this function always returns `false`.
19pub(crate) fn is_seq_in_window(base: u8, next_seq: u8, seq: u8) -> bool {
20    seq.wrapping_sub(base) < next_seq.wrapping_sub(base)
21}
22
23#[derive(Debug, Default)]
24struct AckTrackerInner {
25    state: AtomicU64,
26    waker: AtomicWaker,
27}
28
29/// Thread-safe cumulative ACK tracker with waker notifications.
30///
31/// In Go-Back-N (and ResendSP), ACKs are cumulative. When receiving multiple
32/// in-order frames in rapid succession, `AckTracker` maintains the latest
33/// session ID and highest acknowledged sequence number. This prevents reverse-path
34/// flooding (e.g. queueing hundreds of redundant individual ACK frames) and
35/// ensures the writer task is immediately woken up to emit the cumulative ACK.
36///
37/// Both `session_id` and `seq` (along with the pending ACK flag) are packed
38/// into a single atomic `u64` to prevent mixed-state torn reads between concurrent
39/// `set_ack` and `take_ack` calls.
40#[derive(Debug, Clone, Default)]
41pub struct AckTracker {
42    inner: Arc<AckTrackerInner>,
43}
44
45impl AckTracker {
46    /// Creates a new, empty ACK tracker.
47    pub fn new() -> Self {
48        Self::default()
49    }
50
51    /// Records a new cumulative ACK for `session_id` up to `seq` and notifies any waiting writer.
52    pub fn set_ack(&self, session_id: u32, seq: u8) {
53        let packed = HAS_ACK_FLAG | ((seq as u64) << 32) | (session_id as u64);
54        self.inner.state.store(packed, Ordering::Release);
55        self.inner.waker.wake();
56    }
57
58    /// Retrieves and clears the latest cumulative ACK if one is pending.
59    pub fn take_ack(&self) -> Option<(u32, u8)> {
60        let val = self.inner.state.swap(0, Ordering::AcqRel);
61        if (val & HAS_ACK_FLAG) != 0 {
62            let sid = val as u32;
63            let seq = ((val >> 32) & 0xFF) as u8;
64            Some((sid, seq))
65        } else {
66            None
67        }
68    }
69
70    /// Registers the current task waker to be notified when a new ACK is recorded.
71    pub fn register_waker(&self, cx: &mut Context<'_>) {
72        self.inner.waker.register(cx.waker());
73    }
74
75    /// Asynchronously waits until a cumulative ACK is recorded and returns `(session_id, seq)`.
76    ///
77    /// Note: Designed for a single waiting task. If multiple tasks await `wait_ack()`,
78    /// only the most recently registered waker is notified when an ACK arrives.
79    pub async fn wait_ack(&self) -> (u32, u8) {
80        poll_fn(|cx| {
81            if let Some(ack) = self.take_ack() {
82                Poll::Ready(ack)
83            } else {
84                self.register_waker(cx);
85                if let Some(ack) = self.take_ack() { Poll::Ready(ack) } else { Poll::Pending }
86            }
87        })
88        .await
89    }
90}
91
92#[cfg(test)]
93mod tests {
94    use super::*;
95
96    fn assert_send<T: Send>() {}
97    fn assert_sync<T: Sync>() {}
98
99    #[fuchsia::test]
100    fn test_tracker_thread_safe() {
101        assert_send::<AckTracker>();
102        assert_sync::<AckTracker>();
103    }
104
105    #[fuchsia::test]
106    async fn test_ack_tracker_wait_ack() {
107        let tracker = AckTracker::new();
108        tracker.set_ack(0x12345678, 42);
109        let (sid, seq) = tracker.wait_ack().await;
110        assert_eq!(sid, 0x12345678);
111        assert_eq!(seq, 42);
112    }
113
114    #[fuchsia::test]
115    async fn test_ack_tracker_wait_ack_async_wake() {
116        let tracker = AckTracker::new();
117        let tracker_clone = tracker.clone();
118        let join_handle = std::thread::spawn(move || {
119            std::thread::sleep(std::time::Duration::from_millis(10));
120            tracker_clone.set_ack(0x12345678, 42);
121        });
122        let (sid, seq) = tracker.wait_ack().await;
123        assert_eq!(sid, 0x12345678);
124        assert_eq!(seq, 42);
125        join_handle.join().unwrap();
126    }
127
128    #[fuchsia::test]
129    fn test_ack_tracker() {
130        let tracker = AckTracker::new();
131        assert_eq!(tracker.take_ack(), None);
132
133        tracker.set_ack(0x12345678, 42);
134        assert_eq!(tracker.take_ack(), Some((0x12345678, 42)));
135        assert_eq!(tracker.take_ack(), None);
136
137        // Coalescing updates
138        tracker.set_ack(0x11111111, 1);
139        tracker.set_ack(0x11111111, 2);
140        assert_eq!(tracker.take_ack(), Some((0x11111111, 2)));
141    }
142
143    #[fuchsia::test]
144    fn test_is_seq_in_window() {
145        // Normal case without wrapping: [10, 20)
146        assert!(!is_seq_in_window(10, 20, 9));
147        assert!(is_seq_in_window(10, 20, 10));
148        assert!(is_seq_in_window(10, 20, 15));
149        assert!(is_seq_in_window(10, 20, 19));
150        assert!(!is_seq_in_window(10, 20, 20));
151
152        // Empty window: base == next_seq
153        assert!(!is_seq_in_window(10, 10, 10));
154        assert!(!is_seq_in_window(10, 10, 9));
155
156        // Wrapping case: [250, 10)
157        assert!(is_seq_in_window(250, 10, 250));
158        assert!(is_seq_in_window(250, 10, 255));
159        assert!(is_seq_in_window(250, 10, 0));
160        assert!(is_seq_in_window(250, 10, 9));
161        assert!(!is_seq_in_window(250, 10, 10));
162        assert!(!is_seq_in_window(250, 10, 249));
163    }
164}
uart_fpl/window.rs

uart_fpl/
window.rs
