Revises VMM logging (#1214)

gui/src/graphics/vulkan/window.rs

@@ -47,8 +47,9 @@ impl Drop for VulkanWindow {
 }
 
 impl RuntimeWindow for VulkanWindow {
-    fn on_resized(&self, new: PhysicalSize<u32>) -> Result<(), Box<dyn Error + Send + Sync>> {
-        todo!()
+    fn on_resized(&self, _: PhysicalSize<u32>) -> Result<(), Box<dyn Error + Send + Sync>> {
+        // Vulkan windows does not allowed to resize.
+        Ok(())
     }
 
     fn on_close_requested(&self) -> Result<(), Box<dyn Error + Send + Sync>> {
@@ -83,35 +84,39 @@ impl RuntimeWindow for VulkanWindow {
 
     fn on_scale_factor_changed(
         &self,
-        new: f64,
-        sw: InnerSizeWriter,
+        _: f64,
+        _: InnerSizeWriter,
     ) -> Result<(), Box<dyn Error + Send + Sync>> {
-        todo!()
+        Ok(())
     }
 
     fn on_redraw_requested(&self) -> Result<(), Box<dyn Error + Send + Sync>> {
-        todo!()
+        self.window.request_redraw();
+        Ok(())
     }
 }
 
 impl Hook for VulkanWindow {
-    fn new_events(&self, cause: &StartCause) -> Result<(), Box<dyn Error + Send + Sync>> {
-        todo!()
+    fn new_events(&self, _: &StartCause) -> Result<(), Box<dyn Error + Send + Sync>> {
+        Ok(())
     }
 
     fn pre_window_event(&self) -> Result<(), Box<dyn Error + Send + Sync>> {
-        todo!()
+        Ok(())
     }
 
-    fn window_destroyed(&self, id: WindowId) -> Result<(), Box<dyn Error + Send + Sync>> {
-        todo!()
+    fn window_destroyed(&self, _: WindowId) -> Result<(), Box<dyn Error + Send + Sync>> {
+        // This never be our window since we live forever until the event loop exit.
+        Ok(())
     }
 
     fn post_window_event(&self) -> Result<(), Box<dyn Error + Send + Sync>> {
-        todo!()
+        Ok(())
     }
 
     fn about_to_wait(&self) -> Result<ControlFlow, Box<dyn Error + Send + Sync>> {
-        todo!()
+        // TODO: Not sure if we need Poll here since we always request a redraw when we received
+        // redraw requested.
+        Ok(ControlFlow::Wait)
     }
 }

gui/src/main.rs

@@ -13,9 +13,11 @@ use async_net::{TcpListener, TcpStream};
 use clap::{Parser, ValueEnum};
 use erdp::ErrorDisplay;
 use futures::{select_biased, AsyncReadExt, FutureExt};
+use obconf::ConsoleType;
 use slint::{ComponentHandle, ModelRc, SharedString, ToSharedString, VecModel, WindowHandle};
 use std::cell::Cell;
 use std::future::Future;
+use std::io::{stderr, stdout, Write};
 use std::net::SocketAddrV4;
 use std::path::PathBuf;
 use std::pin::pin;
@@ -277,8 +279,14 @@ async fn run(args: ProgramArgs, exe: PathBuf) -> Result<(), ProgramError> {
         if let Some(vmm) = vmm {
             match vmm {
                 VmmEvent::Exit(_, _) => todo!(),
-                VmmEvent::Log(console_type, _) => todo!(),
-                VmmEvent::Breakpoint(base_stop_reason) => todo!(),
+                VmmEvent::Log(t, m) => {
+                    let m = m.as_bytes();
+
+                    match t {
+                        ConsoleType::Info => stdout().write_all(m).unwrap(),
+                        ConsoleType::Warn | ConsoleType::Error => stderr().write_all(m).unwrap(),
+                    }
+                }
             }
         }
 

gui/src/vmm/channel/mod.rs

@@ -1,14 +1,3 @@
-pub use self::main::*;
 pub use self::vmm::*;
 
-mod main;
 mod vmm;
-
-/// Create a new channel to communicate with the VMM.
-pub fn create_channel() -> (VmmStream, MainStream) {
-    // Create streams.
-    let vmm = VmmStream::new();
-    let main = MainStream::new();
-
-    (vmm, main)
-}

gui/src/vmm/channel/vmm.rs

@@ -1,14 +1,106 @@
-use crate::vmm::VmmEvent;
+use std::collections::{BTreeMap, VecDeque};
+use std::future::Future;
+use std::num::NonZero;
+use std::pin::Pin;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Condvar, Mutex};
+use std::task::{Context, Poll, Waker};
 
 /// Provides method to send and receive events from the VMM.
-pub struct VmmStream {}
+///
+/// The main different from [`futures::channel::mpsc::channel()`] is our implementation will block
+/// the sender when the buffer is full.
+pub struct VmmStream<T> {
+    max: NonZero<usize>,
+    state: Mutex<State<T>>,
+    cv: Condvar,
+}
+
+impl<T> VmmStream<T> {
+    pub fn new(buffer: NonZero<usize>) -> Self {
+        Self {
+            max: buffer,
+            state: Mutex::new(State {
+                items: VecDeque::default(),
+                wakers: BTreeMap::default(),
+            }),
+            cv: Condvar::default(),
+        }
+    }
+
+    pub fn recv(&self) -> impl Future<Output = T> + '_ {
+        Recv {
+            stream: self,
+            pending: None,
+        }
+    }
+
+    pub fn send(&self, v: T) {
+        // Wait for available room.
+        let state = self.state.lock().unwrap();
+        let mut state = self
+            .cv
+            .wait_while(state, |s| s.items.len() >= self.max.get())
+            .unwrap();
 
-impl VmmStream {
-    pub(super) fn new() -> Self {
-        Self {}
+        // Store the value and wake one task.
+        state.items.push_back(v);
+
+        if let Some((_, w)) = state.wakers.pop_first() {
+            w.wake();
+        }
     }
+}
+
+/// Implementation of [`Future`] to receive a value from [`VmmStream`].
+struct Recv<'a, T> {
+    stream: &'a VmmStream<T>,
+    pending: Option<u64>,
+}
+
+impl<'a, T> Drop for Recv<'a, T> {
+    fn drop(&mut self) {
+        let id = match self.pending.take() {
+            Some(v) => v,
+            None => return,
+        };
 
-    pub async fn recv(&mut self) -> Option<VmmEvent> {
-        todo!()
+        self.stream.state.lock().unwrap().wakers.remove(&id);
     }
 }
+
+impl<'a, T> Future for Recv<'a, T> {
+    type Output = T;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        // Get item if available.
+        let mut state = self.stream.state.lock().unwrap();
+
+        if let Some(v) = state.items.pop_front() {
+            if let Some(id) = self.pending.take() {
+                // The future may poll without a wakeup from a waker.
+                state.wakers.remove(&id);
+            }
+
+            self.stream.cv.notify_one();
+            return Poll::Ready(v);
+        }
+
+        // Store waker.
+        let id = self
+            .pending
+            .get_or_insert_with(|| WAKER_ID.fetch_add(1, Ordering::Relaxed));
+
+        state.wakers.insert(*id, cx.waker().clone());
+
+        Poll::Pending
+    }
+}
+
+/// State of [`VmmStream`].
+struct State<T> {
+    items: VecDeque<T>,
+    wakers: BTreeMap<u64, Waker>,
+}
+
+static WAKER_ID: AtomicU64 = AtomicU64::new(0);

gui/src/vmm/hw/console/context.rs

@@ -1,7 +1,7 @@
 // SPDX-License-Identifier: MIT OR Apache-2.0
 use super::Console;
 use crate::hv::{Cpu, CpuExit, CpuIo, Hypervisor};
-use crate::vmm::channel::MainStream;
+use crate::vmm::channel::VmmStream;
 use crate::vmm::hw::{read_ptr, read_u8, read_usize, DeviceContext, MmioError};
 use obconf::{ConsoleMemory, ConsoleType};
 use std::error::Error;
@@ -13,17 +13,17 @@ use thiserror::Error;
 pub struct Context<'a, H> {
     dev: &'a Console,
     hv: &'a H,
-    main: &'a MainStream,
+    logs: &'a VmmStream<(ConsoleType, String)>,
     msg_len: Option<NonZero<usize>>,
     msg: Vec<u8>,
 }
 
 impl<'a, H> Context<'a, H> {
-    pub fn new(dev: &'a Console, hv: &'a H, main: &'a MainStream) -> Self {
+    pub fn new(dev: &'a Console, hv: &'a H, logs: &'a VmmStream<(ConsoleType, String)>) -> Self {
         Self {
             dev,
             hv,
-            main,
+            logs,
             msg_len: None,
             msg: Vec::new(),
         }
@@ -70,7 +70,7 @@ impl<H: Hypervisor, C: Cpu> DeviceContext<C> for Context<'_, H> {
             // single allocation when the handler clone the string.
             let msg = std::str::from_utf8(&self.msg).map_err(|_| ExecError::InvalidMsg)?;
 
-            self.main.log(ty, msg);
+            self.logs.send((ty, msg.to_owned()));
             self.msg.clear();
         } else {
             return Err(Box::new(ExecError::UnknownField(off)));

gui/src/vmm/hw/console/mod.rs

@@ -2,8 +2,8 @@
 use self::context::Context;
 use super::{Device, DeviceContext};
 use crate::hv::Hypervisor;
-use crate::vmm::channel::MainStream;
-use obconf::ConsoleMemory;
+use crate::vmm::channel::VmmStream;
+use obconf::{ConsoleMemory, ConsoleType};
 use std::num::NonZero;
 
 mod context;
@@ -27,9 +27,9 @@ impl Console {
     pub fn create_context<'a, H: Hypervisor>(
         &'a self,
         hv: &'a H,
-        main: &'a MainStream,
+        logs: &'a VmmStream<(ConsoleType, String)>,
     ) -> Box<dyn DeviceContext<H::Cpu<'a>> + 'a> {
-        Box::new(Context::new(self, hv, main))
+        Box::new(Context::new(self, hv, logs))
     }
 }