serial_core/
debugger.rs

1// Copyright (c) Microsoft Corporation.
2// Licensed under the MIT License.
3
4//! Debugger-mode serial relay.
5
6use crate::SerialIo;
7use futures::AsyncRead;
8use futures::AsyncWrite;
9use futures::future::poll_fn;
10use inspect::InspectMut;
11use pal_async::driver::Driver;
12use pal_async::task::Spawn;
13use pal_async::task::Task;
14use pal_async::timer::Instant;
15use pal_async::timer::PolledTimer;
16use parking_lot::Mutex;
17use std::collections::VecDeque;
18use std::io;
19use std::pin::Pin;
20use std::sync::Arc;
21use std::task::Context;
22use std::task::Poll;
23use std::task::Waker;
24use std::time::Duration;
25
26const RX_RING_CAP: usize = 16 * 1024;
27const TX_RING_CAP: usize = 16 * 1024;
28const PUMP_CHUNK: usize = 1024;
29
30/// How often the pump is allowed to drain (and drop from) an already-full RX
31/// ring.
32///
33/// Once the guest stops draining and the RX ring fills, the pump must keep
34/// reading the backend so the debugger transport never deadlocks, dropping the
35/// newest bytes that do not fit. Doing that as fast as an always-ready backend
36/// can supply data burns CPU for no benefit, so while the ring stays full the
37/// pump only drains once per this interval. This bounds the wasted CPU without
38/// changing the drop-newest / never-backpressure-the-guest semantics: bytes are
39/// still dropped when the ring is full, the transport is still drained, and the
40/// pump resumes draining at full speed the instant the device frees ring space.
41const PUMP_DRAIN_THROTTLE: Duration = Duration::from_millis(1);
42
43/// Maximum number of productive pump iterations in a single poll before the pump
44/// yields back to the executor.
45///
46/// This bounds how long the pump runs in one poll so that a fast, always-ready
47/// backend cannot monopolize the executor. In particular, once the guest stops
48/// draining and the RX ring fills, the pump must keep reading (and dropping) the
49/// backend to avoid deadlocking the debugger transport; this budget ensures it
50/// does so cooperatively rather than spinning without ever yielding. It is a
51/// scheduler-safety guard, not a KD rate limiter (a genuine throttle is future
52/// work); a backend that is synchronously always-ready can still consume CPU.
53const PUMP_POLL_BUDGET: u32 = 256;
54
55/// A [`SerialIo`] adapter for WinDbg / KD-over-serial debugger mode.
56///
57/// The relay keeps the wrapped backend continuously drained from an independent
58/// pump task and never applies guest-visible backpressure. If either bounded
59/// relay ring fills, the newest bytes that do not fit are dropped.
60///
61/// Because writes are accepted into a lossy relay, `poll_write` always reports
62/// the full write as accepted and `poll_flush`/`poll_close` do not guarantee the
63/// bytes reached the real backend. `poll_connect`/`poll_disconnect` assume a
64/// single consumer (the serial emulator, which waits on only one of them at a
65/// time).
66pub struct DebuggerRelay {
67    inner: Arc<Mutex<Inner>>,
68    _pump: Task<()>,
69}
70
71struct Inner {
72    rx: RingBuf,
73    tx: RingBuf,
74    connected: bool,
75    eof: bool,
76    device_rx_waker: Option<Waker>,
77    device_conn_waker: Option<Waker>,
78    pump_waker: Option<Waker>,
79    rx_dropped: u64,
80    tx_dropped: u64,
81}
82
83struct RingBuf {
84    buf: VecDeque<u8>,
85    cap: usize,
86}
87
88#[derive(Default)]
89struct WakeList(Vec<Waker>);
90
91impl DebuggerRelay {
92    /// Wraps `inner`, spawning a pump task on `driver` to relay data in both
93    /// directions. `name` labels the spawned task for diagnostics.
94    pub fn new(driver: impl Spawn + Driver, name: &str, inner: Box<dyn SerialIo>) -> Self {
95        let connected = inner.is_connected();
96        let shared = Arc::new(Mutex::new(Inner {
97            rx: RingBuf::new(RX_RING_CAP),
98            tx: RingBuf::new(TX_RING_CAP),
99            connected,
100            eof: !connected,
101            device_rx_waker: None,
102            device_conn_waker: None,
103            pump_waker: None,
104            rx_dropped: 0,
105            tx_dropped: 0,
106        }));
107
108        let pump_shared = shared.clone();
109        let timer = PolledTimer::new(&driver);
110        let task_name = format!("{name}-serial-debugger-relay");
111        let pump = driver.spawn(task_name, async move {
112            run_pump(inner, pump_shared, Some(timer)).await;
113        });
114
115        Self {
116            inner: shared,
117            _pump: pump,
118        }
119    }
120}
121
122/// Wraps `io` in a [`DebuggerRelay`] (spawning a pump task on `driver`) when
123/// `debugger_mode` is set, otherwise returns `io` unchanged.
124///
125/// Shared by the serial device resolvers so the wrapping logic lives in one
126/// place.
127pub fn apply_debugger_mode(
128    debugger_mode: bool,
129    driver: impl Spawn + Driver,
130    name: &str,
131    io: Box<dyn SerialIo>,
132) -> Box<dyn SerialIo> {
133    if debugger_mode {
134        Box::new(DebuggerRelay::new(driver, name, io))
135    } else {
136        io
137    }
138}
139
140impl RingBuf {
141    fn new(cap: usize) -> Self {
142        Self {
143            buf: VecDeque::new(),
144            cap,
145        }
146    }
147
148    fn len(&self) -> usize {
149        self.buf.len()
150    }
151
152    fn is_empty(&self) -> bool {
153        self.buf.is_empty()
154    }
155
156    fn is_full(&self) -> bool {
157        self.buf.len() >= self.cap
158    }
159
160    fn clear(&mut self) -> usize {
161        let len = self.buf.len();
162        self.buf.clear();
163        len
164    }
165
166    fn push_drop_newest(&mut self, data: &[u8]) -> usize {
167        let available = self.cap.saturating_sub(self.buf.len());
168        let accepted = available.min(data.len());
169        self.buf.extend(&data[..accepted]);
170        data.len() - accepted
171    }
172
173    fn pop_into(&mut self, buf: &mut [u8]) -> usize {
174        let n = buf.len().min(self.buf.len());
175        for (dst, src) in buf.iter_mut().zip(self.buf.drain(..n)) {
176            *dst = src;
177        }
178        n
179    }
180
181    fn copy_front(&self, buf: &mut [u8]) -> usize {
182        let n = buf.len().min(self.buf.len());
183        for (dst, src) in buf.iter_mut().zip(self.buf.iter().copied().take(n)) {
184            *dst = src;
185        }
186        n
187    }
188
189    fn pop_front(&mut self, n: usize) {
190        let n = n.min(self.buf.len());
191        drop(self.buf.drain(..n));
192    }
193}
194
195impl WakeList {
196    fn take(&mut self, waker: &mut Option<Waker>) {
197        if let Some(waker) = waker.take() {
198            self.0.push(waker);
199        }
200    }
201
202    fn wake(self) {
203        for waker in self.0 {
204            waker.wake();
205        }
206    }
207}
208
209impl Inner {
210    fn connect(&mut self, wakes: &mut WakeList) -> bool {
211        let changed = !self.connected || self.eof;
212        self.connected = true;
213        self.eof = false;
214        if changed {
215            wakes.take(&mut self.device_conn_waker);
216            wakes.take(&mut self.device_rx_waker);
217        }
218        changed
219    }
220
221    fn disconnect(&mut self, wakes: &mut WakeList) -> bool {
222        let was_connected = self.connected;
223        let eof_changed = self.rx.is_empty() && !self.eof;
224        self.connected = false;
225        if self.rx.is_empty() {
226            self.eof = true;
227        }
228        if was_connected {
229            wakes.take(&mut self.device_conn_waker);
230        }
231        if was_connected || eof_changed {
232            wakes.take(&mut self.device_rx_waker);
233        }
234        was_connected || eof_changed
235    }
236
237    fn wake_rx(&mut self, wakes: &mut WakeList) {
238        wakes.take(&mut self.device_rx_waker);
239    }
240
241    fn wake_pump(&mut self, wakes: &mut WakeList) {
242        wakes.take(&mut self.pump_waker);
243    }
244}
245
246async fn run_pump(
247    mut inner: Box<dyn SerialIo>,
248    shared: Arc<Mutex<Inner>>,
249    mut throttle: Option<PolledTimer>,
250) {
251    let mut rx_buf = [0; PUMP_CHUNK];
252    let mut tx_buf = [0; PUMP_CHUNK];
253    // The earliest time the pump may drain an already-full RX ring again. See
254    // [`PUMP_DRAIN_THROTTLE`].
255    let mut rx_next_drain: Option<Instant> = None;
256
257    poll_fn(move |cx| {
258        let mut budget = PUMP_POLL_BUDGET;
259        loop {
260            let mut made_progress = false;
261
262            if !shared.lock().connected {
263                match inner.poll_connect(cx) {
264                    Poll::Ready(Ok(())) => {
265                        let mut wakes = WakeList::default();
266                        made_progress |= shared.lock().connect(&mut wakes);
267                        wakes.wake();
268                    }
269                    Poll::Ready(Err(_)) => {
270                        let mut wakes = WakeList::default();
271                        made_progress |= shared.lock().disconnect(&mut wakes);
272                        wakes.wake();
273                    }
274                    Poll::Pending => {}
275                }
276            }
277
278            if shared.lock().connected {
279                // While the RX ring is full we must keep draining the backend
280                // (dropping the newest bytes) so the transport never deadlocks,
281                // but draining as fast as an always-ready backend can supply
282                // data wastes CPU. Bound the drain rate in that case. This does
283                // not change semantics: bytes are still dropped when full, the
284                // transport is still drained, and full-speed draining resumes
285                // the instant the device frees ring space. `throttle` is `None`
286                // only in unit tests that drive the un-throttled loop directly.
287                let drain_now = if shared.lock().rx.is_full() {
288                    match &mut throttle {
289                        Some(timer) => {
290                            let now = Instant::now();
291                            match rx_next_drain {
292                                Some(deadline) if now < deadline => {
293                                    // Not yet time to drain again. Wake on the
294                                    // timer, or early via `pump_waker` when the
295                                    // device frees ring space.
296                                    timer.poll_until(cx, deadline).is_ready()
297                                }
298                                _ => {
299                                    rx_next_drain = Some(now + PUMP_DRAIN_THROTTLE);
300                                    true
301                                }
302                            }
303                        }
304                        None => true,
305                    }
306                } else {
307                    rx_next_drain = None;
308                    true
309                };
310
311                if drain_now {
312                    match Pin::new(&mut inner).poll_read(cx, &mut rx_buf) {
313                        Poll::Ready(Ok(0)) => {
314                            let mut wakes = WakeList::default();
315                            made_progress |= shared.lock().disconnect(&mut wakes);
316                            wakes.wake();
317                        }
318                        Poll::Ready(Ok(n)) => {
319                            let mut wakes = WakeList::default();
320                            {
321                                let mut state = shared.lock();
322                                let dropped = state.rx.push_drop_newest(&rx_buf[..n]);
323                                state.rx_dropped += dropped as u64;
324                                if dropped < n {
325                                    state.wake_rx(&mut wakes);
326                                }
327                            }
328                            wakes.wake();
329                            made_progress = true;
330                        }
331                        Poll::Ready(Err(_)) => {
332                            let mut wakes = WakeList::default();
333                            made_progress |= shared.lock().disconnect(&mut wakes);
334                            wakes.wake();
335                        }
336                        Poll::Pending => {}
337                    }
338                }
339            }
340
341            let (tx_len, had_tx) = {
342                let mut state = shared.lock();
343                if state.connected {
344                    let len = state.tx.copy_front(&mut tx_buf);
345                    (len, len != 0)
346                } else {
347                    let dropped = state.tx.clear();
348                    state.tx_dropped += dropped as u64;
349                    if dropped != 0 {
350                        made_progress = true;
351                    }
352                    (0, dropped != 0)
353                }
354            };
355
356            if tx_len != 0 {
357                match Pin::new(&mut inner).poll_write(cx, &tx_buf[..tx_len]) {
358                    Poll::Ready(Ok(0)) => {
359                        let mut wakes = WakeList::default();
360                        {
361                            let mut state = shared.lock();
362                            let dropped = state.tx.clear();
363                            state.tx_dropped += dropped as u64;
364                            made_progress |= state.disconnect(&mut wakes);
365                        }
366                        wakes.wake();
367                    }
368                    Poll::Ready(Ok(n)) => {
369                        shared.lock().tx.pop_front(n);
370                        made_progress = true;
371                    }
372                    Poll::Ready(Err(_err)) => {
373                        let mut wakes = WakeList::default();
374                        {
375                            let mut state = shared.lock();
376                            let dropped = state.tx.clear();
377                            state.tx_dropped += dropped as u64;
378                            made_progress |= state.disconnect(&mut wakes);
379                        }
380                        wakes.wake();
381                    }
382                    Poll::Pending => {}
383                }
384            }
385
386            if made_progress {
387                budget -= 1;
388                if budget == 0 {
389                    // Yield cooperatively so a continuously-ready backend cannot
390                    // monopolize the executor. Reschedule immediately to keep
391                    // draining on the next poll.
392                    cx.waker().wake_by_ref();
393                    return Poll::Pending;
394                }
395                continue;
396            }
397
398            let should_park = {
399                let mut state = shared.lock();
400                if !had_tx && !state.tx.is_empty() {
401                    false
402                } else {
403                    state.pump_waker = Some(cx.waker().clone());
404                    true
405                }
406            };
407            if should_park {
408                return Poll::Pending;
409            }
410        }
411    })
412    .await
413}
414
415impl SerialIo for DebuggerRelay {
416    fn is_connected(&self) -> bool {
417        self.inner.lock().connected
418    }
419
420    fn poll_connect(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
421        let mut state = self.inner.lock();
422        if state.connected {
423            Poll::Ready(Ok(()))
424        } else {
425            state.device_conn_waker = Some(cx.waker().clone());
426            Poll::Pending
427        }
428    }
429
430    fn poll_disconnect(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
431        let mut state = self.inner.lock();
432        if !state.connected {
433            Poll::Ready(Ok(()))
434        } else {
435            state.device_conn_waker = Some(cx.waker().clone());
436            Poll::Pending
437        }
438    }
439}
440
441impl AsyncRead for DebuggerRelay {
442    fn poll_read(
443        self: Pin<&mut Self>,
444        cx: &mut Context<'_>,
445        buf: &mut [u8],
446    ) -> Poll<io::Result<usize>> {
447        let mut wakes = WakeList::default();
448        let result = {
449            let mut state = self.inner.lock();
450            if !state.rx.is_empty() || buf.is_empty() {
451                let n = state.rx.pop_into(buf);
452                if !state.connected && state.rx.is_empty() {
453                    state.eof = true;
454                }
455                state.wake_pump(&mut wakes);
456                Poll::Ready(Ok(n))
457            } else if state.eof {
458                Poll::Ready(Ok(0))
459            } else {
460                state.device_rx_waker = Some(cx.waker().clone());
461                Poll::Pending
462            }
463        };
464        wakes.wake();
465        result
466    }
467}
468
469impl AsyncWrite for DebuggerRelay {
470    fn poll_write(
471        self: Pin<&mut Self>,
472        _cx: &mut Context<'_>,
473        buf: &[u8],
474    ) -> Poll<io::Result<usize>> {
475        let mut wakes = WakeList::default();
476        {
477            let mut state = self.inner.lock();
478            let dropped = state.tx.push_drop_newest(buf);
479            state.tx_dropped += dropped as u64;
480            state.wake_pump(&mut wakes);
481        }
482        wakes.wake();
483        Poll::Ready(Ok(buf.len()))
484    }
485
486    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
487        Poll::Ready(Ok(()))
488    }
489
490    fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
491        Poll::Ready(Ok(()))
492    }
493}
494
495impl InspectMut for DebuggerRelay {
496    fn inspect_mut(&mut self, req: inspect::Request<'_>) {
497        let state = self.inner.lock();
498        req.respond()
499            .field("connected", state.connected)
500            .field("rx_depth", state.rx.len())
501            .field("tx_depth", state.tx.len())
502            .field("rx_dropped", state.rx_dropped)
503            .field("tx_dropped", state.tx_dropped);
504    }
505}
506
507#[cfg(test)]
508mod tests {
509    use super::*;
510    use pal_async::DefaultDriver;
511    use pal_async::async_test;
512    use std::io::ErrorKind;
513    use std::task::Context;
514    use test_with_tracing::test;
515
516    #[derive(Clone)]
517    struct MockHandle {
518        state: Arc<Mutex<MockState>>,
519    }
520
521    struct MockSerialIo {
522        state: Arc<Mutex<MockState>>,
523    }
524
525    struct MockState {
526        connected: bool,
527        read_buf: VecDeque<u8>,
528        infinite_read: bool,
529        read_polls: usize,
530        panic_after_read_polls: Option<usize>,
531        write_zero: bool,
532        written: Vec<u8>,
533        write_limit: usize,
534        fail_write_disconnect: bool,
535        dropped: bool,
536        read_waker: Option<Waker>,
537        write_waker: Option<Waker>,
538        connect_waker: Option<Waker>,
539        disconnect_waker: Option<Waker>,
540        wait_waker: Option<Waker>,
541    }
542
543    #[derive(Debug)]
544    struct RelaySnapshot {
545        connected: bool,
546        eof: bool,
547        rx_depth: usize,
548        tx_depth: usize,
549        rx_dropped: u64,
550        tx_dropped: u64,
551    }
552
553    impl MockSerialIo {
554        fn new() -> (Self, MockHandle) {
555            let state = Arc::new(Mutex::new(MockState {
556                connected: true,
557                read_buf: VecDeque::new(),
558                infinite_read: false,
559                read_polls: 0,
560                panic_after_read_polls: None,
561                write_zero: false,
562                written: Vec::new(),
563                write_limit: usize::MAX,
564                fail_write_disconnect: false,
565                dropped: false,
566                read_waker: None,
567                write_waker: None,
568                connect_waker: None,
569                disconnect_waker: None,
570                wait_waker: None,
571            }));
572            (
573                Self {
574                    state: state.clone(),
575                },
576                MockHandle { state },
577            )
578        }
579    }
580
581    impl Drop for MockSerialIo {
582        fn drop(&mut self) {
583            let mut wakes = WakeList::default();
584            {
585                let mut state = self.state.lock();
586                state.dropped = true;
587                state.wake_waiter(&mut wakes);
588            }
589            wakes.wake();
590        }
591    }
592
593    impl InspectMut for MockSerialIo {
594        fn inspect_mut(&mut self, req: inspect::Request<'_>) {
595            req.ignore();
596        }
597    }
598
599    impl SerialIo for MockSerialIo {
600        fn is_connected(&self) -> bool {
601            self.state.lock().connected
602        }
603
604        fn poll_connect(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
605            let mut state = self.state.lock();
606            if state.connected {
607                Poll::Ready(Ok(()))
608            } else {
609                state.connect_waker = Some(cx.waker().clone());
610                Poll::Pending
611            }
612        }
613
614        fn poll_disconnect(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
615            let mut state = self.state.lock();
616            if state.connected {
617                state.disconnect_waker = Some(cx.waker().clone());
618                Poll::Pending
619            } else {
620                Poll::Ready(Ok(()))
621            }
622        }
623    }
624
625    impl AsyncRead for MockSerialIo {
626        fn poll_read(
627            self: Pin<&mut Self>,
628            cx: &mut Context<'_>,
629            buf: &mut [u8],
630        ) -> Poll<io::Result<usize>> {
631            let mut wakes = WakeList::default();
632            let result = {
633                let mut state = self.state.lock();
634                state.read_polls += 1;
635                if let Some(limit) = state.panic_after_read_polls {
636                    assert!(
637                        state.read_polls <= limit,
638                        "backend read polled {} times in one scheduling turn (pump is spinning without yielding)",
639                        state.read_polls
640                    );
641                }
642                if !state.connected {
643                    state.wake_waiter(&mut wakes);
644                    Poll::Ready(Ok(0))
645                } else if state.infinite_read {
646                    // A synchronously always-ready source, used to prove the pump
647                    // keeps draining without spinning the executor.
648                    buf.fill(0xAB);
649                    state.wake_waiter(&mut wakes);
650                    Poll::Ready(Ok(buf.len()))
651                } else if state.read_buf.is_empty() {
652                    state.read_waker = Some(cx.waker().clone());
653                    Poll::Pending
654                } else {
655                    let n = buf.len().min(state.read_buf.len());
656                    for (dst, src) in buf.iter_mut().zip(state.read_buf.drain(..n)) {
657                        *dst = src;
658                    }
659                    state.wake_waiter(&mut wakes);
660                    Poll::Ready(Ok(n))
661                }
662            };
663            wakes.wake();
664            result
665        }
666    }
667
668    impl AsyncWrite for MockSerialIo {
669        fn poll_write(
670            self: Pin<&mut Self>,
671            cx: &mut Context<'_>,
672            buf: &[u8],
673        ) -> Poll<io::Result<usize>> {
674            let mut wakes = WakeList::default();
675            let result = {
676                let mut state = self.state.lock();
677                if state.fail_write_disconnect {
678                    state.connected = false;
679                    state.wake_waiter(&mut wakes);
680                    Poll::Ready(Err(ErrorKind::BrokenPipe.into()))
681                } else if !state.connected {
682                    Poll::Ready(Err(ErrorKind::BrokenPipe.into()))
683                } else if state.write_zero {
684                    // A misbehaving backend that accepts zero bytes on a
685                    // non-empty write; the pump must treat this as a disconnect
686                    // rather than looping forever making no progress.
687                    state.connected = false;
688                    state.wake_waiter(&mut wakes);
689                    Poll::Ready(Ok(0))
690                } else if state.write_limit == 0 {
691                    state.write_waker = Some(cx.waker().clone());
692                    Poll::Pending
693                } else {
694                    let n = state.write_limit.min(buf.len());
695                    state.written.extend_from_slice(&buf[..n]);
696                    state.wake_waiter(&mut wakes);
697                    Poll::Ready(Ok(n))
698                }
699            };
700            wakes.wake();
701            result
702        }
703
704        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
705            Poll::Ready(Ok(()))
706        }
707
708        fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
709            Poll::Ready(Ok(()))
710        }
711    }
712
713    impl MockState {
714        fn wake_waiter(&mut self, wakes: &mut WakeList) {
715            wakes.take(&mut self.wait_waker);
716        }
717    }
718
719    impl MockHandle {
720        fn inject_rx(&self, data: &[u8]) {
721            let mut wakes = WakeList::default();
722            {
723                let mut state = self.state.lock();
724                state.read_buf.extend(data);
725                wakes.take(&mut state.read_waker);
726                state.wake_waiter(&mut wakes);
727            }
728            wakes.wake();
729        }
730
731        fn set_infinite_read(&self) {
732            let mut wakes = WakeList::default();
733            {
734                let mut state = self.state.lock();
735                state.infinite_read = true;
736                wakes.take(&mut state.read_waker);
737                state.wake_waiter(&mut wakes);
738            }
739            wakes.wake();
740        }
741
742        fn set_connected(&self, connected: bool) {
743            let mut wakes = WakeList::default();
744            {
745                let mut state = self.state.lock();
746                state.connected = connected;
747                if connected {
748                    wakes.take(&mut state.connect_waker);
749                } else {
750                    wakes.take(&mut state.disconnect_waker);
751                    wakes.take(&mut state.read_waker);
752                    wakes.take(&mut state.write_waker);
753                }
754                state.wake_waiter(&mut wakes);
755            }
756            wakes.wake();
757        }
758
759        fn set_write_limit(&self, limit: usize) {
760            let mut wakes = WakeList::default();
761            {
762                let mut state = self.state.lock();
763                state.write_limit = limit;
764                wakes.take(&mut state.write_waker);
765                state.wake_waiter(&mut wakes);
766            }
767            wakes.wake();
768        }
769
770        fn fail_writes_with_disconnect(&self) {
771            let mut wakes = WakeList::default();
772            {
773                let mut state = self.state.lock();
774                state.fail_write_disconnect = true;
775                wakes.take(&mut state.write_waker);
776                state.wake_waiter(&mut wakes);
777            }
778            wakes.wake();
779        }
780
781        /// Makes the backend accept zero bytes on writes (a `Ok(0)` write-zero).
782        fn set_write_zero(&self) {
783            let mut wakes = WakeList::default();
784            {
785                let mut state = self.state.lock();
786                state.write_zero = true;
787                wakes.take(&mut state.write_waker);
788                state.wake_waiter(&mut wakes);
789            }
790            wakes.wake();
791        }
792
793        /// Panic (rather than hang) if the backend is read-polled more than
794        /// `limit` times without the pump yielding, so a spin fails cleanly.
795        fn panic_after_read_polls(&self, limit: usize) {
796            self.state.lock().panic_after_read_polls = Some(limit);
797        }
798
799        fn read_polls(&self) -> usize {
800            self.state.lock().read_polls
801        }
802
803        fn written(&self) -> Vec<u8> {
804            self.state.lock().written.clone()
805        }
806
807        async fn wait_until(&self, mut predicate: impl FnMut(&MockState) -> bool) {
808            poll_fn(|cx| {
809                let mut state = self.state.lock();
810                if predicate(&state) {
811                    Poll::Ready(())
812                } else {
813                    state.wait_waker = Some(cx.waker().clone());
814                    Poll::Pending
815                }
816            })
817            .await
818        }
819    }
820
821    fn sequence(len: usize) -> Vec<u8> {
822        (0..len).map(|x| (x % 251) as u8).collect()
823    }
824
825    /// A `Waker` that counts how many times it was woken, for asserting whether
826    /// a future re-armed itself (cooperative yield) or stayed parked (idle).
827    struct CountingWaker(std::sync::atomic::AtomicUsize);
828
829    impl std::task::Wake for CountingWaker {
830        fn wake(self: Arc<Self>) {
831            self.0.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
832        }
833        fn wake_by_ref(self: &Arc<Self>) {
834            self.0.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
835        }
836    }
837
838    impl CountingWaker {
839        fn new() -> (Arc<Self>, Waker) {
840            let arc = Arc::new(CountingWaker(std::sync::atomic::AtomicUsize::new(0)));
841            let waker = Waker::from(arc.clone());
842            (arc, waker)
843        }
844        fn count(&self) -> usize {
845            self.0.load(std::sync::atomic::Ordering::Relaxed)
846        }
847    }
848
849    fn new_shared(connected: bool) -> Arc<Mutex<Inner>> {
850        Arc::new(Mutex::new(Inner {
851            rx: RingBuf::new(RX_RING_CAP),
852            tx: RingBuf::new(TX_RING_CAP),
853            connected,
854            eof: !connected,
855            device_rx_waker: None,
856            device_conn_waker: None,
857            pump_waker: None,
858            rx_dropped: 0,
859            tx_dropped: 0,
860        }))
861    }
862
863    fn snapshot(relay: &DebuggerRelay) -> RelaySnapshot {
864        let state = relay.inner.lock();
865        RelaySnapshot {
866            connected: state.connected,
867            eof: state.eof,
868            rx_depth: state.rx.len(),
869            tx_depth: state.tx.len(),
870            rx_dropped: state.rx_dropped,
871            tx_dropped: state.tx_dropped,
872        }
873    }
874
875    async fn wait_for_relay(
876        relay: &DebuggerRelay,
877        mut predicate: impl FnMut(&RelaySnapshot) -> bool,
878    ) {
879        poll_fn(|cx| {
880            if predicate(&snapshot(relay)) {
881                Poll::Ready(())
882            } else {
883                cx.waker().wake_by_ref();
884                Poll::Pending
885            }
886        })
887        .await
888    }
889
890    fn poll_read_now(relay: &mut DebuggerRelay, buf: &mut [u8]) -> Poll<io::Result<usize>> {
891        let mut cx = Context::from_waker(Waker::noop());
892        Pin::new(relay).poll_read(&mut cx, buf)
893    }
894
895    fn poll_write_now(relay: &mut DebuggerRelay, buf: &[u8]) -> Poll<io::Result<usize>> {
896        let mut cx = Context::from_waker(Waker::noop());
897        Pin::new(relay).poll_write(&mut cx, buf)
898    }
899
900    fn poll_connect_now(relay: &mut DebuggerRelay) -> Poll<io::Result<()>> {
901        let mut cx = Context::from_waker(Waker::noop());
902        relay.poll_connect(&mut cx)
903    }
904
905    #[async_test]
906    async fn rx_drains_without_reader(driver: DefaultDriver) {
907        let (io, handle) = MockSerialIo::new();
908        let relay = DebuggerRelay::new(driver, "rx-drain", Box::new(io));
909        let burst = sequence(RX_RING_CAP + 512);
910
911        handle.inject_rx(&burst);
912        handle.wait_until(|state| state.read_buf.is_empty()).await;
913
914        let snapshot = snapshot(&relay);
915        assert_eq!(snapshot.rx_depth, RX_RING_CAP);
916        assert_eq!(snapshot.rx_dropped, 512);
917    }
918
919    /// With a synchronously always-ready backend and a full RX ring, the pump
920    /// must keep reading and dropping (so the transport never deadlocks) but must
921    /// still yield back to the executor rather than spinning forever in a single
922    /// poll. Polling `run_pump` directly asserts this deterministically: a fixed
923    /// pump returns `Pending` (after bounded work) and re-arms its own waker. The
924    /// mock's `panic_after_read_polls` guard turns a regressed spin into a clean
925    /// panic instead of a hang.
926    #[test]
927    fn pump_yields_instead_of_spinning() {
928        let shared = new_shared(true);
929        let (io, handle) = MockSerialIo::new();
930        handle.set_infinite_read();
931        handle.panic_after_read_polls(PUMP_POLL_BUDGET as usize + 16);
932
933        let fut = run_pump(Box::new(io), shared.clone(), None);
934        let mut fut = std::pin::pin!(fut);
935
936        let (counter, waker) = CountingWaker::new();
937        let mut cx = Context::from_waker(&waker);
938
939        // A single poll must return: the pump drains and drops into the full ring
940        // but yields once its per-poll budget is spent.
941        assert!(fut.as_mut().poll(&mut cx).is_pending());
942
943        // It kept draining/dropping the always-ready backend within that poll...
944        assert!(shared.lock().rx_dropped >= RX_RING_CAP as u64);
945        // ...and re-armed itself to continue on the next poll (cooperative yield).
946        assert!(counter.count() >= 1);
947    }
948
949    /// When there is nothing to do, the pump must park (return Pending) after a
950    /// single backend poll WITHOUT re-arming its own waker, otherwise it would
951    /// spin at 100% CPU while the guest is idle.
952    #[test]
953    fn pump_parks_when_idle() {
954        let shared = new_shared(true);
955        let (io, handle) = MockSerialIo::new();
956
957        let fut = run_pump(Box::new(io), shared.clone(), None);
958        let mut fut = std::pin::pin!(fut);
959
960        let (counter, waker) = CountingWaker::new();
961        let mut cx = Context::from_waker(&waker);
962
963        assert!(fut.as_mut().poll(&mut cx).is_pending());
964        // Parked: polled the backend once, then waited without self-waking.
965        assert_eq!(handle.read_polls(), 1);
966        assert_eq!(counter.count(), 0);
967    }
968
969    /// With the throttle timer present (as in production), a full RX ring fed by
970    /// an always-ready backend must be drained at a bounded rate rather than
971    /// busy-dropping: a single poll fills the ring and then parks on the timer,
972    /// dropping only a bounded amount instead of the whole per-poll budget's
973    /// worth. Semantics are unchanged (the ring is still kept full/drained and
974    /// the newest bytes are still dropped); only the drop *rate* is bounded.
975    #[async_test]
976    async fn pump_throttles_drain_of_full_ring(driver: DefaultDriver) {
977        let shared = new_shared(true);
978        let (io, handle) = MockSerialIo::new();
979        handle.set_infinite_read();
980        handle.panic_after_read_polls(PUMP_POLL_BUDGET as usize + 16);
981
982        let timer = PolledTimer::new(&driver);
983        let fut = run_pump(Box::new(io), shared.clone(), Some(timer));
984        let mut fut = std::pin::pin!(fut);
985
986        let (_counter, waker) = CountingWaker::new();
987        let mut cx = Context::from_waker(&waker);
988
989        assert!(fut.as_mut().poll(&mut cx).is_pending());
990
991        let state = shared.lock();
992        // The ring was filled and is being kept drained...
993        assert_eq!(state.rx.len(), RX_RING_CAP);
994        // ...but the always-ready backend was throttled after filling it, so far
995        // fewer bytes were dropped than the un-throttled spin path would drop in
996        // a single poll (which drops >= RX_RING_CAP; see
997        // `pump_yields_instead_of_spinning`).
998        assert!(
999            state.rx_dropped <= 2 * PUMP_CHUNK as u64,
1000            "dropped {} bytes; expected throttled (<= {})",
1001            state.rx_dropped,
1002            2 * PUMP_CHUNK
1003        );
1004    }
1005
1006    #[async_test]
1007    async fn rx_fifo_preserves_oldest_and_drops_newest(driver: DefaultDriver) {
1008        let (io, handle) = MockSerialIo::new();
1009        let mut relay = DebuggerRelay::new(driver, "rx-drop", Box::new(io));
1010        let burst = sequence(RX_RING_CAP + 123);
1011
1012        handle.inject_rx(&burst);
1013        handle.wait_until(|state| state.read_buf.is_empty()).await;
1014
1015        let mut delivered = vec![0; RX_RING_CAP];
1016        assert!(matches!(
1017            poll_read_now(&mut relay, &mut delivered),
1018            Poll::Ready(Ok(RX_RING_CAP))
1019        ));
1020        assert_eq!(delivered, burst[..RX_RING_CAP]);
1021        assert_eq!(snapshot(&relay).rx_dropped, 123);
1022    }
1023
1024    #[async_test]
1025    async fn rx_eof_and_reconnect(driver: DefaultDriver) {
1026        let (io, handle) = MockSerialIo::new();
1027        let mut relay = DebuggerRelay::new(driver, "rx-eof", Box::new(io));
1028
1029        handle.inject_rx(&[1, 2, 3]);
1030        handle.wait_until(|state| state.read_buf.is_empty()).await;
1031        handle.set_connected(false);
1032        wait_for_relay(&relay, |state| !state.connected).await;
1033
1034        let mut delivered = [0; 8];
1035        assert!(matches!(
1036            poll_read_now(&mut relay, &mut delivered),
1037            Poll::Ready(Ok(3))
1038        ));
1039        assert_eq!(&delivered[..3], &[1, 2, 3]);
1040        assert!(matches!(
1041            poll_read_now(&mut relay, &mut delivered),
1042            Poll::Ready(Ok(0))
1043        ));
1044        assert!(matches!(poll_connect_now(&mut relay), Poll::Pending));
1045
1046        handle.set_connected(true);
1047        wait_for_relay(&relay, |state| state.connected && !state.eof).await;
1048        assert!(matches!(poll_connect_now(&mut relay), Poll::Ready(Ok(()))));
1049        assert!(relay.is_connected());
1050    }
1051
1052    #[async_test]
1053    async fn tx_never_blocks_and_drops_overflow(driver: DefaultDriver) {
1054        let (io, handle) = MockSerialIo::new();
1055        handle.set_write_limit(0); // backend stalled
1056        let mut relay = DebuggerRelay::new(driver, "tx-drop", Box::new(io));
1057
1058        // Write more than the TX ring holds; device writes must never block.
1059        let data = sequence(TX_RING_CAP + 1024);
1060        for chunk in data.chunks(1024) {
1061            assert!(matches!(
1062                poll_write_now(&mut relay, chunk),
1063                Poll::Ready(Ok(n)) if n == chunk.len()
1064            ));
1065        }
1066
1067        let snapshot = snapshot(&relay);
1068        assert!(snapshot.tx_depth <= TX_RING_CAP);
1069        assert_eq!(snapshot.tx_dropped, 1024);
1070
1071        // Unstall the backend: the retained bytes must be the earliest ones
1072        // (drop-newest), forwarded in order.
1073        handle.set_write_limit(usize::MAX);
1074        handle
1075            .wait_until(|state| state.written.len() >= TX_RING_CAP)
1076            .await;
1077        assert_eq!(handle.written(), data[..TX_RING_CAP]);
1078    }
1079
1080    /// A backend that accepts zero bytes on a non-empty write must be treated as
1081    /// a disconnect by the pump, not an infinite no-progress loop.
1082    #[async_test]
1083    async fn tx_write_zero_disconnects(driver: DefaultDriver) {
1084        let (io, handle) = MockSerialIo::new();
1085        handle.set_write_zero();
1086        let mut relay = DebuggerRelay::new(driver, "tx-write-zero", Box::new(io));
1087
1088        // The device-facing write still never blocks.
1089        assert!(matches!(
1090            poll_write_now(&mut relay, b"windbg"),
1091            Poll::Ready(Ok(6))
1092        ));
1093        // The pump forwards, receives Ok(0), and disconnects rather than spinning.
1094        wait_for_relay(&relay, |state| !state.connected).await;
1095        assert!(!relay.is_connected());
1096    }
1097
1098    /// The relay must wake a device that is blocked in `poll_read` when new RX
1099    /// arrives. Guards against a lost wakeup that would hang guest input forever.
1100    #[async_test]
1101    async fn device_read_waker_is_woken_on_new_rx(driver: DefaultDriver) {
1102        let (io, handle) = MockSerialIo::new();
1103        let mut relay = DebuggerRelay::new(driver, "rx-wake", Box::new(io));
1104        let (counter, waker) = CountingWaker::new();
1105
1106        // With the relay empty, the device read is Pending and registers `waker`.
1107        let mut buf = [0u8; 4];
1108        assert!(matches!(
1109            Pin::new(&mut relay).poll_read(&mut Context::from_waker(&waker), &mut buf),
1110            Poll::Pending
1111        ));
1112        assert_eq!(counter.count(), 0);
1113
1114        // New RX arrives at the backend; the pump must drain it and wake `waker`.
1115        handle.inject_rx(b"hi");
1116        poll_fn(|cx| {
1117            if counter.count() >= 1 {
1118                Poll::Ready(())
1119            } else {
1120                cx.waker().wake_by_ref();
1121                Poll::Pending
1122            }
1123        })
1124        .await;
1125
1126        // And the bytes are now readable.
1127        assert!(matches!(
1128            Pin::new(&mut relay).poll_read(&mut Context::from_waker(&waker), &mut buf),
1129            Poll::Ready(Ok(2))
1130        ));
1131        assert_eq!(&buf[..2], b"hi");
1132    }
1133
1134    #[async_test]
1135    async fn tx_forwards_in_fifo_order(driver: DefaultDriver) {
1136        let (io, handle) = MockSerialIo::new();
1137        handle.set_write_limit(3);
1138        let mut relay = DebuggerRelay::new(driver, "tx-forward", Box::new(io));
1139        let data = sequence(97);
1140
1141        assert!(matches!(
1142            poll_write_now(&mut relay, &data),
1143            Poll::Ready(Ok(97))
1144        ));
1145        handle
1146            .wait_until(|state| state.written.len() == data.len())
1147            .await;
1148
1149        assert_eq!(handle.written(), data);
1150    }
1151
1152    #[async_test]
1153    async fn tx_error_disconnects_without_blocking(driver: DefaultDriver) {
1154        let (io, handle) = MockSerialIo::new();
1155        handle.fail_writes_with_disconnect();
1156        let mut relay = DebuggerRelay::new(driver, "tx-error", Box::new(io));
1157
1158        assert!(matches!(
1159            poll_write_now(&mut relay, b"debug"),
1160            Poll::Ready(Ok(5))
1161        ));
1162        wait_for_relay(&relay, |state| !state.connected).await;
1163        assert!(!relay.is_connected());
1164        assert!(matches!(
1165            poll_write_now(&mut relay, b"still accepted"),
1166            Poll::Ready(Ok(14))
1167        ));
1168    }
1169
1170    #[async_test]
1171    async fn dropping_relay_cancels_pump_and_drops_inner(driver: DefaultDriver) {
1172        let (io, handle) = MockSerialIo::new();
1173        let relay = DebuggerRelay::new(driver, "teardown", Box::new(io));
1174
1175        drop(relay);
1176        handle.wait_until(|state| state.dropped).await;
1177    }
1178
1179    /// TX must keep being forwarded to the backend even while the RX side is a
1180    /// firehose that is always ready. Driven by a single manual poll so it does
1181    /// not depend on executor fairness: within one scheduling turn the pump must
1182    /// service TX, not drain RX forever. The read-poll guard turns a regressed
1183    /// spin into a clean panic instead of a hang.
1184    #[test]
1185    fn tx_is_serviced_even_with_always_ready_rx() {
1186        let shared = new_shared(true);
1187        // Queue guest->host TX directly in the ring.
1188        shared.lock().tx.push_drop_newest(b"windbg-tx");
1189
1190        let (io, handle) = MockSerialIo::new();
1191        handle.set_infinite_read();
1192        handle.panic_after_read_polls(PUMP_POLL_BUDGET as usize + 16);
1193
1194        let fut = run_pump(Box::new(io), shared.clone(), None);
1195        let mut fut = std::pin::pin!(fut);
1196        let (_counter, waker) = CountingWaker::new();
1197        let mut cx = Context::from_waker(&waker);
1198
1199        // One scheduling turn is enough to forward the queued TX despite the RX
1200        // firehose.
1201        let _ = fut.as_mut().poll(&mut cx);
1202        assert_eq!(handle.written(), b"windbg-tx");
1203        // And RX was concurrently drained/dropped (both directions ran).
1204        assert!(shared.lock().rx_dropped > 0);
1205    }
1206
1207    /// A backend that is already disconnected at construction must be reflected:
1208    /// the relay reports disconnected, reads return EOF, and a later connect is
1209    /// observable. Guards the `connected`/`eof` initialization in `new`.
1210    #[async_test]
1211    async fn reflects_backend_disconnected_at_construction(driver: DefaultDriver) {
1212        let (io, handle) = MockSerialIo::new();
1213        handle.set_connected(false);
1214        let mut relay = DebuggerRelay::new(driver, "start-disconnected", Box::new(io));
1215
1216        assert!(!relay.is_connected());
1217        let mut buf = [0; 4];
1218        assert!(matches!(
1219            poll_read_now(&mut relay, &mut buf),
1220            Poll::Ready(Ok(0))
1221        ));
1222        assert!(matches!(poll_connect_now(&mut relay), Poll::Pending));
1223
1224        // Once the backend connects, the relay follows.
1225        handle.set_connected(true);
1226        wait_for_relay(&relay, |state| state.connected && !state.eof).await;
1227        assert!(relay.is_connected());
1228        assert!(matches!(poll_connect_now(&mut relay), Poll::Ready(Ok(()))));
1229    }
1230
1231    /// `apply_debugger_mode` wraps the backend in a relay only when enabled: a
1232    /// stalled backend blocks device writes when passed through, but the relay
1233    /// accepts them immediately when debugger mode is on.
1234    #[async_test]
1235    async fn apply_debugger_mode_wraps_only_when_enabled(driver: DefaultDriver) {
1236        let (io, handle) = MockSerialIo::new();
1237        handle.set_write_limit(0); // backend writes stall
1238        let mut passthrough = apply_debugger_mode(false, driver.clone(), "serial", Box::new(io));
1239        let mut cx = Context::from_waker(Waker::noop());
1240        assert!(matches!(
1241            Pin::new(&mut passthrough).poll_write(&mut cx, b"x"),
1242            Poll::Pending
1243        ));
1244
1245        let (io, _handle) = MockSerialIo::new();
1246        let mut wrapped = apply_debugger_mode(true, driver, "serial", Box::new(io));
1247        let mut cx = Context::from_waker(Waker::noop());
1248        assert!(matches!(
1249            Pin::new(&mut wrapped).poll_write(&mut cx, b"x"),
1250            Poll::Ready(Ok(1))
1251        ));
1252    }
1253}
serial_core/debugger.rs

serial_core/
debugger.rs
