[bug] Avoid holding the pipeline lock across management uart writes

lib/propolis/src/hw/virtio/softnpu.rs

@@ -8,7 +8,7 @@ use std::{
     io::{Result, Write},
     sync::{Arc, Mutex},
     thread::{sleep, spawn},
-    time::Duration,
+    time::{Duration, Instant},
 };
 
 use crate::{
@@ -244,6 +244,7 @@ impl SoftNpu {
         log: Logger,
     ) {
         info!(log, "management handler thread started");
+        let mut needs_resync = false;
         loop {
             let r = ManagementMessageReader::new(uart.clone(), log.clone());
             let msg = r.read();
@@ -252,7 +253,14 @@ impl SoftNpu {
             let pipeline = pipeline.clone();
             let uart = uart.clone();
             let log = log.clone();
-            handle_management_message(msg, pipeline, uart, radix, log.clone());
+            handle_management_message(
+                msg,
+                pipeline,
+                uart,
+                radix,
+                &mut needs_resync,
+                log.clone(),
+            );
             info!(log, "handled management message");
         }
     }
@@ -664,18 +672,77 @@ fn read_buf(mem: &MemCtx, chain: &mut Chain, buf: &mut [u8]) -> usize {
     })
 }
 
+/// Write each byte of `buf` to the uart, yielding while the one-byte FIFO
+/// is full. Gives up once `deadline` passes.
+///
+/// Returns the number of bytes written.
+fn write_with_deadline(uart: &LpcUart, buf: &[u8], deadline: Instant) -> usize {
+    for (i, b) in buf.iter().enumerate() {
+        while !uart.write(*b) {
+            if Instant::now() >= deadline {
+                return i;
+            }
+            // If we cannot write to the uart, yield and come back once
+            // scheduled again.
+            std::thread::yield_now();
+        }
+    }
+    buf.len()
+}
+
+/// Write a response buffer to the management uart, yielding while the guest
+/// drains the FIFO. This gives up once a deadline passes so that a guest that
+/// has stopped reading the management tty cannot pin this thread forever.
+///
+/// A timed out write can leave a partial, unterminated frame in the tty.
+/// `needs_resync` carries that fact to the next call, which writes a newline
+/// first to terminate the stale frame so this response starts cleanly.
+///
+/// Returns true if the full buffer was written.
+fn write_management_response(
+    uart: &LpcUart,
+    buf: &[u8],
+    needs_resync: &mut bool,
+    log: &Logger,
+) -> bool {
+    // The management protocol has no client side timeout to inherit: scadm
+    // reads the tty with a blocking loop and a 10 KiB buffer, which also
+    // bounds the largest usable response. Hitting this deadline means the guest
+    // has stopped reading (not that it is slow).
+    const WRITE_TIMEOUT: Duration = Duration::from_secs(30);
+    let deadline = Instant::now() + WRITE_TIMEOUT;
+
+    if *needs_resync {
+        if write_with_deadline(uart, b"\n", deadline) != 1 {
+            warn!(log, "management uart write timed out, dropping response");
+            return false;
+        }
+        *needs_resync = false;
+    }
+
+    let written = write_with_deadline(uart, buf, deadline);
+    if written == buf.len() {
+        return true;
+    }
+    if written > 0 {
+        *needs_resync = true;
+    }
+    warn!(log, "management uart write timed out, dropping response");
+    false
+}
+
 /// Handle ASIC management messages from the guest using the loaded program.
 fn handle_management_message(
     msg: ManagementRequest,
     pipeline: Arc<Mutex<Option<LoadedP4Program>>>,
     uart: Arc<LpcUart>,
     radix: usize,
+    needs_resync: &mut bool,
     log: Logger,
 ) {
-    let mut pl_opt = pipeline.lock().unwrap();
-
     match msg {
         ManagementRequest::TableAdd(tm) => {
+            let mut pl_opt = pipeline.lock().unwrap();
             let pl = match &mut *pl_opt {
                 Some(pl) => pl,
                 None => return,
@@ -689,6 +756,7 @@ fn handle_management_message(
             );
         }
         ManagementRequest::TableRemove(tm) => {
+            let mut pl_opt = pipeline.lock().unwrap();
             let pl = match &mut *pl_opt {
                 Some(pl) => pl,
                 None => return,
@@ -705,16 +773,22 @@ fn handle_management_message(
             let mut buf: Vec<u8> = Vec::new();
             buf.extend_from_slice(radix.to_string().as_bytes());
             buf.push(b'\n');
-            for b in &buf {
-                while !uart.write(*b) {
-                    std::thread::yield_now();
-                }
+            if write_management_response(&uart, &buf, needs_resync, &log) {
+                info!(log, "wrote: {:?}", buf.len());
             }
-            info!(log, "wrote: {:?}", buf.len());
         }
         ManagementRequest::DumpRequest => {
             info!(log, "dumping state");
+            // Collect the table state under the pipeline lock, then serialize
+            // and write the response after releasing it. Holding the lock
+            // across the uart write loop below can deadlock the whole guest
+            // because if the guest stops draining the management tty, this
+            // thread spins with the lock held while a vcpu servicing a queue
+            // notify blocks on the same lock in process_guest_packet. That
+            // vcpu never leaves its PIO exit, the guest loses a CPU, and the
+            // tty is never drained.
             let result = {
+                let mut pl_opt = pipeline.lock().unwrap();
                 let pl = match &mut *pl_opt {
                     Some(pl) => &pl.1,
                     None => return,
@@ -726,34 +800,30 @@ fn handle_management_message(
 
                 for id in pl.get_table_ids() {
                     let entries = pl.get_table_entries(id);
-                    result.insert(id, entries);
+                    // The table ids borrow from the pipeline, so own them to
+                    // let the map outlive the lock.
+                    result.insert(id.to_owned(), entries);
                 }
                 result
             };
 
             let buf = match serde_json::to_string(&result) {
                 Ok(j) => {
                     let mut buf = j.as_bytes().to_vec();
-                    info!(log, "writing: {}", j);
+                    info!(log, "writing: {j}");
                     // Add trailing newline for proper tty handling.
                     buf.push(b'\n');
                     buf
                 }
                 Err(e) => {
-                    warn!(log, "failed to serialize table state: {}", e);
+                    warn!(log, "failed to serialize table state: {e}");
                     b"{}\n".to_vec()
                 }
             };
 
-            for b in &buf {
-                while !uart.write(*b) {
-                    // If we cannot write to the uart, yield and come back once
-                    // scheduled again.
-                    std::thread::yield_now();
-                }
+            if write_management_response(&uart, &buf, needs_resync, &log) {
+                info!(log, "management wrote: {}", buf.len());
             }
-
-            info!(log, "management wrote: {}", buf.len());
         }
     }
 }