devices: remove user_command from proxy device

The only device that used user_command was Serial. This change makes
Serial device use a thread to read from its input instead of using
user_command.

BUG=chromium:1033787
TEST=./build_test
     run crosvm with stdio serial with and without sandbox

Change-Id: Ia0f2ee83d94ad2fee3f1f4f89aa734b976e33507
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/platform/crosvm/+/1966435
Tested-by: Zach Reizner <zachr@chromium.org>
Tested-by: kokoro <noreply+kokoro@google.com>
Reviewed-by: Dylan Reid <dgreid@chromium.org>
Commit-Queue: Zach Reizner <zachr@chromium.org>
Auto-Submit: Zach Reizner <zachr@chromium.org>

9 files changed, 212 insertions, 198 deletions

diff --git a/aarch64/src/lib.rs b/aarch64/src/lib.rs
index 1a3b049..13855a6 100644
--- a/aarch64/src/lib.rs
+++ b/aarch64/src/lib.rs
@@ -251,7 +251,7 @@ impl arch::LinuxArch for AArch64 {
 
         let com_evt_1_3 = EventFd::new().map_err(Error::CreateEventFd)?;
         let com_evt_2_4 = EventFd::new().map_err(Error::CreateEventFd)?;
-        let (stdio_serial_num, stdio_serial) = arch::add_serial_devices(
+        let stdio_serial_num = arch::add_serial_devices(
             &mut mmio_bus,
             &com_evt_1_3,
             &com_evt_2_4,
@@ -306,7 +306,6 @@ impl arch::LinuxArch for AArch64 {
             vm,
             kvm,
             resources,
-            stdio_serial,
             exit_evt,
             vcpus,
             vcpu_affinity,
diff --git a/arch/src/lib.rs b/arch/src/lib.rs
index 7337afb..ca9cd66 100644
--- a/arch/src/lib.rs
+++ b/arch/src/lib.rs
@@ -16,13 +16,13 @@ use std::sync::Arc;
 use devices::virtio::VirtioDevice;
 use devices::{
     Bus, BusDevice, BusError, PciDevice, PciDeviceError, PciInterruptPin, PciRoot, ProxyDevice,
-    SerialInput, SerialParameters, DEFAULT_SERIAL_PARAMS, SERIAL_ADDR,
+    SerialParameters, DEFAULT_SERIAL_PARAMS, SERIAL_ADDR,
 };
 use io_jail::Minijail;
 use kvm::{IoeventAddress, Kvm, Vcpu, Vm};
 use resources::SystemAllocator;
 use sync::Mutex;
-use sys_util::{pipe, poll_in, syslog, warn, EventFd, GuestAddress, GuestMemory, GuestMemoryError};
+use sys_util::{syslog, EventFd, GuestAddress, GuestMemory, GuestMemoryError};
 
 pub enum VmImage {
     Kernel(File),
@@ -47,7 +47,6 @@ pub struct RunnableLinuxVm {
     pub vm: Vm,
     pub kvm: Kvm,
     pub resources: SystemAllocator,
-    pub stdio_serial: Option<SerialInput>,
     pub exit_evt: EventFd,
     pub vcpus: Vec<Vcpu>,
     pub vcpu_affinity: Vec<usize>,
@@ -245,9 +244,8 @@ pub fn add_serial_devices(
     com_evt_2_4: &EventFd,
     serial_parameters: &BTreeMap<u8, SerialParameters>,
     serial_jail: Option<Minijail>,
-) -> Result<(Option<u8>, Option<SerialInput>), DeviceRegistrationError> {
+) -> Result<Option<u8>, DeviceRegistrationError> {
     let mut stdio_serial_num = None;
-    let mut stdio_serial = None;
 
     for x in 0..=3 {
         let com_evt = match x {
@@ -273,45 +271,24 @@ pub fn add_serial_devices(
 
         match serial_jail.as_ref() {
             Some(jail) => {
-                let (rx, tx) = pipe(true).map_err(DeviceRegistrationError::CreatePipe)?;
-                preserved_fds.push(rx.as_raw_fd());
                 let com = Arc::new(Mutex::new(
-                    ProxyDevice::new_with_user_command(com, &jail, preserved_fds, move |serial| {
-                        let mut rx_buf = [0u8; 32];
-                        // This loop may end up stealing bytes from future user callbacks, so we
-                        // check to make sure the pipe is readable so as not to block the proxy
-                        // device's loop.
-                        while poll_in(&rx) {
-                            if let Ok(count) = (&rx).read(&mut rx_buf) {
-                                if let Err(e) = serial.queue_input_bytes(&rx_buf[..count]) {
-                                    warn!("failed to queue bytes into serial device {}: {}", x, e);
-                                }
-                            }
-                        }
-                    })
-                    .map_err(DeviceRegistrationError::ProxyDeviceCreation)?,
+                    ProxyDevice::new(com, &jail, preserved_fds)
+                        .map_err(DeviceRegistrationError::ProxyDeviceCreation)?,
                 ));
                 io_bus
                     .insert(com.clone(), SERIAL_ADDR[x as usize], 0x8, false)
                     .unwrap();
-                if param.stdin {
-                    stdio_serial = Some(SerialInput::new_remote(tx, com));
-                }
             }
             None => {
                 let com = Arc::new(Mutex::new(com));
                 io_bus
                     .insert(com.clone(), SERIAL_ADDR[x as usize], 0x8, false)
                     .unwrap();
-
-                if param.stdin {
-                    stdio_serial = Some(SerialInput::new_local(com));
-                }
             }
         }
     }
 
-    Ok((stdio_serial_num, stdio_serial))
+    Ok(stdio_serial_num)
 }
 
 /// Errors for image loading.
diff --git a/devices/src/lib.rs b/devices/src/lib.rs
index d96ce19..512d08b 100644
--- a/devices/src/lib.rs
+++ b/devices/src/lib.rs
@@ -38,8 +38,7 @@ pub use self::proxy::Error as ProxyError;
 pub use self::proxy::ProxyDevice;
 pub use self::serial::Error as SerialError;
 pub use self::serial::{
-    get_serial_tty_string, Serial, SerialInput, SerialParameters, SerialType,
-    DEFAULT_SERIAL_PARAMS, SERIAL_ADDR,
+    get_serial_tty_string, Serial, SerialParameters, SerialType, DEFAULT_SERIAL_PARAMS, SERIAL_ADDR,
 };
 pub use self::usb::host_backend::host_backend_device_provider::HostBackendDeviceProvider;
 pub use self::usb::xhci::xhci_controller::XhciController;
diff --git a/devices/src/proxy.rs b/devices/src/proxy.rs
index 927ea7a..cb7ebd6 100644
--- a/devices/src/proxy.rs
+++ b/devices/src/proxy.rs
@@ -56,7 +56,6 @@ enum Command {
         data: [u8; 4],
     },
     Shutdown,
-    RunUserCommand,
 }
 
 #[derive(MsgOnSocket)]
@@ -66,11 +65,7 @@ enum CommandResult {
     ReadConfigResult(u32),
 }
 
-fn child_proc<D: BusDevice, F: FnMut(&mut D)>(
-    sock: UnixSeqpacket,
-    device: &mut D,
-    mut user_command: F,
-) {
+fn child_proc<D: BusDevice>(sock: UnixSeqpacket, device: &mut D) {
     let mut running = true;
     let sock = MsgSocket::<CommandResult, Command>::new(sock);
 
@@ -114,10 +109,6 @@ fn child_proc<D: BusDevice, F: FnMut(&mut D)>(
                 running = false;
                 sock.send(&CommandResult::Ok)
             }
-            Command::RunUserCommand => {
-                user_command(device);
-                sock.send(&CommandResult::Ok)
-            }
         };
         if let Err(e) = res {
             error!("child device process failed send: {}", e);
@@ -146,31 +137,9 @@ impl ProxyDevice {
     /// * `jail` - The jail to use for isolating the given device.
     /// * `keep_fds` - File descriptors that will be kept open in the child.
     pub fn new<D: BusDevice>(
-        device: D,
-        jail: &Minijail,
-        keep_fds: Vec<RawFd>,
-    ) -> Result<ProxyDevice> {
-        Self::new_with_user_command(device, jail, keep_fds, |_| {})
-    }
-
-    /// Similar to `ProxyDevice::new`, but adds an additional custom command to be run in the forked
-    /// process when `run_user_command` is called.
-    ///
-    /// Note that the custom command closure is run in the main thread of the child process, which
-    /// also services `BusDevice` requests. Therefore, do not run blocking calls in the closure
-    /// without a timeout, or you will block any VCPU which touches this device, and every other
-    /// thread which needs to lock this device's mutex.
-    ///
-    /// # Arguments
-    /// * `device` - The device to isolate to another process.
-    /// * `jail` - The jail to use for isolating the given device.
-    /// * `keep_fds` - File descriptors that will be kept open in the child.
-    /// * `user_command` - Closure to be run in the forked process.
-    pub fn new_with_user_command<D: BusDevice, F: FnMut(&mut D)>(
         mut device: D,
         jail: &Minijail,
         mut keep_fds: Vec<RawFd>,
-        user_command: F,
     ) -> Result<ProxyDevice> {
         let debug_label = device.debug_label();
         let (child_sock, parent_sock) = UnixSeqpacket::pair().map_err(Error::Io)?;
@@ -181,7 +150,7 @@ impl ProxyDevice {
             match jail.fork(Some(&keep_fds)).map_err(Error::ForkingJail)? {
                 0 => {
                     device.on_sandboxed();
-                    child_proc(child_sock, &mut device, user_command);
+                    child_proc(child_sock, &mut device);
 
                     // We're explicitly not using std::process::exit here to avoid the cleanup of
                     // stdout/stderr globals. This can cause cascading panics and SIGILL if a worker
@@ -214,11 +183,6 @@ impl ProxyDevice {
         self.pid
     }
 
-    /// Runs the callback given in `new_with_custom_command` in the child device process.
-    pub fn run_user_command(&self) {
-        self.sync_send(&Command::RunUserCommand);
-    }
-
     /// Send a command that does not expect a response from the child device process.
     fn send_no_result(&self, cmd: &Command) {
         let res = self.sock.send(cmd);
diff --git a/devices/src/serial.rs b/devices/src/serial.rs
index 9485497..98f39dc 100644
--- a/devices/src/serial.rs
+++ b/devices/src/serial.rs
@@ -5,17 +5,18 @@
 use std::collections::VecDeque;
 use std::fmt::{self, Display};
 use std::fs::File;
-use std::io::{self, stdout, Write};
+use std::io::{self, stdin, stdout, Read, Write};
 use std::os::unix::io::{AsRawFd, RawFd};
 use std::path::PathBuf;
 use std::str::FromStr;
+use std::sync::atomic::{AtomicU8, Ordering};
+use std::sync::mpsc::{channel, Receiver, TryRecvError};
 use std::sync::Arc;
+use std::thread::{self};
 
-use sync::Mutex;
-use sys_util::{error, syslog, EventFd, Result};
+use sys_util::{error, read_raw_stdin, syslog, EventFd, Result};
 
 use crate::BusDevice;
-use crate::ProxyDevice;
 
 const LOOP_SIZE: usize = 0x40;
 
@@ -151,7 +152,24 @@ impl SerialParameters {
         match self.type_ {
             SerialType::Stdout => {
                 keep_fds.push(stdout().as_raw_fd());
-                Ok(Serial::new_out(evt_fd, Box::new(stdout())))
+                if self.stdin {
+                    keep_fds.push(stdin().as_raw_fd());
+                    // This wrapper is used in place of the libstd native version because we don't
+                    // want buffering for stdin.
+                    struct StdinWrapper;
+                    impl io::Read for StdinWrapper {
+                        fn read(&mut self, out: &mut [u8]) -> io::Result<usize> {
+                            read_raw_stdin(out).map_err(|e| e.into())
+                        }
+                    }
+                    Ok(Serial::new_in_out(
+                        evt_fd,
+                        Box::new(StdinWrapper),
+                        Box::new(stdout()),
+                    ))
+                } else {
+                    Ok(Serial::new_out(evt_fd, Box::new(stdout())))
+                }
             }
             SerialType::Sink => Ok(Serial::new_sink(evt_fd)),
             SerialType::Syslog => {
@@ -227,56 +245,14 @@ pub fn get_serial_tty_string(stdio_serial_num: u8) -> String {
     }
 }
 
-/// A type for queueing input bytes to a serial device that abstracts if the device is local or part
-/// of a sandboxed device process.
-pub enum SerialInput {
-    #[doc(hidden)]
-    Local(Arc<Mutex<Serial>>),
-    #[doc(hidden)]
-    Remote {
-        input: File,
-        proxy: Arc<Mutex<ProxyDevice>>,
-    },
-}
-
-impl SerialInput {
-    /// Creates a `SerialInput` that trivially forwards queued bytes to the device in the local
-    /// process.
-    pub fn new_local(serial: Arc<Mutex<Serial>>) -> SerialInput {
-        SerialInput::Local(serial)
-    }
-
-    /// Creates a `SerialInput` that will forward bytes via `input` and will wake up the child
-    /// device process which contains the serial device so that it may process the other end of the
-    /// `input` pipe. This will use the `ProxyDevice::run_user_command` method, so the user command
-    /// closure (registered with `ProxyDevice::new_with_user_command`) should own the other side of
-    /// `input` and read from the pipe whenever the closure is run.
-    pub fn new_remote(input: File, proxy: Arc<Mutex<ProxyDevice>>) -> SerialInput {
-        SerialInput::Remote { input, proxy }
-    }
-
-    /// Just like `Serial::queue_input_bytes`, but abstracted over local and sandboxed serial
-    /// devices. In the case that the serial device is sandboxed, this method will also trigger the
-    /// user command in the `ProxyDevice`'s child process.
-    pub fn queue_input_bytes(&self, bytes: &[u8]) -> Result<()> {
-        match self {
-            SerialInput::Local(device) => device.lock().queue_input_bytes(bytes),
-            SerialInput::Remote { input, proxy } => {
-                let mut input = input;
-                input.write_all(bytes)?;
-                proxy.lock().run_user_command();
-                Ok(())
-            }
-        }
-    }
-}
-
 /// Emulates serial COM ports commonly seen on x86 I/O ports 0x3f8/0x2f8/0x3e8/0x2e8.
 ///
 /// This can optionally write the guest's output to a Write trait object. To send input to the
-/// guest, use `queue_input_bytes`.
+/// guest, use `queue_input_bytes` directly, or give a Read trait object which will be used queue
+/// bytes when `used_command` is called.
 pub struct Serial {
-    interrupt_enable: u8,
+    // Serial port registers
+    interrupt_enable: Arc<AtomicU8>,
     interrupt_identification: u8,
     interrupt_evt: EventFd,
     line_control: u8,
@@ -285,14 +261,22 @@ pub struct Serial {
     modem_status: u8,
     scratch: u8,
     baud_divisor: u16,
+
+    // Host input/output
     in_buffer: VecDeque<u8>,
+    in_channel: Option<Receiver<u8>>,
+    input: Option<Box<dyn io::Read + Send>>,
     out: Option<Box<dyn io::Write + Send>>,
 }
 
 impl Serial {
-    fn new(interrupt_evt: EventFd, out: Option<Box<dyn io::Write + Send>>) -> Serial {
+    fn new(
+        interrupt_evt: EventFd,
+        input: Option<Box<dyn io::Read + Send>>,
+        out: Option<Box<dyn io::Write + Send>>,
+    ) -> Serial {
         Serial {
-            interrupt_enable: 0,
+            interrupt_enable: Default::default(),
             interrupt_identification: DEFAULT_INTERRUPT_IDENTIFICATION,
             interrupt_evt,
             line_control: DEFAULT_LINE_CONTROL,
@@ -301,31 +285,130 @@ impl Serial {
             modem_status: DEFAULT_MODEM_STATUS,
             scratch: 0,
             baud_divisor: DEFAULT_BAUD_DIVISOR,
-            in_buffer: VecDeque::new(),
+            in_buffer: Default::default(),
+            in_channel: None,
+            input,
             out,
         }
     }
 
-    /// Constructs a Serial port ready for output.
+    /// Constructs a Serial port ready for input and output.
+    ///
+    /// The stream `input` should not block, instead returning 0 bytes if are no bytes available.
+    pub fn new_in_out(
+        interrupt_evt: EventFd,
+        input: Box<dyn io::Read + Send>,
+        out: Box<dyn io::Write + Send>,
+    ) -> Serial {
+        Self::new(interrupt_evt, Some(input), Some(out))
+    }
+
+    /// Constructs a Serial port ready for output but not input.
     pub fn new_out(interrupt_evt: EventFd, out: Box<dyn io::Write + Send>) -> Serial {
-        Self::new(interrupt_evt, Some(out))
+        Self::new(interrupt_evt, None, Some(out))
     }
 
-    /// Constructs a Serial port with no connected output.
+    /// Constructs a Serial port with no connected input or output.
     pub fn new_sink(interrupt_evt: EventFd) -> Serial {
-        Self::new(interrupt_evt, None)
+        Self::new(interrupt_evt, None, None)
     }
 
     /// Queues raw bytes for the guest to read and signals the interrupt if the line status would
-    /// change.
+    /// change. These bytes will be read by the guest before any bytes from the input stream that
+    /// have not already been queued.
     pub fn queue_input_bytes(&mut self, c: &[u8]) -> Result<()> {
-        if !self.is_loop() {
+        if !c.is_empty() && !self.is_loop() {
             self.in_buffer.extend(c);
-            self.recv_data()?;
+            self.set_data_bit();
+            self.trigger_recv_interrupt()?;
         }
+
         Ok(())
     }
 
+    fn spawn_input_thread(&mut self) {
+        let mut rx = match self.input.take() {
+            Some(input) => input,
+            None => return,
+        };
+
+        let (send_channel, recv_channel) = channel();
+
+        // The interrupt enable and interrupt event are used to trigger the guest serial driver to
+        // read the serial device, which will give the VCPU threads time to queue input bytes from
+        // the input thread's buffer, changing the serial device state accordingly.
+        let interrupt_enable = self.interrupt_enable.clone();
+        let interrupt_evt = match self.interrupt_evt.try_clone() {
+            Ok(e) => e,
+            Err(e) => {
+                error!("failed to clone interrupt eventfd: {}", e);
+                return;
+            }
+        };
+
+        // The input thread runs in detached mode and will exit when channel is disconnected because
+        // the serial device has been dropped. Initial versions of this kept a `JoinHandle` and had
+        // the drop implementation of serial join on this thread, but the input thread can block
+        // indefinitely depending on the `Box<io::Read>` implementation.
+        let res = thread::Builder::new()
+            .name(format!("{} input thread", self.debug_label()))
+            .spawn(move || {
+                let mut rx_buf = [0u8; 1];
+                loop {
+                    match rx.read(&mut rx_buf) {
+                        Ok(0) => break, // Assume the stream of input has ended.
+                        Ok(_) => {
+                            if send_channel.send(rx_buf[0]).is_err() {
+                                // The receiver has disconnected.
+                                break;
+                            }
+                            if (interrupt_enable.load(Ordering::SeqCst) & IER_RECV_BIT) != 0 {
+                                interrupt_evt.write(1).unwrap();
+                            }
+                        }
+                        Err(e) => {
+                            // Being interrupted is not an error, but everything else is.
+                            if e.kind() != io::ErrorKind::Interrupted {
+                                error!(
+                                    "failed to read for bytes to queue into serial device: {}",
+                                    e
+                                );
+                                break;
+                            }
+                        }
+                    }
+                }
+            });
+        if let Err(e) = res {
+            error!("failed to spawn input thread: {}", e);
+            return;
+        }
+        self.in_channel = Some(recv_channel);
+    }
+
+    fn handle_input_thread(&mut self) {
+        if self.input.is_some() {
+            self.spawn_input_thread();
+        }
+
+        loop {
+            let in_channel = match self.in_channel.as_ref() {
+                Some(v) => v,
+                None => return,
+            };
+            match in_channel.try_recv() {
+                Ok(byte) => {
+                    self.queue_input_bytes(&[byte]).unwrap();
+                }
+                Err(TryRecvError::Empty) => break,
+                Err(TryRecvError::Disconnected) => {
+                    self.in_channel = None;
+                    return;
+                }
+            }
+        }
+    }
+
     /// Gets the interrupt eventfd used to interrupt the driver when it needs to respond to this
     /// device.
     pub fn interrupt_eventfd(&self) -> &EventFd {
@@ -337,11 +420,11 @@ impl Serial {
     }
 
     fn is_recv_intr_enabled(&self) -> bool {
-        (self.interrupt_enable & IER_RECV_BIT) != 0
+        (self.interrupt_enable.load(Ordering::SeqCst) & IER_RECV_BIT) != 0
     }
 
     fn is_thr_intr_enabled(&self) -> bool {
-        (self.interrupt_enable & IER_THR_BIT) != 0
+        (self.interrupt_enable.load(Ordering::SeqCst) & IER_THR_BIT) != 0
     }
 
     fn is_loop(&self) -> bool {
@@ -360,7 +443,7 @@ impl Serial {
         }
     }
 
-    fn thr_empty(&mut self) -> Result<()> {
+    fn trigger_thr_empty(&mut self) -> Result<()> {
         if self.is_thr_intr_enabled() {
             self.add_intr_bit(IIR_THR_BIT);
             self.trigger_interrupt()?
@@ -368,12 +451,15 @@ impl Serial {
         Ok(())
     }
 
-    fn recv_data(&mut self) -> Result<()> {
+    fn trigger_recv_interrupt(&mut self) -> Result<()> {
         if self.is_recv_intr_enabled() {
-            self.add_intr_bit(IIR_RECV_BIT);
-            self.trigger_interrupt()?
+            // Only bother triggering the interrupt if the identification bit wasn't set or
+            // acknowledged.
+            if self.interrupt_identification & IIR_RECV_BIT == 0 {
+                self.add_intr_bit(IIR_RECV_BIT);
+                self.trigger_interrupt()?
+            }
         }
-        self.line_status |= LSR_DATA_BIT;
         Ok(())
     }
 
@@ -381,6 +467,10 @@ impl Serial {
         self.interrupt_evt.write(1)
     }
 
+    fn set_data_bit(&mut self) {
+        self.line_status |= LSR_DATA_BIT;
+    }
+
     fn iir_reset(&mut self) {
         self.interrupt_identification = DEFAULT_INTERRUPT_IDENTIFICATION;
     }
@@ -397,17 +487,20 @@ impl Serial {
                 if self.is_loop() {
                     if self.in_buffer.len() < LOOP_SIZE {
                         self.in_buffer.push_back(v);
-                        self.recv_data()?;
+                        self.set_data_bit();
+                        self.trigger_recv_interrupt()?;
                     }
                 } else {
                     if let Some(out) = self.out.as_mut() {
                         out.write_all(&[v])?;
                         out.flush()?;
                     }
-                    self.thr_empty()?;
+                    self.trigger_thr_empty()?;
                 }
             }
-            IER => self.interrupt_enable = v & IER_FIFO_BITS,
+            IER => self
+                .interrupt_enable
+                .store(v & IER_FIFO_BITS, Ordering::SeqCst),
             LCR => self.line_control = v,
             MCR => self.modem_control = v,
             SCR => self.scratch = v,
@@ -437,6 +530,8 @@ impl BusDevice for Serial {
             return;
         }
 
+        self.handle_input_thread();
+
         data[0] = match offset as u8 {
             DLAB_LOW if self.is_dlab_set() => self.baud_divisor as u8,
             DLAB_HIGH if self.is_dlab_set() => (self.baud_divisor >> 8) as u8,
@@ -447,7 +542,7 @@ impl BusDevice for Serial {
                 }
                 self.in_buffer.pop_front().unwrap_or_default()
             }
-            IER => self.interrupt_enable,
+            IER => self.interrupt_enable.load(Ordering::SeqCst),
             IIR => {
                 let v = self.interrupt_identification | IIR_FIFO_BITS;
                 self.iir_reset();
diff --git a/src/linux.rs b/src/linux.rs
index 2aa135d..c7e6b0b 100644
--- a/src/linux.rs
+++ b/src/linux.rs
@@ -1602,7 +1602,6 @@ fn run_control(
     #[derive(PollToken)]
     enum Token {
         Exit,
-        Stdin,
         ChildSignal,
         CheckAvailableMemory,
         LowMemory,
@@ -1611,9 +1610,7 @@ fn run_control(
         VmControl { index: usize },
     }
 
-    let stdin_handle = stdin();
-    let stdin_lock = stdin_handle.lock();
-    stdin_lock
+    stdin()
         .set_raw_mode()
         .expect("failed to set terminal raw mode");
 
@@ -1623,10 +1620,6 @@ fn run_control(
     ])
     .map_err(Error::PollContextAdd)?;
 
-    if let Err(e) = poll_ctx.add(&stdin_handle, Token::Stdin) {
-        warn!("failed to add stdin to poll context: {}", e);
-    }
-
     if let Some(socket_server) = &control_server_socket {
         poll_ctx
             .add(socket_server, Token::VmControlServer)
@@ -1715,26 +1708,6 @@ fn run_control(
                     info!("vcpu requested shutdown");
                     break 'poll;
                 }
-                Token::Stdin => {
-                    let mut out = [0u8; 64];
-                    match stdin_lock.read_raw(&mut out[..]) {
-                        Ok(0) => {
-                            // Zero-length read indicates EOF. Remove from pollables.
-                            let _ = poll_ctx.delete(&stdin_handle);
-                        }
-                        Err(e) => {
-                            warn!("error while reading stdin: {}", e);
-                            let _ = poll_ctx.delete(&stdin_handle);
-                        }
-                        Ok(count) => {
-                            if let Some(ref stdio_serial) = linux.stdio_serial {
-                                stdio_serial
-                                    .queue_input_bytes(&out[..count])
-                                    .expect("failed to queue bytes into serial port");
-                            }
-                        }
-                    }
-                }
                 Token::ChildSignal => {
                     // Print all available siginfo structs, then exit the loop.
                     while let Some(siginfo) = sigchld_fd.read().map_err(Error::SignalFd)? {
@@ -1928,9 +1901,6 @@ fn run_control(
         for event in events.iter_hungup() {
             match event.token() {
                 Token::Exit => {}
-                Token::Stdin => {
-                    let _ = poll_ctx.delete(&stdin_handle);
-                }
                 Token::ChildSignal => {}
                 Token::CheckAvailableMemory => {}
                 Token::LowMemory => {}
@@ -2003,7 +1973,7 @@ fn run_control(
     // control sockets are closed when this function exits.
     mem::drop(linux);
 
-    stdin_lock
+    stdin()
         .set_canon_mode()
         .expect("failed to restore canonical mode for terminal");
 
diff --git a/sys_util/src/errno.rs b/sys_util/src/errno.rs
index cf54aae..607d6b3 100644
--- a/sys_util/src/errno.rs
+++ b/sys_util/src/errno.rs
@@ -40,11 +40,17 @@ impl From<io::Error> for Error {
     }
 }
 
+impl From<Error> for io::Error {
+    fn from(e: Error) -> io::Error {
+        io::Error::from_raw_os_error(e.0)
+    }
+}
+
 impl std::error::Error for Error {}
 
 impl Display for Error {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        io::Error::from_raw_os_error(self.0).fmt(f)
+        Into::<io::Error>::into(*self).fmt(f)
     }
 }
 
diff --git a/sys_util/src/terminal.rs b/sys_util/src/terminal.rs
index eb5141c..89456f9 100644
--- a/sys_util/src/terminal.rs
+++ b/sys_util/src/terminal.rs
@@ -2,7 +2,7 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-use std::io::StdinLock;
+use std::io::Stdin;
 use std::mem::zeroed;
 use std::os::unix::io::RawFd;
 
@@ -37,6 +37,26 @@ fn modify_mode<F: FnOnce(&mut termios)>(fd: RawFd, f: F) -> Result<()> {
     Ok(())
 }
 
+/// Safe only when the FD given is valid and reading the fd will have no Rust safety implications.
+unsafe fn read_raw(fd: RawFd, out: &mut [u8]) -> Result<usize> {
+    let ret = read(fd, out.as_mut_ptr() as *mut _, out.len());
+    if ret < 0 {
+        return errno_result();
+    }
+
+    Ok(ret as usize)
+}
+
+/// Read raw bytes from stdin.
+///
+/// This will block depending on the underlying mode of stdin. This will ignore the usual lock
+/// around stdin that the stdlib usually uses. If other code is using stdin, it is undefined who
+/// will get the underlying bytes.
+pub fn read_raw_stdin(out: &mut [u8]) -> Result<usize> {
+    // Safe because reading from stdin shouldn't have any safety implications.
+    unsafe { read_raw(STDIN_FILENO, out) }
+}
+
 /// Trait for file descriptors that are TTYs, according to `isatty(3)`.
 ///
 /// This is marked unsafe because the implementation must promise that the returned RawFd is a valid
@@ -66,25 +86,10 @@ pub unsafe trait Terminal {
             clear_fd_flags(self.tty_fd(), O_NONBLOCK)
         }
     }
-
-    /// Reads up to `out.len()` bytes from this terminal without any buffering.
-    ///
-    /// This may block, depending on if non-blocking was enabled with `set_non_block` or if there
-    /// are any bytes to read. If there is at least one byte that is readable, this will not block.
-    fn read_raw(&self, out: &mut [u8]) -> Result<usize> {
-        // Safe because read will only modify the pointer up to the length we give it and we check
-        // the return result.
-        let ret = unsafe { read(self.tty_fd(), out.as_mut_ptr() as *mut _, out.len()) };
-        if ret < 0 {
-            return errno_result();
-        }
-
-        Ok(ret as usize)
-    }
 }
 
 // Safe because we return a genuine terminal fd that never changes and shares our lifetime.
-unsafe impl<'a> Terminal for StdinLock<'a> {
+unsafe impl Terminal for Stdin {
     fn tty_fd(&self) -> RawFd {
         STDIN_FILENO
     }
diff --git a/x86_64/src/lib.rs b/x86_64/src/lib.rs
index a2fdf27..161235a 100644
--- a/x86_64/src/lib.rs
+++ b/x86_64/src/lib.rs
@@ -366,7 +366,7 @@ impl arch::LinuxArch for X8664arch {
             components.memory_size,
         )?;
 
-        let (stdio_serial_num, stdio_serial) =
+        let stdio_serial_num =
             Self::setup_serial_devices(&mut vm, &mut io_bus, serial_parameters, serial_jail)?;
 
         match components.vm_image {
@@ -398,7 +398,6 @@ impl arch::LinuxArch for X8664arch {
             vm,
             kvm,
             resources,
-            stdio_serial,
             exit_evt,
             vcpus,
             vcpu_affinity,
@@ -712,11 +711,11 @@ impl X8664arch {
         io_bus: &mut devices::Bus,
         serial_parameters: &BTreeMap<u8, SerialParameters>,
         serial_jail: Option<Minijail>,
-    ) -> Result<(Option<u8>, Option<devices::SerialInput>)> {
+    ) -> Result<(Option<u8>)> {
         let com_evt_1_3 = EventFd::new().map_err(Error::CreateEventFd)?;
         let com_evt_2_4 = EventFd::new().map_err(Error::CreateEventFd)?;
 
-        let (stdio_serial_num, stdio_serial) = arch::add_serial_devices(
+        let stdio_serial_num = arch::add_serial_devices(
             io_bus,
             &com_evt_1_3,
             &com_evt_2_4,
@@ -730,7 +729,7 @@ impl X8664arch {
         vm.register_irqfd(&com_evt_2_4, X86_64_SERIAL_2_4_IRQ)
             .map_err(Error::RegisterIrqfd)?;
 
-        Ok((stdio_serial_num, stdio_serial))
+        Ok(stdio_serial_num)
     }
 
     /// Configures the vcpu and should be called once per vcpu from the vcpu's thread.