1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159
// Copyright 2022 The ChromiumOS Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
use std::collections::VecDeque;
use std::io;
use std::sync::Arc;
use std::time::Duration;
use std::time::Instant;
use base::error;
use base::Event;
use base::EventToken;
use base::FileSync;
use base::RawDescriptor;
use base::WaitContext;
use base::WorkerThread;
use sync::Mutex;
use crate::serial::sys::InStreamType;
use crate::serial_device::SerialInput;
use crate::serial_device::SerialOptions;
use crate::virtio::console::Console;
use crate::virtio::ProtectionType;
use crate::SerialDevice;
impl SerialDevice for Console {
fn new(
protection_type: ProtectionType,
_event: Event,
input: Option<Box<dyn SerialInput>>,
out: Option<Box<dyn io::Write + Send>>,
// TODO(b/171331752): connect filesync functionality.
_sync: Option<Box<dyn FileSync + Send>>,
options: SerialOptions,
keep_rds: Vec<RawDescriptor>,
) -> Console {
Console::new(protection_type, input, out, keep_rds, options.pci_address)
}
}
fn is_a_fatal_input_error(e: &io::Error) -> bool {
e.kind() != io::ErrorKind::Interrupted
}
/// Starts a thread that reads rx and sends the input back via the returned buffer.
///
/// The caller should listen on `in_avail_evt` for events. When `in_avail_evt` signals that data
/// is available, the caller should lock the returned `Mutex` and read data out of the inner
/// `VecDeque`. The data should be removed from the beginning of the `VecDeque` as it is processed.
///
/// # Arguments
///
/// * `rx` - Data source that the reader thread will wait on to send data back to the buffer
/// * `in_avail_evt` - Event triggered by the thread when new input is available on the buffer
pub(in crate::virtio::console) fn spawn_input_thread(
mut rx: InStreamType,
in_avail_evt: &Event,
input_buffer: VecDeque<u8>,
) -> (Arc<Mutex<VecDeque<u8>>>, WorkerThread<InStreamType>) {
let buffer = Arc::new(Mutex::new(input_buffer));
let buffer_cloned = buffer.clone();
let thread_in_avail_evt = in_avail_evt
.try_clone()
.expect("failed to clone in_avail_evt");
let res = WorkerThread::start("v_console_input", move |kill_evt| {
// If there is already data, signal immediately.
if !buffer.lock().is_empty() {
thread_in_avail_evt.signal().unwrap();
}
read_input(&mut rx, &thread_in_avail_evt, buffer, kill_evt);
rx
});
(buffer_cloned, res)
}
pub(in crate::virtio::console) fn read_input(
rx: &mut InStreamType,
thread_in_avail_evt: &Event,
buffer: Arc<Mutex<VecDeque<u8>>>,
kill_evt: Event,
) {
#[derive(EventToken)]
enum Token {
ConsoleEvent,
Kill,
}
let wait_ctx: WaitContext<Token> = match WaitContext::build_with(&[
(&kill_evt, Token::Kill),
(rx.get_read_notifier(), Token::ConsoleEvent),
]) {
Ok(ctx) => ctx,
Err(e) => {
error!("failed creating WaitContext {:?}", e);
return;
}
};
let mut kill_timeout = None;
let mut rx_buf = [0u8; 1 << 12];
'wait: loop {
let events = match wait_ctx.wait() {
Ok(events) => events,
Err(e) => {
error!("Failed to wait for events. {}", e);
return;
}
};
for event in events.iter() {
match event.token {
Token::Kill => {
// Ignore the kill event until there are no other events to process so that
// we drain `rx` as much as possible. The next `wait_ctx.wait()` call will
// immediately re-entry this case since we don't call `kill_evt.wait()`.
if events.iter().all(|e| matches!(e.token, Token::Kill)) {
break 'wait;
}
const TIMEOUT_DURATION: Duration = Duration::from_millis(500);
match kill_timeout {
None => {
kill_timeout = Some(Instant::now() + TIMEOUT_DURATION);
}
Some(t) => {
if Instant::now() >= t {
error!(
"failed to drain console input within {:?}, giving up",
TIMEOUT_DURATION
);
break 'wait;
}
}
}
}
Token::ConsoleEvent => {
match rx.read(&mut rx_buf) {
Ok(0) => break 'wait, // Assume the stream of input has ended.
Ok(size) => {
buffer.lock().extend(&rx_buf[0..size]);
thread_in_avail_evt.signal().unwrap();
}
Err(e) => {
// Being interrupted is not an error, but everything else is.
if is_a_fatal_input_error(&e) {
error!(
"failed to read for bytes to queue into console device: {}",
e
);
break 'wait;
}
}
}
}
}
}
}
}