Update vhost-device-console

Bug-AGL: SPEC-4966
Change-Id: I41cbe30549ebe923f8e3000ece072aa66c5e90a9
Signed-off-by: Timos Ampelikiotis <t.ampelikiotis@virtualopensystems.com>

5 files changed, 1339 insertions, 679 deletions

diff --git a/meta-egvirt/recipes-extended/vhost-device-console/vhost-device-console-0.1.0/src/backend.rs b/meta-egvirt/recipes-extended/vhost-device-console/vhost-device-console-0.1.0/src/backend.rs
index 15b50a38..62d6745c 100644..100755
--- a/meta-egvirt/recipes-extended/vhost-device-console/vhost-device-console-0.1.0/src/backend.rs
+++ b/meta-egvirt/recipes-extended/vhost-device-console/vhost-device-console-0.1.0/src/backend.rs
@@ -1,22 +1,22 @@
 // VIRTIO CONSOLE Emulation via vhost-user
 //
-// Copyright 2023 VIRTUAL OPEN SYSTEMS SAS. All Rights Reserved.
+// Copyright 2023-2024 VIRTUAL OPEN SYSTEMS SAS. All Rights Reserved.
 //          Timos Ampelikiotis <t.ampelikiotis@virtualopensystems.com>
 //
 // SPDX-License-Identifier: Apache-2.0 or BSD-3-Clause
 
 use log::{error, info, warn};
-use std::process::exit;
+use std::any::Any;
+use std::collections::HashMap;
+use std::path::PathBuf;
 use std::sync::{Arc, RwLock};
-use std::thread::{spawn, JoinHandle};
+use std::thread::Builder;
 
-use clap::Parser;
 use thiserror::Error as ThisError;
-use vhost::{vhost_user, vhost_user::Listener};
 use vhost_user_backend::VhostUserDaemon;
 use vm_memory::{GuestMemoryAtomic, GuestMemoryMmap};
 
-use crate::console::{ConsoleController};
+use crate::console::{BackendType, ConsoleController};
 use crate::vhu_console::VhostUserConsoleBackend;
 
 pub(crate) type Result<T> = std::result::Result<T, Error>;
@@ -26,186 +26,288 @@ pub(crate) type Result<T> = std::result::Result<T, Error>;
 pub(crate) enum Error {
     #[error("Invalid socket count: {0}")]
     SocketCountInvalid(usize),
-    #[error("Failed to join threads")]
-    FailedJoiningThreads,
-    #[error("Could not create console controller: {0}")]
-    CouldNotCreateConsoleController(crate::console::Error),
     #[error("Could not create console backend: {0}")]
     CouldNotCreateBackend(crate::vhu_console::Error),
     #[error("Could not create daemon: {0}")]
     CouldNotCreateDaemon(vhost_user_backend::Error),
-}
-
-#[derive(Parser, Debug)]
-#[clap(author, version, about, long_about = None)]
-struct ConsoleArgs {
-    /// Location of vhost-user Unix domain socket. This is suffixed by 0,1,2..socket_count-1.
-    #[clap(short, long)]
-    socket_path: String,
-
-    /// A console device name to be used for reading (ex. vconsole, console0, console1, ... etc.)
-    #[clap(short = 'i', long)]
-    console_path: String,
-
-    /// Number of guests (sockets) to connect to.
-    #[clap(short = 'c', long, default_value_t = 1)]
-    socket_count: u32,
+    #[error("Fatal error: {0}")]
+    ServeFailed(vhost_user_backend::Error),
+    #[error("Thread `{0}` panicked")]
+    ThreadPanic(String, Box<dyn Any + Send>),
+    #[error("Error using multiple sockets with Nested backend")]
+    WrongBackendSocket,
 }
 
 #[derive(PartialEq, Debug)]
-struct ConsoleConfiguration {
-    socket_path: String,
-    socket_count: u32,
-    console_path: String,
+pub struct VuConsoleConfig {
+    pub(crate) socket_path: PathBuf,
+    pub(crate) backend: BackendType,
+    pub(crate) tcp_port: String,
+    pub(crate) socket_count: u32,
 }
 
-impl TryFrom<ConsoleArgs> for ConsoleConfiguration {
-    type Error = Error;
+impl VuConsoleConfig {
+    pub fn generate_socket_paths(&self) -> Vec<PathBuf> {
+        let socket_file_name = self
+            .socket_path
+            .file_name()
+            .expect("socket_path has no filename.");
+        let socket_file_parent = self
+            .socket_path
+            .parent()
+            .expect("socket_path has no parent directory.");
+
+        let make_socket_path = |i: u32| -> PathBuf {
+            let mut file_name = socket_file_name.to_os_string();
+            file_name.push(std::ffi::OsStr::new(&i.to_string()));
+            socket_file_parent.join(&file_name)
+        };
 
-    fn try_from(args: ConsoleArgs) -> Result<Self> {
+        (0..self.socket_count).map(make_socket_path).collect()
+    }
 
-        if args.socket_count == 0 {
-            return Err(Error::SocketCountInvalid(0));
-        }
+    pub fn generate_tcp_addrs(&self) -> Vec<String> {
+        let tcp_port_base = self.tcp_port.clone();
 
-		let console_path = args.console_path.trim().to_string();
+        let make_tcp_port = |i: u32| -> String {
+            let port_num: u32 = tcp_port_base.clone().parse().unwrap();
+            "127.0.0.1:".to_owned() + &(port_num + i).to_string()
+        };
 
-        Ok(ConsoleConfiguration {
-            socket_path: args.socket_path,
-			socket_count: args.socket_count,
-            console_path,
-        })
+        (0..self.socket_count).map(make_tcp_port).collect()
     }
 }
 
-fn start_backend(args: ConsoleArgs) -> Result<()> {
-
-	println!("start_backend function!\n");
-
-    let config = ConsoleConfiguration::try_from(args).unwrap();
-    let mut handles = Vec::new();
+// This is the public API through which an external program starts the
+/// vhost-device-console backend server.
+pub(crate) fn start_backend_server(
+    socket: PathBuf,
+    tcp_addr: String,
+    backend: BackendType,
+) -> Result<()> {
+    loop {
+        let controller = ConsoleController::new(backend);
+        let arc_controller = Arc::new(RwLock::new(controller));
+        let vu_console_backend = Arc::new(RwLock::new(
+            VhostUserConsoleBackend::new(arc_controller).map_err(Error::CouldNotCreateBackend)?,
+        ));
+
+        let mut daemon = VhostUserDaemon::new(
+            String::from("vhost-device-console-backend"),
+            vu_console_backend.clone(),
+            GuestMemoryAtomic::new(GuestMemoryMmap::new()),
+        )
+        .map_err(Error::CouldNotCreateDaemon)?;
+
+        let vring_workers = daemon.get_epoll_handlers();
+        vu_console_backend
+            .read()
+            .unwrap()
+            .set_vring_worker(&vring_workers[0]);
+
+        // Start the corresponding console thread
+        let read_handle = if backend == BackendType::Nested {
+            VhostUserConsoleBackend::start_console_thread(&vu_console_backend)
+        } else {
+            VhostUserConsoleBackend::start_tcp_console_thread(&vu_console_backend, tcp_addr.clone())
+        };
 
-    for _ in 0..config.socket_count {
-        let socket = config.socket_path.to_owned();
-        let console_path = config.console_path.to_owned();
+        daemon.serve(&socket).map_err(Error::ServeFailed)?;
 
-        let handle: JoinHandle<Result<()>> = spawn(move || loop {
-            // A separate thread is spawned for each socket and console connect to a separate guest.
-            // These are run in an infinite loop to not require the daemon to be restarted once a
-            // guest exits.
-            //
-            // There isn't much value in complicating code here to return an error from the
-            // threads, and so the code uses unwrap() instead. The panic on a thread won't cause
-            // trouble to other threads/guests or the main() function and should be safe for the
-            // daemon.
+        // Kill console input thread
+        vu_console_backend.read().unwrap().kill_console_thread();
 
-            let controller =
-                ConsoleController::new(console_path.clone()).map_err(Error::CouldNotCreateConsoleController)?;
-			let arc_controller =  Arc::new(RwLock::new(controller));
-            let vu_console_backend = Arc::new(RwLock::new(
-                VhostUserConsoleBackend::new(arc_controller).map_err(Error::CouldNotCreateBackend)?,
-            ));
+        // Wait for read thread to exit
+        match read_handle.join() {
+            Ok(_) => info!("The read thread returned successfully"),
+            Err(e) => warn!("The read thread returned the error: {:?}", e),
+        }
+    }
+}
 
-            let mut daemon = VhostUserDaemon::new(
-                String::from("vhost-device-console-backend"),
-                vu_console_backend.clone(),
-                GuestMemoryAtomic::new(GuestMemoryMmap::new()),
-            )
-            .map_err(Error::CouldNotCreateDaemon)?;
-
-            /* Start the read thread -- need to handle it after termination */
-            let vring_workers = daemon.get_epoll_handlers();
-            vu_console_backend.read()
-                          .unwrap()
-                          .set_vring_worker(&vring_workers[0]);
-			VhostUserConsoleBackend::start_console_thread(&vu_console_backend);
-
-            let listener = Listener::new(socket.clone(), true).unwrap();
-            daemon.start(listener).unwrap();
-
-            match daemon.wait() {
-                Ok(()) => {
-                    info!("Stopping cleanly.");
-                }
-                Err(vhost_user_backend::Error::HandleRequest(
-                    vhost_user::Error::PartialMessage | vhost_user::Error::Disconnected,
-                )) => {
-                    info!("vhost-user connection closed with partial message. If the VM is shutting down, this is expected behavior; otherwise, it might be a bug.");
-                }
-                Err(e) => {
-                    warn!("Error running daemon: {:?}", e);
-                }
-            }
-
-            // No matter the result, we need to shut down the worker thread.
-            vu_console_backend.read().unwrap().exit_event.write(1).unwrap();
-        });
-
-        handles.push(handle);
+pub fn start_backend(config: VuConsoleConfig) -> Result<()> {
+    let mut handles = HashMap::new();
+    let (senders, receiver) = std::sync::mpsc::channel();
+    let tcp_addrs = config.generate_tcp_addrs();
+    let backend = config.backend;
+
+    for (thread_id, (socket, tcp_addr)) in config
+        .generate_socket_paths()
+        .into_iter()
+        .zip(tcp_addrs.iter())
+        .enumerate()
+    {
+        let tcp_addr = tcp_addr.clone();
+        info!("thread_id: {}, socket: {:?}", thread_id, socket);
+
+        let name = format!("vhu-console-{}", tcp_addr);
+        let sender = senders.clone();
+        let handle = Builder::new()
+            .name(name.clone())
+            .spawn(move || {
+                let result = std::panic::catch_unwind(move || {
+                    start_backend_server(socket, tcp_addr.to_string(), backend)
+                });
+
+                // Notify the main thread that we are done.
+                sender.send(thread_id).unwrap();
+
+                result.map_err(|e| Error::ThreadPanic(name, e))?
+            })
+            .unwrap();
+        handles.insert(thread_id, handle);
     }
 
-    for handle in handles {
-        handle.join().map_err(|_| Error::FailedJoiningThreads)??;
+    while !handles.is_empty() {
+        let thread_id = receiver.recv().unwrap();
+        handles
+            .remove(&thread_id)
+            .unwrap()
+            .join()
+            .map_err(std::panic::resume_unwind)
+            .unwrap()?;
     }
 
     Ok(())
 }
 
-pub(crate) fn console_init() {
-    env_logger::init();
-	println!("Console_init function!");
-    if let Err(e) = start_backend(ConsoleArgs::parse()) {
-        error!("{e}");
-        exit(1);
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::ConsoleArgs;
+    use assert_matches::assert_matches;
 
     #[test]
-    fn test_console_configuration_try_from_valid_args() {
+    fn test_console_valid_configuration_nested() {
         let args = ConsoleArgs {
-            socket_path: String::from("/path/to/socket"),
-            console_path: String::from("vconsole"),
-            socket_count: 3,
+            socket_path: String::from("/tmp/vhost.sock").into(),
+            backend: BackendType::Nested,
+            tcp_port: String::from("12345"),
+            socket_count: 1,
         };
 
-        let result = ConsoleConfiguration::try_from(args);
+        assert!(VuConsoleConfig::try_from(args).is_ok());
+    }
 
-        assert!(result.is_ok());
+    #[test]
+    fn test_console_invalid_configuration_nested_1() {
+        let args = ConsoleArgs {
+            socket_path: String::from("/tmp/vhost.sock").into(),
+            backend: BackendType::Nested,
+            tcp_port: String::from("12345"),
+            socket_count: 0,
+        };
 
-        let config = result.unwrap();
-        assert_eq!(config.socket_path, "/path/to/socket");
-        assert_eq!(config.console_path, "vconsole");
-        assert_eq!(config.socket_count, 3);
+        assert_matches!(
+            VuConsoleConfig::try_from(args),
+            Err(Error::SocketCountInvalid(0))
+        );
     }
 
     #[test]
-    fn test_console_configuration_try_from_invalid_args() {
-        // Test with socket_count = 0
-        let args_invalid_count = ConsoleArgs {
-            socket_path: String::from("/path/to/socket"),
-            console_path: String::from("vconsole"),
-            socket_count: 0,
+    fn test_console_invalid_configuration_nested_2() {
+        let args = ConsoleArgs {
+            socket_path: String::from("/tmp/vhost.sock").into(),
+            backend: BackendType::Nested,
+            tcp_port: String::from("12345"),
+            socket_count: 2,
+        };
+
+        assert_matches!(
+            VuConsoleConfig::try_from(args),
+            Err(Error::WrongBackendSocket)
+        );
+    }
+
+    #[test]
+    fn test_console_valid_configuration_network_1() {
+        let args = ConsoleArgs {
+            socket_path: String::from("/tmp/vhost.sock").into(),
+            backend: BackendType::Network,
+            tcp_port: String::from("12345"),
+            socket_count: 1,
         };
 
-        let result_invalid_count = ConsoleConfiguration::try_from(args_invalid_count);
-        assert!(result_invalid_count.is_err());
+        assert!(VuConsoleConfig::try_from(args).is_ok());
     }
 
     #[test]
-    fn test_start_backend_success() {
-        // Test start_backend with valid arguments
+    fn test_console_valid_configuration_network_2() {
         let args = ConsoleArgs {
-            socket_path: String::from("/path/to/socket"),
-            console_path: String::from("vconsole"),
+            socket_path: String::from("/tmp/vhost.sock").into(),
+            backend: BackendType::Network,
+            tcp_port: String::from("12345"),
             socket_count: 2,
         };
 
-        let result = start_backend(args);
+        assert!(VuConsoleConfig::try_from(args).is_ok());
+    }
+
+    fn test_backend_start_and_stop(args: ConsoleArgs) {
+        let config = VuConsoleConfig::try_from(args).expect("Wrong config");
+
+        let tcp_addrs = config.generate_tcp_addrs();
+        let backend = config.backend;
+
+        for (_socket, tcp_addr) in config
+            .generate_socket_paths()
+            .into_iter()
+            .zip(tcp_addrs.iter())
+        {
+            let controller = ConsoleController::new(backend);
+            let arc_controller = Arc::new(RwLock::new(controller));
+            let vu_console_backend = Arc::new(RwLock::new(
+                VhostUserConsoleBackend::new(arc_controller)
+                    .map_err(Error::CouldNotCreateBackend)
+                    .expect("Fail create vhuconsole backend"),
+            ));
+
+            let mut _daemon = VhostUserDaemon::new(
+                String::from("vhost-device-console-backend"),
+                vu_console_backend.clone(),
+                GuestMemoryAtomic::new(GuestMemoryMmap::new()),
+            )
+            .map_err(Error::CouldNotCreateDaemon)
+            .expect("Failed create daemon");
+
+            // Start the corresponinding console thread
+            let read_handle = if backend == BackendType::Nested {
+                VhostUserConsoleBackend::start_console_thread(&vu_console_backend)
+            } else {
+                VhostUserConsoleBackend::start_tcp_console_thread(
+                    &vu_console_backend,
+                    tcp_addr.clone(),
+                )
+            };
+
+            // Kill console input thread
+            vu_console_backend.read().unwrap().kill_console_thread();
+
+            // Wait for read thread to exit
+            assert_matches!(read_handle.join(), Ok(_));
+        }
+    }
+    #[test]
+    fn test_start_net_backend_success() {
+        let args = ConsoleArgs {
+            socket_path: String::from("/tmp/vhost.sock").into(),
+            backend: BackendType::Network,
+            tcp_port: String::from("12345"),
+            socket_count: 1,
+        };
+
+        test_backend_start_and_stop(args);
+    }
+
+    #[test]
+    fn test_start_nested_backend_success() {
+        let args = ConsoleArgs {
+            socket_path: String::from("/tmp/vhost.sock").into(),
+            backend: BackendType::Nested,
+            tcp_port: String::from("12345"),
+            socket_count: 1,
+        };
 
-        assert!(result.is_ok());
+        test_backend_start_and_stop(args);
     }
 }
diff --git a/meta-egvirt/recipes-extended/vhost-device-console/vhost-device-console-0.1.0/src/console.rs b/meta-egvirt/recipes-extended/vhost-device-console/vhost-device-console-0.1.0/src/console.rs
index 2e2972b2..079fe0cb 100644..100755
--- a/meta-egvirt/recipes-extended/vhost-device-console/vhost-device-console-0.1.0/src/console.rs
+++ b/meta-egvirt/recipes-extended/vhost-device-console/vhost-device-console-0.1.0/src/console.rs
@@ -1,86 +1,44 @@
-// CAN backend device
+// Console backend device
 //
-// Copyright 2023 VIRTUAL OPEN SYSTEMS SAS. All Rights Reserved.
+// Copyright 2023-2024 VIRTUAL OPEN SYSTEMS SAS. All Rights Reserved.
 //          Timos Ampelikiotis <t.ampelikiotis@virtualopensystems.com>
 //
 // SPDX-License-Identifier: Apache-2.0 or BSD-3-Clause
 
-use log::{warn, error};
-use std::sync::{Arc, RwLock};
+use crate::virtio_console::VirtioConsoleConfig;
+use clap::ValueEnum;
+use log::trace;
 
-use thiserror::Error as ThisError;
-use vm_memory::{ByteValued, Le16};
-
-use crate::vhu_console::{VirtioConsoleConfig, VirtioConsoleControl};
-
-type Result<T> = std::result::Result<T, Error>;
-
-#[derive(Copy, Clone, Debug, PartialEq, ThisError)]
-pub(crate) enum Error {
-    #[error("Console not enabled yet")]
-    ConsoleNotEnabled,
+#[derive(ValueEnum, Clone, Copy, Default, Debug, Eq, PartialEq)]
+pub enum BackendType {
+    #[default]
+    Nested,
+    Network,
 }
 
 #[derive(Debug)]
 pub(crate) struct ConsoleController {
     config: VirtioConsoleConfig,
-    pub console_name: String,
+    pub backend: BackendType,
+    pub exit: bool,
 }
 
 impl ConsoleController {
-    // Creates a new controller corresponding to `device`.
-    pub(crate) fn new(console_name: String) -> Result<ConsoleController> {
-
-        let console_name = console_name.to_owned();
-		println!("console_name: {:?}", console_name);
-
-        Ok(ConsoleController {
+    pub(crate) fn new(backend: BackendType) -> ConsoleController {
+        ConsoleController {
             config: VirtioConsoleConfig {
-						cols: 20.into(),
-						rows: 20.into(),
-						max_nr_ports: 1.into(),
-						emerg_wr: 64.into(),
-					},
-            console_name,
-        })
+                cols: 20.into(),
+                rows: 20.into(),
+                max_nr_ports: 1.into(),
+                emerg_wr: 64.into(),
+            },
+            backend,
+            exit: false,
+        }
     }
 
     pub(crate) fn config(&self) -> &VirtioConsoleConfig {
-		log::trace!("Get config\n");
+        trace!("Get config\n");
         &self.config
     }
-
-    pub(crate) fn operation(&self, tx_request: VirtioConsoleControl) -> Result<()> {
-		log::trace!("Console operation\n");
-		Ok(())
-    }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_new_console_controller() {
-        let console_name = String::from("test_console");
-        let controller = ConsoleController::new(console_name.clone());
-
-        assert!(controller.is_ok());
-
-        let controller = controller.unwrap();
-        assert_eq!(controller.console_name, "test_console");
-    }
-
-    #[test]
-    fn test_console_controller_config() {
-        let console_name = String::from("test_console");
-        let controller = ConsoleController::new(console_name).unwrap();
-
-        let config = controller.config();
-        assert_eq!(config.cols, 20);
-        assert_eq!(config.rows, 20);
-        assert_eq!(config.max_nr_ports, 1);
-        assert_eq!(config.emerg_wr, 64);
-    }
-}
-
diff --git a/meta-egvirt/recipes-extended/vhost-device-console/vhost-device-console-0.1.0/src/main.rs b/meta-egvirt/recipes-extended/vhost-device-console/vhost-device-console-0.1.0/src/main.rs
index cceafb4f..216a01e6 100644..100755
--- a/meta-egvirt/recipes-extended/vhost-device-console/vhost-device-console-0.1.0/src/main.rs
+++ b/meta-egvirt/recipes-extended/vhost-device-console/vhost-device-console-0.1.0/src/main.rs
@@ -1,18 +1,69 @@
 // VIRTIO CONSOLE Emulation via vhost-user
 //
-// Copyright 2023 VIRTUAL OPEN SYSTEMS SAS. All Rights Reserved.
+// Copyright 2023-2024 VIRTUAL OPEN SYSTEMS SAS. All Rights Reserved.
 //          Timos Ampelikiotis <t.ampelikiotis@virtualopensystems.com>
 //
 // SPDX-License-Identifier: Apache-2.0 or BSD-3-Clause
 
-#[cfg(target_env = "gnu")]
 mod backend;
-#[cfg(target_env = "gnu")]
 mod console;
-#[cfg(target_env = "gnu")]
 mod vhu_console;
+mod virtio_console;
+use crate::console::BackendType;
+use clap::Parser;
+use log::error;
+use std::path::PathBuf;
+use std::process::exit;
+
+pub(crate) type Result<T> = std::result::Result<T, Error>;
+use crate::backend::{start_backend, Error, VuConsoleConfig};
+
+#[derive(Parser, Debug)]
+#[clap(author, version, about, long_about = None)]
+struct ConsoleArgs {
+    /// Location of vhost-user Unix domain socket. This is suffixed by 0,1,2..socket_count-1.
+    #[clap(short = 's', long, value_name = "SOCKET")]
+    socket_path: PathBuf,
+
+    /// Number of guests (sockets) to connect to.
+    #[clap(short = 'c', long, default_value_t = 1)]
+    socket_count: u32,
+
+    /// Console backend (Network, Nested) to be used.
+    #[clap(short = 'b', long, value_enum, default_value = "nested")]
+    backend: BackendType,
+
+    /// Initial tcp port to be used with "network" backend. If socket_count is N then
+    /// the following tcp ports will be created: tcp_port, tcp_port + 1, ..., tcp_port + (N - 1).
+    #[clap(short = 'p', long, value_name = "PORT", default_value = "12345")]
+    tcp_port: String,
+}
+
+impl TryFrom<ConsoleArgs> for VuConsoleConfig {
+    type Error = Error;
+
+    fn try_from(args: ConsoleArgs) -> Result<Self> {
+        if args.socket_count == 0 {
+            return Err(Error::SocketCountInvalid(0));
+        }
+
+        if (args.backend == BackendType::Nested) && (args.socket_count != 1) {
+            return Err(Error::WrongBackendSocket);
+        }
+
+        Ok(VuConsoleConfig {
+            socket_path: args.socket_path,
+            backend: args.backend,
+            tcp_port: args.tcp_port,
+            socket_count: args.socket_count,
+        })
+    }
+}
 
-#[cfg(target_env = "gnu")]
 fn main() {
-    backend::console_init()
+    env_logger::init();
+    if let Err(e) = VuConsoleConfig::try_from(ConsoleArgs::parse()).and_then(start_backend) {
+        error!("{e}");
+        exit(1);
+    }
 }
diff --git a/meta-egvirt/recipes-extended/vhost-device-console/vhost-device-console-0.1.0/src/vhu_console.rs b/meta-egvirt/recipes-extended/vhost-device-console/vhost-device-console-0.1.0/src/vhu_console.rs
index ebddb00d..d5cca74a 100644..100755
--- a/meta-egvirt/recipes-extended/vhost-device-console/vhost-device-console-0.1.0/src/vhu_console.rs
+++ b/meta-egvirt/recipes-extended/vhost-device-console/vhost-device-console-0.1.0/src/vhu_console.rs
@@ -1,22 +1,31 @@
 // vhost device console
 //
-// Copyright 2023 VIRTUAL OPEN SYSTEMS SAS. All Rights Reserved.
+// Copyright 2023-2024 VIRTUAL OPEN SYSTEMS SAS. All Rights Reserved.
 //          Timos Ampelikiotis <t.ampelikiotis@virtualopensystems.com>
 //
 // SPDX-License-Identifier: Apache-2.0 or BSD-3-Clause
 
-use log::{warn, error};
-use std::mem::size_of;
+use crate::console::{BackendType, ConsoleController};
+use crate::virtio_console::{
+    VirtioConsoleControl, VIRTIO_CONSOLE_CONSOLE_PORT, VIRTIO_CONSOLE_DEVICE_READY,
+    VIRTIO_CONSOLE_F_MULTIPORT, VIRTIO_CONSOLE_PORT_ADD, VIRTIO_CONSOLE_PORT_NAME,
+    VIRTIO_CONSOLE_PORT_OPEN, VIRTIO_CONSOLE_PORT_READY,
+};
+use console::Key;
+use crossterm::terminal::{disable_raw_mode, enable_raw_mode};
+use log::{error, trace};
+use nix::sys::select::{select, FdSet};
+use std::os::fd::AsRawFd;
 use std::slice::from_raw_parts;
 use std::sync::{Arc, RwLock};
+use std::thread::JoinHandle;
 use std::{
     convert,
     io::{self, Result as IoResult},
 };
-use std::io::{Write};
-use std::os::fd::AsRawFd;
 use thiserror::Error as ThisError;
 use vhost::vhost_user::message::{VhostUserProtocolFeatures, VhostUserVirtioFeatures};
+use vhost_user_backend::VringEpollHandler;
 use vhost_user_backend::{VhostUserBackendMut, VringRwLock, VringT};
 use virtio_bindings::bindings::virtio_config::{VIRTIO_F_NOTIFY_ON_EMPTY, VIRTIO_F_VERSION_1};
 use virtio_bindings::bindings::virtio_ring::{
@@ -24,76 +33,61 @@ use virtio_bindings::bindings::virtio_ring::{
 };
 use virtio_queue::{DescriptorChain, QueueOwnedT};
 use vm_memory::{
-    ByteValued, Bytes, GuestAddressSpace, GuestMemoryAtomic, GuestMemoryLoadGuard,
-    GuestMemoryMmap, Le16, Le32,
+    ByteValued, GuestAddressSpace, GuestMemoryAtomic, GuestMemoryLoadGuard, GuestMemoryMmap,
 };
 use vmm_sys_util::epoll::EventSet;
 use vmm_sys_util::eventfd::{EventFd, EFD_NONBLOCK};
-use vhost_user_backend::VringEpollHandler;
-use crate::console::{ConsoleController};
 
-use std::thread::{JoinHandle, spawn};
-use queues::{Queue, IsQueue};
-use std::thread;
-use std::time::Duration;
 use console::Term;
-
-/// Feature bit numbers
-pub const VIRTIO_CONSOLE_F_SIZE: u16 = 0;
-pub const VIRTIO_CONSOLE_F_MULTIPORT: u16 = 1;
-pub const VIRTIO_CONSOLE_F_EMERG_WRITE: u16 = 2;
+use queues::{IsQueue, Queue};
+use std::io::Read;
+use std::io::Write;
+use std::net::TcpListener;
+use std::thread::spawn;
 
 /// Virtio configuration
 const QUEUE_SIZE: usize = 128;
 const NUM_QUEUES: usize = 4;
 
-/// Queues
+/// Queue events
 const RX_QUEUE: u16 = 0;
 const TX_QUEUE: u16 = 1;
 const CTRL_RX_QUEUE: u16 = 2;
 const CTRL_TX_QUEUE: u16 = 3;
-const BACKEND_EFD: u16 = (NUM_QUEUES + 1) as u16;
-const BACKEND_RX_EFD: u16 = (NUM_QUEUES + 2) as u16;
 
+/// The two following events are used to help the vhu_console
+/// backend trigger events to itself. For example:
+/// a) BACKEND_RX_EFD is being triggered when the backend
+///    has new data to send to the RX queue.
+/// b) BACKEND_CTRL_RX_EFD event is used when the backend
+///    needs to write to the RX control queue.
+const BACKEND_RX_EFD: u16 = (NUM_QUEUES + 1) as u16;
+const BACKEND_CTRL_RX_EFD: u16 = (NUM_QUEUES + 2) as u16;
 
-/// Console virtio control messages
-const VIRTIO_CONSOLE_DEVICE_READY: u16 = 0 ;
-const VIRTIO_CONSOLE_PORT_ADD: u16 = 1;
-const VIRTIO_CONSOLE_PORT_REMOVE: u16 = 2; 
-const VIRTIO_CONSOLE_PORT_READY: u16 = 3;
-const VIRTIO_CONSOLE_CONSOLE_PORT: u16 = 4; 
-const VIRTIO_CONSOLE_RESIZE: u16 = 5;
-const VIRTIO_CONSOLE_PORT_OPEN: u16 = 6; 
-const VIRTIO_CONSOLE_PORT_NAME: u16 = 7;
+/// Port name - Need to be updated when MULTIPORT feature
+///             is supported for more than one devices.
+const PORT_NAME: &[u8] = b"org.test.foo!";
 
 type Result<T> = std::result::Result<T, Error>;
 
 #[derive(Copy, Clone, Debug, PartialEq, ThisError)]
 pub(crate) enum Error {
-    #[error("Failed to handle event, didn't match EPOLLIN")]
-    HandleEventNotEpollIn,
     #[error("Failed to handle unknown event")]
     HandleEventUnknown,
-    #[error("Received unexpected write only descriptor at index {0}")]
-    UnexpectedWriteOnlyDescriptor(usize),
-    #[error("Received unexpected readable descriptor at index {0}")]
-    UnexpectedReadableDescriptor(usize),
-    #[error("Invalid descriptor count {0}")]
-    UnexpectedDescriptorCount(usize),
-    #[error("Invalid descriptor size, expected: {0}, found: {1}")]
-    UnexpectedDescriptorSize(usize, u32),
-	#[error("Descriptor not found")]
-	DescriptorNotFound,
-	#[error("Failed to send notification")]
-	NotificationFailed,
+    #[error("Descriptor not found")]
+    DescriptorNotFound,
+    #[error("Failed to send notification")]
+    NotificationFailed,
     #[error("Descriptor read failed")]
     DescriptorReadFailed,
     #[error("Descriptor write failed")]
     DescriptorWriteFailed,
+    #[error("Add used element in vring {0} failed")]
+    AddUsedElemFailed(u16),
     #[error("Failed to create new EventFd")]
     EventFdFailed,
-    #[error("Failed to remove rx queue")]
-    EmptyQueue,
+    #[error("Failed to add control message in the internal queue")]
+    RxCtrlQueueAddFailed,
 }
 
 impl convert::From<Error> for io::Error {
@@ -102,53 +96,19 @@ impl convert::From<Error> for io::Error {
     }
 }
 
-/// Virtio Console Config
-#[derive(Copy, Clone, Debug, Default, PartialEq)]
-#[repr(C)]
-pub(crate) struct VirtioConsoleConfig {
-    pub cols: Le16,
-    pub rows: Le16,
-    pub max_nr_ports: Le32,
-    pub emerg_wr: Le32,
-}
-
-// SAFETY: The layout of the structure is fixed and can be initialized by
-// reading its content from byte array.
-unsafe impl ByteValued for VirtioConsoleConfig {}
-
-#[derive(Copy, Clone, Debug, Default, PartialEq)]
-#[repr(C)]
-pub(crate) struct VirtioConsoleControl {
-    pub id: Le32,
-    pub event: Le16,
-    pub value: Le16,
-}
-
-use std::io::Cursor;
-
-impl VirtioConsoleControl {
-    fn to_le_bytes(&self) -> Vec<u8> {
-        let mut buffer = Vec::new();
-
-		buffer.extend_from_slice(&self.id.to_native().to_le_bytes());
-		buffer.extend_from_slice(&self.event.to_native().to_le_bytes());
-		buffer.extend_from_slice(&self.value.to_native().to_le_bytes());
-        buffer
-    }
-}
-
 // SAFETY: The layout of the structure is fixed and can be initialized by
 // reading its content from byte array.
 unsafe impl ByteValued for VirtioConsoleControl {}
 
 pub(crate) struct VhostUserConsoleBackend {
     controller: Arc<RwLock<ConsoleController>>,
-	acked_features: u64,
+    acked_features: u64,
     event_idx: bool,
-	rx_fifo: Queue<VirtioConsoleControl>,
+    rx_ctrl_fifo: Queue<VirtioConsoleControl>,
+    rx_data_fifo: Queue<String>,
     pub(crate) ready: bool,
     pub(crate) ready_to_write: bool,
-	pub(crate) output_buffer: String,
+    pub(crate) output_queue: Queue<String>,
     pub(crate) rx_event: EventFd,
     pub(crate) rx_ctrl_event: EventFd,
     pub(crate) exit_event: EventFd,
@@ -160,13 +120,14 @@ type ConsoleDescriptorChain = DescriptorChain<GuestMemoryLoadGuard<GuestMemoryMm
 impl VhostUserConsoleBackend {
     pub(crate) fn new(controller: Arc<RwLock<ConsoleController>>) -> Result<Self> {
         Ok(VhostUserConsoleBackend {
-            controller: controller,
+            controller,
             event_idx: false,
-			rx_fifo: Queue::new(),
-			acked_features: 0x0,
-			ready: false,
-			ready_to_write: false,
-			output_buffer: String::new(),
+            rx_ctrl_fifo: Queue::new(),
+            rx_data_fifo: Queue::new(),
+            acked_features: 0x0,
+            ready: false,
+            ready_to_write: false,
+            output_queue: Queue::new(),
             rx_event: EventFd::new(EFD_NONBLOCK).map_err(|_| Error::EventFdFailed)?,
             rx_ctrl_event: EventFd::new(EFD_NONBLOCK).map_err(|_| Error::EventFdFailed)?,
             exit_event: EventFd::new(EFD_NONBLOCK).map_err(|_| Error::EventFdFailed)?,
@@ -174,292 +135,229 @@ impl VhostUserConsoleBackend {
         })
     }
 
-	fn check_features (&self, features: u16) -> bool {
-		(self.acked_features & (1 << features)) != 0
-	}
-
-    fn print_console_frame (&self, control_msg: VirtioConsoleControl) {
-        println!("id 0x{:x}", control_msg.id.to_native());
-        println!("event 0x{:x}", control_msg.event.to_native());
-        println!("value 0x{:x}", control_msg.value.to_native());
+    fn print_console_frame(&self, control_msg: VirtioConsoleControl) {
+        trace!("id 0x{:x}", control_msg.id.to_native());
+        trace!("event 0x{:x}", control_msg.event.to_native());
+        trace!("value 0x{:x}", control_msg.value.to_native());
     }
 
     fn process_rx_requests(
-		&mut self,
-		requests: Vec<ConsoleDescriptorChain>,
-		vring: &VringRwLock
-	) -> Result<bool> {
-		log::trace!("process_rx_requests");
-
+        &mut self,
+        requests: Vec<ConsoleDescriptorChain>,
+        vring: &VringRwLock,
+    ) -> Result<bool> {
         if requests.is_empty() {
-			log::trace!("requests.is_empty");
-			vring.signal_used_queue();
             return Ok(true);
         }
 
-		log::trace!("requests.len: {:?}", requests.len());
-        let desc_chain = &requests[0];
-        let descriptors: Vec<_> = desc_chain.clone().collect();
-
-		log::trace!("descriptors.len(): {:?}", descriptors.len());
-        if descriptors.len() != 1 {
-			log::trace!("Error::UnexpectedDescriptorCount");
-            return Err(Error::UnexpectedDescriptorCount(descriptors.len()));
-        }
-
-        let desc_request = descriptors[0];
-        if !desc_request.is_write_only() {
-			log::trace!("!desc_request.is_write_only()");
-            return Err(Error::UnexpectedReadableDescriptor(1));
-        }
-
-        // TODO: if buffer is more than the the desc_request length,
-        //       write the remaining in the next chain.
-		log::trace!("desc_request.len(): {}", desc_request.len());
-		let response = self.output_buffer.clone();
+        for desc_chain in requests {
+            let atomic_mem = self.mem.as_ref().unwrap().memory();
+            let mut writer = desc_chain
+                .clone()
+                .writer(&atomic_mem)
+                .map_err(|_| Error::DescriptorWriteFailed)?;
+
+            let response: String = match self.rx_data_fifo.remove() {
+                Ok(item) => item,
+                _ => {
+                    return Ok(false);
+                }
+            };
 
-        desc_chain
-            .memory()
-            .write_slice(response.as_bytes(), desc_request.addr())
-            .map_err(|_| Error::DescriptorWriteFailed)?;
+            for b in response.bytes() {
+                writer
+                    .write_obj::<u8>(b)
+                    .map_err(|_| Error::DescriptorWriteFailed)?;
+            }
 
-        if vring.add_used(desc_chain.head_index(), response.as_bytes().len() as u32).is_err() {
-            warn!("Couldn't return used descriptors to the ring");
+            vring
+                .add_used(desc_chain.head_index(), writer.bytes_written() as u32)
+                .map_err(|_| Error::AddUsedElemFailed(RX_QUEUE))?;
         }
 
         Ok(true)
     }
 
     fn process_tx_requests(
-		&self,
-		requests: Vec<ConsoleDescriptorChain>,
-		vring: &VringRwLock
-	) -> Result<bool> {
-		log::trace!("process_tx_requests");
-
+        &mut self,
+        requests: Vec<ConsoleDescriptorChain>,
+        vring: &VringRwLock,
+    ) -> Result<bool> {
         if requests.is_empty() {
-			log::trace!("requests.is_empty");
             return Ok(true);
         }
 
-		log::trace!("requests.len: {:?}", requests.len());
         for desc_chain in requests {
-            let descriptors: Vec<_> = desc_chain.clone().collect();
-
-			log::trace!("descriptors.len(): {:?}", descriptors.len());
-            if descriptors.len() != 1 {
-				log::trace!("Error::UnexpectedDescriptorCount");
-                return Err(Error::UnexpectedDescriptorCount(descriptors.len()));
-            }
-
-            let desc_request = descriptors[0];
-            if desc_request.is_write_only() {
-				log::trace!("Error::UnexpectedReadOnlyDescriptor");
-                return Err(Error::UnexpectedWriteOnlyDescriptor(0));
-			}
-
-			log::trace!("desc_request.len(): {}", desc_request.len());
-			let desc_len = desc_request.len();
-
-			let mut buffer = [0 as u8; 4096];
-
-            let request = desc_chain
-                .memory()
-                .read_slice(&mut buffer, desc_request.addr())
+            let atomic_mem = self.mem.as_ref().unwrap().memory();
+            let mut reader = desc_chain
+                .clone()
+                .reader(&atomic_mem)
                 .map_err(|_| Error::DescriptorReadFailed)?;
 
-			let new_buff = &buffer[0..desc_len as usize];
-
-		    let my_string = String::from_utf8(new_buff.to_vec()).unwrap();
-		    log::trace!("{}", my_string);
-		    print!("{}", my_string);
-    	    io::stdout().flush().unwrap(); // Ensure the prompt is displayed.
+            let mut tx_data: Vec<u8> = Vec::new();
+            let data_len = reader.available_bytes();
+            for _i in 0..data_len {
+                let data_byte = reader
+                    .read_obj::<u8>()
+                    .map_err(|_| Error::DescriptorReadFailed)?;
+                tx_data.push(data_byte);
+            }
 
-            if vring.add_used(desc_chain.head_index(), desc_request.len()).is_err() {
-                log::trace!("Couldn't return used descriptors to the ring");
-                warn!("Couldn't return used descriptors to the ring");
+            let my_string = String::from_utf8(tx_data).unwrap();
+            if self.controller.read().unwrap().backend == BackendType::Nested {
+                print!("{}", my_string);
+                io::stdout().flush().unwrap();
+            } else {
+                self.output_queue
+                    .add(my_string)
+                    .expect("Failed to add element in the output queue");
+                //.map_err(|_| Error::RxCtrlQueueAddFailed)?;
             }
+
+            vring
+                .add_used(desc_chain.head_index(), reader.bytes_read() as u32)
+                .map_err(|_| Error::AddUsedElemFailed(TX_QUEUE))?;
         }
 
         Ok(true)
     }
 
     fn process_ctrl_rx_requests(
-		&mut self,
-		requests: Vec<ConsoleDescriptorChain>,
-		vring: &VringRwLock,
-	) -> Result<bool> {
-		log::trace!("process_ctrl_rx_requests");
+        &mut self,
+        requests: Vec<ConsoleDescriptorChain>,
+        vring: &VringRwLock,
+    ) -> Result<bool> {
+        let mut used_flag = false;
 
         if requests.is_empty() {
-			log::trace!("requests.is_empty()");
             return Ok(true);
         }
-		log::trace!("\trequests.len(): {}", requests.len());
 
         for desc_chain in requests {
+            let atomic_mem = self.mem.as_ref().unwrap().memory();
+            let mut writer = desc_chain
+                .clone()
+                .writer(&atomic_mem)
+                .map_err(|_| Error::DescriptorWriteFailed)?;
+
+            let ctrl_msg: VirtioConsoleControl = match self.rx_ctrl_fifo.remove() {
+                Ok(item) => {
+                    used_flag = true;
+                    item
+                }
+                _ => {
+                    return Ok(used_flag);
+                }
+            };
 
-			let descriptors: Vec<_> = desc_chain.clone().collect();
+            self.print_console_frame(ctrl_msg);
 
-			log::trace!("descriptors.len(): {:?}", descriptors.len());
-            if descriptors.len() < 1 {
-				warn!("Error::UnexpectedDescriptorCount");
-                return Err(Error::UnexpectedDescriptorCount(descriptors.len()));
-            }
+            let mut buffer: Vec<u8> = Vec::new();
+            buffer.extend_from_slice(&ctrl_msg.to_le_bytes());
 
-			log::trace!("self.rx_fifo.size(): {}", self.rx_fifo.size());
-			let ctrl_msg: VirtioConsoleControl = match self.rx_fifo.remove() {
-				Ok(item) => item,         
-				_ => {
-						log::trace!("No rx elements");
-						return Ok(false)
-					 },
-			};
-
-            let desc_request = descriptors[0];
-            if !desc_request.is_write_only() {
-				warn!("Error::UnexpectedWriteOnlyDescriptor");
-                return Err(Error::UnexpectedWriteOnlyDescriptor(0));
-            }
+            if ctrl_msg.event.to_native() == VIRTIO_CONSOLE_PORT_NAME {
+                buffer.extend_from_slice(PORT_NAME);
+            };
 
-            if (desc_request.len() as usize) < size_of::<VirtioConsoleControl>() {
-				log::trace!("UnexpectedDescriptorSize, len = {:?}", desc_request.len());
-                return Err(Error::UnexpectedDescriptorSize(
-                    size_of::<VirtioConsoleControl>(),
-                    desc_request.len(),
-                ));
-            }
+            writer
+                .write(buffer.as_slice())
+                .map_err(|_| Error::DescriptorWriteFailed)?;
 
-			log::trace!("desc_request.len(): {}", desc_request.len());
-			self.print_console_frame(ctrl_msg);
-
-		    let mut buffer: Vec<u8> = Vec::new();
-		    buffer.extend_from_slice(&ctrl_msg.to_le_bytes());
-		
-			if ctrl_msg.event.to_native() == VIRTIO_CONSOLE_PORT_NAME {
-		    	let string_bytes = "org.fedoraproject.console.foo!".as_bytes();
-		    	buffer.extend_from_slice(string_bytes);
-			};
-
-			desc_chain
-        	    .memory()
-        	    .write_slice(&buffer, desc_request.addr())
-        	    .map_err(|_| Error::DescriptorWriteFailed)?;
-
-            if vring.add_used(desc_chain.head_index(), desc_request.len()).is_err() {
-                log::trace!("Couldn't return used descriptors to the ring");
-                warn!("Couldn't return used descriptors to the ring");
-            }
+            vring
+                .add_used(desc_chain.head_index(), writer.bytes_written() as u32)
+                .map_err(|_| Error::AddUsedElemFailed(CTRL_RX_QUEUE))?;
         }
 
         Ok(true)
     }
 
-	fn handle_control_msg (
-		&mut self,
-		vring: &VringRwLock,
-		ctrl_msg: VirtioConsoleControl
-	) -> Result<()> {
-
-		let mut ctrl_msg_reply = VirtioConsoleControl {
-								id: 0.into(),
-								event: 0.into(),
-								value: 1.into(),
-							};
-		match ctrl_msg.event.to_native() {
-			VIRTIO_CONSOLE_DEVICE_READY => {
-				log::trace!("VIRTIO_CONSOLE_DEVICE_READY");
-				self.ready = true;
-				ctrl_msg_reply.event = VIRTIO_CONSOLE_PORT_ADD.into();
-				self.rx_fifo.add(ctrl_msg_reply);
-				self.process_ctrl_rx_queue(vring)?;
-			},
-		    VIRTIO_CONSOLE_PORT_READY => {
-				log::trace!("VIRTIO_CONSOLE_PORT_READY");
-				ctrl_msg_reply.event = VIRTIO_CONSOLE_CONSOLE_PORT.into();
-				self.rx_fifo.add(ctrl_msg_reply.clone());
-				self.process_ctrl_rx_queue(vring)?;
-
-				ctrl_msg_reply.event = VIRTIO_CONSOLE_PORT_NAME.into();
-				self.rx_fifo.add(ctrl_msg_reply.clone());
-				self.process_ctrl_rx_queue(vring)?;
-
-				ctrl_msg_reply.event = VIRTIO_CONSOLE_PORT_OPEN.into();
-				self.rx_fifo.add(ctrl_msg_reply.clone());
-				self.process_ctrl_rx_queue(vring)?;
-			},
-		    VIRTIO_CONSOLE_PORT_OPEN => {
-				log::trace!("VIRTIO_CONSOLE_PORT_OPEN");
-			},
-		    _ => {
-				log::trace!("Uknown control event");
-				return Err(Error::HandleEventUnknown);
-			}
-		};
-        Ok(())
-	}
+    fn handle_control_msg(&mut self, ctrl_msg: VirtioConsoleControl) -> Result<()> {
+        let mut ctrl_msg_reply = VirtioConsoleControl {
+            id: 0.into(),
+            event: 0.into(),
+            value: 1.into(),
+        };
+        match ctrl_msg.event.to_native() {
+            VIRTIO_CONSOLE_DEVICE_READY => {
+                trace!("VIRTIO_CONSOLE_DEVICE_READY");
 
-    fn process_ctrl_tx_requests(
-		&mut self,
-		requests: Vec<ConsoleDescriptorChain>,
-		vring: &VringRwLock,
-		rx_ctrl_vring: &VringRwLock
-	) -> Result<bool> {
-		log::trace!("process_ctrl_tx_requests");
+                if ctrl_msg.value != 1 {
+                    trace!("Guest failure in adding device");
+                    return Ok(());
+                }
 
-        if requests.is_empty() {
-			log::trace!("requests.is_empty()");
-            return Ok(true);
-        }
+                self.ready = true;
+                ctrl_msg_reply.event = VIRTIO_CONSOLE_PORT_ADD.into();
+                self.rx_ctrl_fifo
+                    .add(ctrl_msg_reply)
+                    .map_err(|_| Error::RxCtrlQueueAddFailed)?;
+            }
+            VIRTIO_CONSOLE_PORT_READY => {
+                trace!("VIRTIO_CONSOLE_PORT_READY");
 
-        for desc_chain in requests {
-			let descriptors: Vec<_> = desc_chain.clone().collect();
+                if ctrl_msg.value != 1 {
+                    trace!("Guest failure in adding port for device");
+                    return Ok(());
+                }
 
-            if descriptors.len() < 1 {
-				warn!("Error::UnexpectedDescriptorCount");
-                return Err(Error::UnexpectedDescriptorCount(descriptors.len()));
-            }
+                ctrl_msg_reply.event = VIRTIO_CONSOLE_CONSOLE_PORT.into();
+                self.rx_ctrl_fifo
+                    .add(ctrl_msg_reply)
+                    .map_err(|_| Error::RxCtrlQueueAddFailed)?;
 
-			log::trace!("descriptors.len(): {:?}", descriptors.len());
+                ctrl_msg_reply.event = VIRTIO_CONSOLE_PORT_NAME.into();
+                self.rx_ctrl_fifo
+                    .add(ctrl_msg_reply)
+                    .map_err(|_| Error::RxCtrlQueueAddFailed)?;
 
-            let desc_request = descriptors[0];
-            if desc_request.is_write_only() {
-				log::trace!("This is write only");
-                return Err(Error::UnexpectedWriteOnlyDescriptor(0));
-            } else {
-				log::trace!("This is read only");
-			}
-
-            if desc_request.len() as usize != size_of::<VirtioConsoleControl>() {
-				log::trace!("UnexpectedDescriptorSize, len = {:?}", desc_request.len());
-                return Err(Error::UnexpectedDescriptorSize(
-                    size_of::<VirtioConsoleControl>(),
-                    desc_request.len(),
-                ));
+                ctrl_msg_reply.event = VIRTIO_CONSOLE_PORT_OPEN.into();
+                self.rx_ctrl_fifo
+                    .add(ctrl_msg_reply)
+                    .map_err(|_| Error::RxCtrlQueueAddFailed)?;
+            }
+            VIRTIO_CONSOLE_PORT_OPEN => {
+                trace!("VIRTIO_CONSOLE_PORT_OPEN");
+            }
+            _ => {
+                trace!("Uknown control event");
+                return Err(Error::HandleEventUnknown);
             }
+        };
+        Ok(())
+    }
+
+    fn process_ctrl_tx_requests(
+        &mut self,
+        requests: Vec<ConsoleDescriptorChain>,
+        vring: &VringRwLock,
+    ) -> Result<bool> {
+        if requests.is_empty() {
+            return Ok(true);
+        }
 
-			log::trace!("desc_request.len: {}", desc_request.len());
+        for desc_chain in requests {
+            let atomic_mem = self.mem.as_ref().unwrap().memory();
+            let mut reader = desc_chain
+                .clone()
+                .reader(&atomic_mem)
+                .map_err(|_| Error::DescriptorReadFailed)?;
 
-            let mut request = desc_chain
-                .memory()
-                .read_obj::<VirtioConsoleControl>(desc_request.addr())
+            let request = reader
+                .read_obj::<VirtioConsoleControl>()
                 .map_err(|_| Error::DescriptorReadFailed)?;
 
-			self.print_console_frame(request);
+            // Print the receive console frame
+            self.print_console_frame(request);
 
-			self.handle_control_msg(rx_ctrl_vring, request);
+            // Process the received control frame
+            self.handle_control_msg(request)?;
 
-			if let Some(event_fd) = rx_ctrl_vring.get_ref().get_kick() {
-				self.rx_ctrl_event.write(1).unwrap();
-			} else {
-			    // Handle the case where `state` is `None`.
-			    log::trace!("EventFd is not available.");
-			}
+            // trigger a kick to the CTRL_RT_QUEUE
+            self.rx_ctrl_event.write(1).unwrap();
 
-            if vring.add_used(desc_chain.head_index(), desc_request.len()).is_err() {
-                log::trace!("Couldn't return used descriptors to the ring");
-                warn!("Couldn't return used descriptors to the ring");
-            }
+            vring
+                .add_used(desc_chain.head_index(), reader.bytes_read() as u32)
+                .map_err(|_| Error::AddUsedElemFailed(CTRL_TX_QUEUE))?;
         }
 
         Ok(true)
@@ -467,7 +365,6 @@ impl VhostUserConsoleBackend {
 
     /// Process the messages in the vring and dispatch replies
     fn process_rx_queue(&mut self, vring: &VringRwLock) -> Result<()> {
-		log::trace!("process_rx_queue");
         let requests: Vec<_> = vring
             .get_mut()
             .get_queue_mut()
@@ -477,20 +374,15 @@ impl VhostUserConsoleBackend {
 
         if self.process_rx_requests(requests, vring)? {
             // Send notification once all the requests are processed
-			log::trace!("Send notification once all the requests of queue 0 are processed");
             vring
                 .signal_used_queue()
-                .map_err(|_| {
-					log::trace!("NotificationFailed");
-					Error::NotificationFailed
-				})?;
+                .map_err(|_| Error::NotificationFailed)?;
         }
-		Ok(())
+        Ok(())
     }
 
     /// Process the messages in the vring and dispatch replies
-    fn process_tx_queue(&self, vring: &VringRwLock) -> Result<()> {
-		log::trace!("process_tx_queue");
+    fn process_tx_queue(&mut self, vring: &VringRwLock) -> Result<()> {
         let requests: Vec<_> = vring
             .get_mut()
             .get_queue_mut()
@@ -500,21 +392,15 @@ impl VhostUserConsoleBackend {
 
         if self.process_tx_requests(requests, vring)? {
             // Send notification once all the requests are processed
-			log::trace!("Send notification once all the requests of queue 1 are processed");
             vring
                 .signal_used_queue()
-                .map_err(|_| {
-					log::trace!("signal_used_queue error");
-					Error::NotificationFailed
-				})?;
+                .map_err(|_| Error::NotificationFailed)?;
         }
-
         Ok(())
     }
 
     /// Process the messages in the vring and dispatch replies
     fn process_ctrl_rx_queue(&mut self, vring: &VringRwLock) -> Result<()> {
-		log::trace!("process_ctrl_rx_queue");
         let requests: Vec<_> = vring
             .get_mut()
             .get_queue_mut()
@@ -523,18 +409,16 @@ impl VhostUserConsoleBackend {
             .collect();
 
         if self.process_ctrl_rx_requests(requests, vring)? {
-			log::trace!("Send notification once all the requests of queue 2 are processed");
             // Send notification once all the requests are processed
             vring
                 .signal_used_queue()
                 .map_err(|_| Error::NotificationFailed)?;
         }
-		Ok(())
+        Ok(())
     }
 
     /// Process the messages in the vring and dispatch replies
-    fn process_ctrl_tx_queue(&mut self, vring: &VringRwLock, rx_ctrl_vring: &VringRwLock) -> Result<()> {
-		log::trace!("process_ctrl_tx_queue");
+    fn process_ctrl_tx_queue(&mut self, vring: &VringRwLock) -> Result<()> {
         let requests: Vec<_> = vring
             .get_mut()
             .get_queue_mut()
@@ -542,113 +426,257 @@ impl VhostUserConsoleBackend {
             .map_err(|_| Error::DescriptorNotFound)?
             .collect();
 
-        if self.process_ctrl_tx_requests(requests, vring, rx_ctrl_vring)? {
+        if self.process_ctrl_tx_requests(requests, vring)? {
             // Send notification once all the requests are processed
             vring
                 .signal_used_queue()
                 .map_err(|_| Error::NotificationFailed)?;
         }
-		Ok(())
+        Ok(())
     }
 
     /// Set self's VringWorker.
     pub(crate) fn set_vring_worker(
-		&self,
-        vring_worker: &Arc<VringEpollHandler<Arc<RwLock<VhostUserConsoleBackend>>, VringRwLock, ()>>,
+        &self,
+        vring_worker: &Arc<VringEpollHandler<Arc<RwLock<VhostUserConsoleBackend>>>>,
     ) {
-		let rx_event_fd = self.rx_event.as_raw_fd();
-		vring_worker
-            .register_listener(
-				rx_event_fd,
-				EventSet::IN,
-				u64::from(BACKEND_EFD))
+        let rx_event_fd = self.rx_event.as_raw_fd();
+        vring_worker
+            .register_listener(rx_event_fd, EventSet::IN, u64::from(BACKEND_RX_EFD))
             .unwrap();
 
-		let rx_ctrl_event_fd = self.rx_ctrl_event.as_raw_fd();
-		vring_worker
+        let rx_ctrl_event_fd = self.rx_ctrl_event.as_raw_fd();
+        vring_worker
             .register_listener(
-				rx_ctrl_event_fd,
-				EventSet::IN,
-				u64::from(BACKEND_RX_EFD))
+                rx_ctrl_event_fd,
+                EventSet::IN,
+                u64::from(BACKEND_CTRL_RX_EFD),
+            )
             .unwrap();
     }
 
+    pub(crate) fn start_tcp_console_thread(
+        vhu_console: &Arc<RwLock<VhostUserConsoleBackend>>,
+        tcplisener_str: String,
+    ) -> JoinHandle<Result<()>> {
+        let vhu_console = Arc::clone(vhu_console);
+        spawn(move || {
+            loop {
+                let ready = vhu_console.read().unwrap().ready_to_write;
+                let exit = vhu_console.read().unwrap().controller.read().unwrap().exit;
+
+                if exit {
+                    trace!("Thread exits!");
+                    break;
+                } else if ready {
+                    let listener = match TcpListener::bind(tcplisener_str.clone()) {
+                        Ok(listener) => listener,
+                        Err(e) => {
+                            eprintln!("Failed to bind to {}: {}", tcplisener_str, e);
+                            return Ok(());
+                        }
+                    };
+                    listener.set_nonblocking(true).expect("Non-blocking error");
+
+                    println!("Server listening on address: {}", tcplisener_str.clone());
+                    for stream in listener.incoming() {
+                        match stream {
+                            Ok(mut stream) => {
+                                trace!("New connection");
+                                stream.set_nonblocking(true).expect("Non-blocking error");
+
+                                let mut buffer = [0; 1024];
+                                loop {
+                                    let exit =
+                                        vhu_console.read().unwrap().controller.read().unwrap().exit;
+                                    if exit {
+                                        trace!("Thread exits!");
+                                        return Ok(());
+                                    }
+                                    // Write to the stream
+                                    if vhu_console.read().unwrap().output_queue.size() > 0 {
+                                        let byte_stream = vhu_console
+                                            .write()
+                                            .unwrap()
+                                            .output_queue
+                                            .remove()
+                                            .expect("Error removing element from output queue")
+                                            .into_bytes();
+                                        if let Err(e) = stream.write_all(&byte_stream) {
+                                            eprintln!("Error writing to stream: {}", e);
+                                        }
+                                    }
+                                    match stream.read(&mut buffer) {
+                                        Ok(bytes_read) => {
+                                            if bytes_read == 0 {
+                                                println!("Close connection");
+                                                break;
+                                            }
+                                            trace!(
+                                                "Received: {}",
+                                                String::from_utf8_lossy(&buffer[..bytes_read])
+                                            );
+                                            let input_buffer =
+                                                String::from_utf8_lossy(&buffer[..bytes_read])
+                                                    .to_string();
+                                            vhu_console
+                                                .write()
+                                                .unwrap()
+                                                .rx_data_fifo
+                                                .add(input_buffer)
+                                                .unwrap();
+                                            vhu_console.write().unwrap().rx_event.write(1).unwrap();
+                                        }
+                                        Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+                                            continue;
+                                        }
+                                        Err(ref e)
+                                            if e.kind() == io::ErrorKind::BrokenPipe
+                                                || e.kind() == io::ErrorKind::ConnectionReset =>
+                                        {
+                                            trace!("Stream has been closed.");
+                                            break;
+                                        }
+                                        Err(e) => {
+                                            eprintln!("Error reading from socket: {}", e);
+                                        }
+                                    }
+                                }
+                            }
+                            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+                                let exit =
+                                    vhu_console.read().unwrap().controller.read().unwrap().exit;
+                                if exit {
+                                    trace!("Thread exits!");
+                                    return Ok(());
+                                }
+                                continue;
+                            }
+                            Err(e) => {
+                                eprintln!("Error accepting connection: {}", e);
+                                break;
+                            }
+                        }
+                    }
+                }
+            }
+            Ok(())
+        })
+    }
+
     /// Start console thread.
     pub(crate) fn start_console_thread(
-		vhu_console: &Arc<RwLock<VhostUserConsoleBackend>>,
-    ) {
+        vhu_console: &Arc<RwLock<VhostUserConsoleBackend>>,
+    ) -> JoinHandle<Result<()>> {
+        let vhu_console = Arc::clone(vhu_console);
+
+        let exit_eventfd = vhu_console.read().unwrap().exit_event.as_raw_fd();
+        // Spawn a new thread to handle input.
+        spawn(move || {
+            let term = Term::stdout();
+            let mut fdset = FdSet::new();
+            fdset.insert(term.as_raw_fd());
+            fdset.insert(exit_eventfd);
+            let max_fd = fdset.highest().expect("Failed to read fdset!") + 1;
 
-		let vhu_console = Arc::clone(&vhu_console);
-    	print!("Enter text and press Enter: ");
-
-		// Spawn a new thread to handle input.
-    	spawn( move || {
-			loop {
+            loop {
                 let ready = vhu_console.read().unwrap().ready_to_write;
-                if ready {
-
-                    let term = Term::stdout();
-                    let character = term.read_char().unwrap();
-				    log::trace!("You entered: {}", character);
+                let exit = vhu_console.read().unwrap().controller.read().unwrap().exit;
 
-                    // Pass the data to vhu_console and trigger an EventFd
-				    vhu_console.write().unwrap().output_buffer = character.to_string();
-            	    vhu_console.write().unwrap().rx_event.write(1).unwrap();
+                if exit {
+                    trace!("Exit!");
+                    break;
+                } else if ready {
+                    let mut fdset_clone = fdset;
+                    enable_raw_mode().expect("Raw mode error");
+
+                    match select(Some(max_fd), Some(&mut fdset_clone), None, None, None) {
+                        Ok(_num_fds) => {
+                            let exit = vhu_console.read().unwrap().controller.read().unwrap().exit;
+                            if (fdset_clone.contains(exit_eventfd)) && exit {
+                                trace!("Exit!");
+                                break;
+                            }
+
+                            if fdset_clone.contains(term.as_raw_fd()) {
+                                if let Some(character) = match term.read_key().unwrap() {
+                                    Key::Char(character) => Some(character),
+                                    Key::Enter => Some('\n'),
+                                    Key::Tab => Some('\t'),
+                                    Key::Backspace => Some('\u{8}'),
+                                    _ => None,
+                                } {
+                                    // Pass the data to vhu_console and trigger an EventFd
+                                    let input_buffer = character.to_string();
+                                    vhu_console
+                                        .write()
+                                        .unwrap()
+                                        .rx_data_fifo
+                                        .add(input_buffer)
+                                        .unwrap();
+                                    vhu_console.write().unwrap().rx_event.write(1).unwrap();
+                                }
+                            }
+                        }
+                        Err(e) => {
+                            eprintln!("Error in select: {}", e);
+                            break;
+                        }
+                    }
                 }
-			}
-    	});
+            }
+
+            disable_raw_mode().expect("Raw mode error");
+            Ok(())
+        })
+    }
+    pub fn kill_console_thread(&self) {
+        trace!("Kill thread");
+        self.controller.write().unwrap().exit = true;
+        self.exit_event.write(1).unwrap();
     }
 }
 
 /// VhostUserBackendMut trait methods
-impl VhostUserBackendMut<VringRwLock, ()>
-    for VhostUserConsoleBackend
-{
+impl VhostUserBackendMut for VhostUserConsoleBackend {
+    type Vring = VringRwLock;
+    type Bitmap = ();
+
     fn num_queues(&self) -> usize {
-		log::trace!("num_queues: {:?}", NUM_QUEUES);
         NUM_QUEUES
     }
 
     fn max_queue_size(&self) -> usize {
-		log::trace!("max_queue_size: {:?}", QUEUE_SIZE);
         QUEUE_SIZE
     }
 
     fn features(&self) -> u64 {
-        // this matches the current libvhost defaults except VHOST_F_LOG_ALL
-        let features = 1 << VIRTIO_F_VERSION_1
+        1 << VIRTIO_F_VERSION_1
             | 1 << VIRTIO_F_NOTIFY_ON_EMPTY
             | 1 << VIRTIO_RING_F_EVENT_IDX
-			| 1 << VIRTIO_CONSOLE_F_EMERG_WRITE
             | 1 << VIRTIO_RING_F_INDIRECT_DESC
-			| 1 << VIRTIO_CONSOLE_F_MULTIPORT // This could be disabled
-            | VhostUserVirtioFeatures::PROTOCOL_FEATURES.bits();
-
-		log::trace!("vhu_can->features: {:x}", features);
-		features
+            | 1 << VIRTIO_CONSOLE_F_MULTIPORT
+            | VhostUserVirtioFeatures::PROTOCOL_FEATURES.bits()
     }
 
-	fn acked_features(&mut self, _features: u64) {
-		log::trace!("\nacked_features: 0x{:x}\n", _features);
-		self.acked_features = _features;
-	}
+    fn acked_features(&mut self, features: u64) {
+        self.acked_features = features;
+    }
 
     fn protocol_features(&self) -> VhostUserProtocolFeatures {
-        let protocol_features = VhostUserProtocolFeatures::MQ
+        VhostUserProtocolFeatures::MQ
             | VhostUserProtocolFeatures::CONFIG
-            | VhostUserProtocolFeatures::REPLY_ACK;
-
-		log::trace!("protocol_features: {:x}", protocol_features);
-		protocol_features
+            | VhostUserProtocolFeatures::REPLY_ACK
     }
 
     fn get_config(&self, offset: u32, size: u32) -> Vec<u8> {
         // SAFETY: The layout of the structure is fixed and can be initialized by
         // reading its content from byte array.
-		log::trace!("vhu_can->get_config");
         unsafe {
             from_raw_parts(
-                self.controller.write().unwrap()
+                self.controller
+                    .write()
+                    .unwrap()
                     .config()
                     .as_slice()
                     .as_ptr()
@@ -660,11 +688,10 @@ impl VhostUserBackendMut<VringRwLock, ()>
     }
 
     fn set_event_idx(&mut self, enabled: bool) {
-        dbg!(self.event_idx = enabled);
+        self.event_idx = enabled;
     }
 
     fn update_memory(&mut self, mem: GuestMemoryAtomic<GuestMemoryMmap>) -> IoResult<()> {
-		log::trace!("update_memory\n");
         self.mem = Some(mem);
         Ok(())
     }
@@ -672,93 +699,81 @@ impl VhostUserBackendMut<VringRwLock, ()>
     fn handle_event(
         &mut self,
         device_event: u16,
-        evset: EventSet,
+        _evset: EventSet,
         vrings: &[VringRwLock],
         _thread_id: usize,
-    ) -> IoResult<bool> {
-		log::trace!("\nhandle_event:");
-
-		if device_event == RX_QUEUE {
-		    log::trace!("RX_QUEUE\n");
-		    return Ok(false);
-		};
-
-		if device_event == CTRL_RX_QUEUE {
-		    log::trace!("CTRL_RX_QUEUE\n");
-			if !self.ready {
-				return Ok(false);
-			}
-		};
-
-		let vring = if device_event == BACKEND_EFD {
-		    log::trace!("BACKEND_EFD\n");
-		    &vrings[RX_QUEUE as usize]
-		} else if device_event == BACKEND_RX_EFD {
-		    log::trace!("BACKEND_RX_EFD\n");
-		    &vrings[CTRL_RX_QUEUE as usize]
-		} else {
-			&vrings[device_event as usize]
-		};
+    ) -> IoResult<()> {
+        if device_event == RX_QUEUE {
+            // Check if there are any available data
+            if self.rx_data_fifo.size() == 0 {
+                return Ok(());
+            }
+        };
+
+        if device_event == CTRL_RX_QUEUE {
+            // Check if there are any available data and the device is ready
+            if (!self.ready) || (self.rx_ctrl_fifo.size() == 0) {
+                return Ok(());
+            }
+        };
+
+        let vring = if device_event == BACKEND_RX_EFD {
+            &vrings[RX_QUEUE as usize]
+        } else if device_event == BACKEND_CTRL_RX_EFD {
+            &vrings[CTRL_RX_QUEUE as usize]
+        } else {
+            &vrings[device_event as usize]
+        };
 
         if self.event_idx {
-            // vm-virtio's Queue implementation only checks avail_index
-            // once, so to properly support EVENT_IDX we need to keep
-            // calling process_request_queue() until it stops finding
-            // new requests on the queue.
             loop {
                 vring.disable_notification().unwrap();
                 match device_event {
+                    RX_QUEUE => self.process_rx_queue(vring),
                     TX_QUEUE => {
-				        self.ready_to_write = true;
+                        self.ready_to_write = true;
                         self.process_tx_queue(vring)
-                    },
+                    }
                     CTRL_RX_QUEUE => self.process_ctrl_rx_queue(vring),
-					CTRL_TX_QUEUE => {
-						let rx_ctrl_vring = &vrings[CTRL_RX_QUEUE as usize];
-						self.process_ctrl_tx_queue(vring, rx_ctrl_vring)
-					},
-                    BACKEND_EFD => {
-						let _ = self.rx_event.read();
-						self.process_rx_queue(vring)
-					},
+                    CTRL_TX_QUEUE => self.process_ctrl_tx_queue(vring),
                     BACKEND_RX_EFD => {
-						let _ = self.rx_ctrl_event.read();
-						self.process_ctrl_rx_queue(vring)
-					},
-                    _ => Err(Error::HandleEventUnknown.into()),
+                        let _ = self.rx_event.read();
+                        self.process_rx_queue(vring)
+                    }
+                    BACKEND_CTRL_RX_EFD => {
+                        let _ = self.rx_ctrl_event.read();
+                        self.process_ctrl_rx_queue(vring)
+                    }
+                    _ => Err(Error::HandleEventUnknown),
                 }?;
                 if !vring.enable_notification().unwrap() {
                     break;
                 }
             }
-        } else {                                                                           
-            // Without EVENT_IDX, a single call is enough.                                 
+        } else {
             match device_event {
+                RX_QUEUE => self.process_rx_queue(vring),
                 TX_QUEUE => {
-			        self.ready_to_write = true;
+                    self.ready_to_write = true;
                     self.process_tx_queue(vring)
-                },
+                }
                 CTRL_RX_QUEUE => self.process_ctrl_rx_queue(vring),
-				CTRL_TX_QUEUE => {
-					let rx_ctrl_vring = &vrings[CTRL_RX_QUEUE as usize];
-					self.process_ctrl_tx_queue(vring, rx_ctrl_vring)
-				},
-                BACKEND_EFD => {
-					let _ = self.rx_event.read();
-					self.process_rx_queue(vring)
-				},
+                CTRL_TX_QUEUE => self.process_ctrl_tx_queue(vring),
                 BACKEND_RX_EFD => {
-					let _ = self.rx_ctrl_event.read();
-					self.process_ctrl_rx_queue(vring)
-				},
-                _ => Err(Error::HandleEventUnknown.into()),
+                    let _ = self.rx_event.read();
+                    self.process_rx_queue(vring)
+                }
+                BACKEND_CTRL_RX_EFD => {
+                    let _ = self.rx_ctrl_event.read();
+                    self.process_ctrl_rx_queue(vring)
+                }
+                _ => Err(Error::HandleEventUnknown),
             }?;
         }
-        Ok(false)
+        Ok(())
     }
 
     fn exit_event(&self, _thread_index: usize) -> Option<EventFd> {
-		dbg!("exit_event\n");
         self.exit_event.try_clone().ok()
     }
 }
@@ -766,29 +781,503 @@ impl VhostUserBackendMut<VringRwLock, ()>
 #[cfg(test)]
 mod tests {
     use super::*;
+    use virtio_bindings::virtio_ring::{VRING_DESC_F_NEXT, VRING_DESC_F_WRITE};
+    use virtio_queue::{mock::MockSplitQueue, Descriptor, Queue};
+    use vm_memory::{Bytes, GuestAddress, GuestMemoryAtomic, GuestMemoryMmap};
+
+    #[test]
+    fn test_vhost_user_console_backend_creation() {
+        let console_controller = Arc::new(RwLock::new(ConsoleController::new(BackendType::Nested)));
+        let vhost_user_console_backend = VhostUserConsoleBackend::new(console_controller)
+            .expect("Failed create vhuconsole backend");
+
+        assert_eq!(vhost_user_console_backend.acked_features, 0);
+        assert!(!vhost_user_console_backend.event_idx);
+        assert!(!vhost_user_console_backend.ready);
+        assert!(!vhost_user_console_backend.ready_to_write);
+    }
+
+    #[test]
+    fn test_virtio_console_empty_handle_request() {
+        let console_controller = Arc::new(RwLock::new(ConsoleController::new(BackendType::Nested)));
+        let mut vu_console_backend = VhostUserConsoleBackend::new(console_controller)
+            .expect("Failed create vhuconsole backend");
+
+        // Artificial memory
+        let mem = GuestMemoryAtomic::new(
+            GuestMemoryMmap::<()>::from_ranges(&[(GuestAddress(0), 0x1000)]).unwrap(),
+        );
+
+        // Update memory
+        vu_console_backend.update_memory(mem.clone()).unwrap();
+
+        // Artificial Vring
+        let vring = VringRwLock::new(mem, 0x1000).unwrap();
+        vring.set_queue_info(0x100, 0x200, 0x300).unwrap();
+        vring.set_queue_ready(true);
+        let list_vrings = [vring.clone(), vring.clone(), vring.clone(), vring.clone()];
+
+        vu_console_backend
+            .handle_event(RX_QUEUE, EventSet::IN, &list_vrings, 0)
+            .unwrap();
+
+        vu_console_backend
+            .handle_event(TX_QUEUE, EventSet::IN, &list_vrings, 0)
+            .unwrap();
+
+        vu_console_backend
+            .handle_event(CTRL_RX_QUEUE, EventSet::IN, &list_vrings, 0)
+            .unwrap();
+
+        vu_console_backend
+            .handle_event(CTRL_TX_QUEUE, EventSet::IN, &list_vrings, 0)
+            .unwrap();
+
+        vu_console_backend
+            .handle_event(BACKEND_RX_EFD, EventSet::IN, &list_vrings, 0)
+            .unwrap();
+
+        vu_console_backend
+            .handle_event(BACKEND_CTRL_RX_EFD, EventSet::IN, &list_vrings, 0)
+            .unwrap();
+    }
+
+    #[test]
+    fn test_virtio_console_empty_requests() {
+        let console_controller = Arc::new(RwLock::new(ConsoleController::new(BackendType::Nested)));
+        let mut vu_console_backend = VhostUserConsoleBackend::new(console_controller)
+            .expect("Failed create vhuconsole backend");
+
+        // Artificial memory
+        let mem = GuestMemoryAtomic::new(
+            GuestMemoryMmap::<()>::from_ranges(&[(GuestAddress(0), 0x1000)]).unwrap(),
+        );
+
+        // Artificial Vring
+        let vring = VringRwLock::new(mem.clone(), 0x1000).unwrap();
+
+        // Empty descriptor chain should be ignored
+        assert!(vu_console_backend
+            .process_rx_requests(Vec::<ConsoleDescriptorChain>::new(), &vring)
+            .is_ok());
+        assert!(vu_console_backend
+            .process_tx_requests(Vec::<ConsoleDescriptorChain>::new(), &vring)
+            .is_ok());
+        assert!(vu_console_backend
+            .process_ctrl_rx_requests(Vec::<ConsoleDescriptorChain>::new(), &vring)
+            .is_ok());
+        assert!(vu_console_backend
+            .process_ctrl_tx_requests(Vec::<ConsoleDescriptorChain>::new(), &vring)
+            .is_ok());
+    }
+
+    fn build_desc_chain(
+        mem: &GuestMemoryMmap,
+        count: u16,
+        flags: Vec<u16>,
+        len: u32,
+    ) -> ConsoleDescriptorChain {
+        let vq = MockSplitQueue::new(mem, 16);
+        let mut desc_vec = Vec::new();
+
+        //Create a descriptor chain with @count descriptors.
+        for i in 0..count {
+            let desc_flags = if i < count - 1 {
+                flags[i as usize] | VRING_DESC_F_NEXT as u16
+            } else {
+                flags[i as usize] & !VRING_DESC_F_NEXT as u16
+            };
+
+            let desc = Descriptor::new((0x100 * (i + 1)) as u64, len, desc_flags, i + 1);
+            desc_vec.push(desc);
+        }
+
+        vq.add_desc_chains(&desc_vec, 0).unwrap();
+
+        // Create descriptor chain from pre-filled memory
+        vq.create_queue::<Queue>()
+            .unwrap()
+            .iter(GuestMemoryAtomic::new(mem.clone()).memory())
+            .unwrap()
+            .next()
+            .unwrap()
+    }
 
     #[test]
-    fn test_virtio_console_control_byte_valued() {
-        let control = VirtioConsoleControl {
-            id: Le32::from(1),
-            event: Le16::from(2),
-            value: Le16::from(3),
+    fn test_virtio_console_ctrl_rx_request() {
+        let console_controller = Arc::new(RwLock::new(ConsoleController::new(BackendType::Nested)));
+        let mut vu_console_backend = VhostUserConsoleBackend::new(console_controller)
+            .expect("Failed create vhuconsole backend");
+
+        // Artificial memory
+        let mem = GuestMemoryMmap::<()>::from_ranges(&[(GuestAddress(0), 0x1000)]).unwrap();
+
+        // Test 1: Empty queue
+        let mem1 = GuestMemoryAtomic::new(mem.clone());
+        let vring = VringRwLock::new(mem1.clone(), 0x1000).unwrap();
+        assert!(vu_console_backend
+            .process_ctrl_rx_requests(vec![], &vring)
+            .unwrap());
+
+        // Test 2: Found no rx elements
+        let desc_chain = build_desc_chain(&mem, 1, vec![0], 0x200);
+        let mem1 = GuestMemoryAtomic::new(mem.clone());
+        let vring = VringRwLock::new(mem1.clone(), 0x1000).unwrap();
+        vu_console_backend.update_memory(mem1).unwrap();
+
+        assert!(!vu_console_backend
+            .process_ctrl_rx_requests(vec![desc_chain], &vring)
+            .unwrap());
+
+        // Test 3: empty queue
+        let desc_chain = build_desc_chain(&mem, 1, vec![VRING_DESC_F_WRITE as u16], 0x200);
+        let mem1 = GuestMemoryAtomic::new(mem.clone());
+        let vring = VringRwLock::new(mem1.clone(), 0x1000).unwrap();
+        vu_console_backend.update_memory(mem1).unwrap();
+        assert!(!vu_console_backend
+            .process_ctrl_rx_requests(vec![desc_chain.clone()], &vring)
+            .unwrap());
+
+        // Test 4: the written desc reply
+        let ctrl_msg = VirtioConsoleControl {
+            id: 0.into(),
+            event: VIRTIO_CONSOLE_PORT_ADD.into(),
+            value: 1.into(),
         };
+        let _ = vu_console_backend.rx_ctrl_fifo.add(ctrl_msg);
 
-        let bytes = control.to_le_bytes();
+        assert!(vu_console_backend
+            .process_ctrl_rx_requests(vec![desc_chain.clone()], &vring)
+            .unwrap());
 
-        assert_eq!(bytes.len(), 10);
+        let ctrl_msg_reply = desc_chain
+            .memory()
+            .read_obj::<VirtioConsoleControl>(vm_memory::GuestAddress(0x100_u64))
+            .map_err(|_| Error::DescriptorReadFailed)
+            .unwrap();
+
+        assert_eq!(ctrl_msg.id, ctrl_msg_reply.id);
+        assert_eq!(ctrl_msg.event, ctrl_msg_reply.event);
+        assert_eq!(ctrl_msg.value, ctrl_msg_reply.value);
+
+        // Test 5: if message is VIRTIO_CONSOLE_PORT_NAME
+        let desc_chain = build_desc_chain(&mem, 1, vec![VRING_DESC_F_WRITE as u16], 0x200);
+        let mem1 = GuestMemoryAtomic::new(mem.clone());
+        let vring = VringRwLock::new(mem1.clone(), 0x1000).unwrap();
+        vu_console_backend.update_memory(mem1).unwrap();
+
+        let ctrl_msg = VirtioConsoleControl {
+            id: 0.into(),
+            event: VIRTIO_CONSOLE_PORT_NAME.into(),
+            value: 1.into(),
+        };
+        let _ = vu_console_backend.rx_ctrl_fifo.add(ctrl_msg);
+
+        assert!(vu_console_backend
+            .process_ctrl_rx_requests(vec![desc_chain.clone()], &vring)
+            .unwrap());
     }
 
     #[test]
-    fn test_vhost_user_console_backend_creation() {
-        let console_controller = Arc::new(RwLock::new(ConsoleController::new(String::from("test_console")).unwrap()));
-        let vhost_user_console_backend = VhostUserConsoleBackend::new(console_controller).unwrap();
+    fn test_virtio_console_ctrl_tx_request() {
+        let console_controller = Arc::new(RwLock::new(ConsoleController::new(BackendType::Nested)));
+        let mut vu_console_backend = VhostUserConsoleBackend::new(console_controller)
+            .expect("Failed create vhuconsole backend");
+
+        // Artificial memory
+        let mem = GuestMemoryMmap::<()>::from_ranges(&[(GuestAddress(0), 0x1000)]).unwrap();
+
+        // Test 1: Empty queue
+        let mem1 = GuestMemoryAtomic::new(mem.clone());
+        let vring = VringRwLock::new(mem1.clone(), 0x1000).unwrap();
+        assert!(vu_console_backend
+            .process_ctrl_tx_requests(vec![], &vring)
+            .unwrap());
+
+        // Test 2: Found no descriptors
+        let desc_chain = build_desc_chain(&mem, 1, vec![VRING_DESC_F_WRITE as u16], 0x200);
+        let mem1 = GuestMemoryAtomic::new(mem.clone());
+        let vring = VringRwLock::new(mem1.clone(), 0x1000).unwrap();
+        vu_console_backend.update_memory(mem1).unwrap();
+
+        assert_eq!(
+            vu_console_backend
+                .process_ctrl_tx_requests(vec![desc_chain], &vring)
+                .unwrap_err(),
+            Error::DescriptorReadFailed
+        );
+
+        // Test 3: Smaller descriptor len than a console ctrl message
+        let desc_chain = build_desc_chain(&mem, 1, vec![0], 0x2);
+        let mem1 = GuestMemoryAtomic::new(mem.clone());
+        let vring = VringRwLock::new(mem1.clone(), 0x1000).unwrap();
+        vu_console_backend.update_memory(mem1).unwrap();
+        assert_eq!(
+            vu_console_backend
+                .process_ctrl_tx_requests(vec![desc_chain.clone()], &vring)
+                .unwrap_err(),
+            Error::DescriptorReadFailed
+        );
+
+        // Test 4: Complete function successfully -- VIRTIO_CONSOLE_PORT_READY message
+        let desc_chain = build_desc_chain(&mem, 1, vec![0], 0x8);
+        let mem1 = GuestMemoryAtomic::new(mem.clone());
+        let vring = VringRwLock::new(mem1.clone(), 0x1000).unwrap();
+        vu_console_backend.update_memory(mem1).unwrap();
+        assert!(vu_console_backend
+            .process_ctrl_tx_requests(vec![desc_chain.clone()], &vring)
+            .unwrap());
+    }
 
-        assert_eq!(vhost_user_console_backend.acked_features, 0);
-        assert_eq!(vhost_user_console_backend.event_idx, false);
-        assert_eq!(vhost_user_console_backend.ready, false);
-        assert_eq!(vhost_user_console_backend.ready_to_write, false);
-        assert_eq!(vhost_user_console_backend.output_buffer, String::new());
+    #[test]
+    fn test_virtio_console_handle_control_msg() {
+        let console_controller = Arc::new(RwLock::new(ConsoleController::new(BackendType::Nested)));
+        let mut vu_console_backend = VhostUserConsoleBackend::new(console_controller)
+            .expect("Failed create vhuconsole backend");
+
+        // Artificial memory & update device's memory
+        let mem = GuestMemoryMmap::<()>::from_ranges(&[(GuestAddress(0), 0x1000)]).unwrap();
+        let mem_1 = GuestMemoryAtomic::new(mem.clone());
+        vu_console_backend.update_memory(mem_1.clone()).unwrap();
+
+        // Test 1: Empty queue
+        let ctrl_msg_1 = VirtioConsoleControl {
+            id: 0.into(),
+            event: VIRTIO_CONSOLE_DEVICE_READY.into(),
+            value: 1.into(),
+        };
+
+        let ctrl_msg_2 = VirtioConsoleControl {
+            id: 0.into(),
+            event: VIRTIO_CONSOLE_PORT_READY.into(),
+            value: 1.into(),
+        };
+
+        let ctrl_msg_3 = VirtioConsoleControl {
+            id: 0.into(),
+            event: VIRTIO_CONSOLE_PORT_OPEN.into(),
+            value: 1.into(),
+        };
+
+        let ctrl_msg_err = VirtioConsoleControl {
+            id: 0.into(),
+            event: 4.into(),
+            value: 1.into(),
+        };
+
+        assert!(vu_console_backend.handle_control_msg(ctrl_msg_3).is_ok());
+
+        assert_eq!(
+            vu_console_backend
+                .handle_control_msg(ctrl_msg_err)
+                .unwrap_err(),
+            Error::HandleEventUnknown
+        );
+
+        assert!(vu_console_backend.handle_control_msg(ctrl_msg_1).is_ok());
+
+        // Update memory
+        let mem_1 = GuestMemoryAtomic::new(mem.clone());
+        vu_console_backend.update_memory(mem_1.clone()).unwrap();
+
+        assert!(vu_console_backend.handle_control_msg(ctrl_msg_2).is_ok());
+    }
+
+    #[test]
+    fn test_virtio_console_tx_request() {
+        let console_controller = Arc::new(RwLock::new(ConsoleController::new(BackendType::Nested)));
+        let mut vu_console_backend = VhostUserConsoleBackend::new(console_controller)
+            .expect("Failed create vhuconsole backend");
+
+        // Artificial memory
+        let mem = GuestMemoryMmap::<()>::from_ranges(&[(GuestAddress(0), 0x1000)]).unwrap();
+
+        // Test 1: Empty queue
+        let mem1 = GuestMemoryAtomic::new(mem.clone());
+        let vring = VringRwLock::new(mem1, 0x1000).unwrap();
+        assert!(vu_console_backend
+            .process_tx_requests(vec![], &vring)
+            .is_ok());
+
+        // Test 2: Empty buffer
+        let desc_chain = build_desc_chain(&mem, 1, vec![0], 0x200);
+        let mem1 = GuestMemoryAtomic::new(mem.clone());
+        let vring = VringRwLock::new(mem1.clone(), 0x1000).unwrap();
+        vu_console_backend.update_memory(mem1).unwrap();
+        assert!(vu_console_backend
+            .process_tx_requests(vec![desc_chain], &vring)
+            .is_ok());
+
+        // Test 3: Fill message to the buffer
+        let desc_chain = build_desc_chain(&mem, 1, vec![0], 0x200);
+        let desc_addr = desc_chain.clone().collect::<Vec<_>>()[0].addr();
+
+        // Build the Vec with the desired string
+        let mut buffer: Vec<u8> = Vec::new();
+        let string_bytes = "Hello!".as_bytes();
+        buffer.extend_from_slice(string_bytes);
+
+        // Write a new buffer into the desc_chain
+        desc_chain.memory().write_slice(&buffer, desc_addr).unwrap();
+
+        // Verify that it is written
+        let mut read_buffer: Vec<u8> = vec![0; 0x200];
+        desc_chain
+            .memory()
+            .read_slice(&mut read_buffer, desc_addr)
+            .expect("Failed to read");
+        let read_buffer: Vec<u8> = read_buffer.iter().take(buffer.len()).copied().collect();
+
+        assert_eq!(
+            String::from_utf8(read_buffer).unwrap(),
+            String::from_utf8(buffer).unwrap()
+        );
+
+        assert!(vu_console_backend
+            .process_tx_requests(vec![desc_chain], &vring)
+            .is_ok());
+    }
+
+    #[test]
+    fn test_virtio_console_tx_request_network() {
+        let console_controller =
+            Arc::new(RwLock::new(ConsoleController::new(BackendType::Network)));
+        let mut vu_console_backend = VhostUserConsoleBackend::new(console_controller)
+            .expect("Failed create vhuconsole backend");
+
+        // Artificial memory
+        let mem = GuestMemoryMmap::<()>::from_ranges(&[(GuestAddress(0), 0x1000)]).unwrap();
+        let desc_chain = build_desc_chain(&mem, 1, vec![0], 0x200);
+        let mem1 = GuestMemoryAtomic::new(mem.clone());
+        let vring = VringRwLock::new(mem1.clone(), 0x1000).unwrap();
+
+        // Test: Fill message to the buffer
+        vu_console_backend.update_memory(mem1).unwrap();
+        let desc_addr = desc_chain.clone().collect::<Vec<_>>()[0].addr();
+
+        // Build the Vec with the desired string
+        let mut buffer: Vec<u8> = Vec::new();
+        let string_bytes = "Hello!".as_bytes();
+        buffer.extend_from_slice(string_bytes);
+
+        // Write a new buffer into the desc_chain
+        desc_chain.memory().write_slice(&buffer, desc_addr).unwrap();
+
+        // Verify that it is written
+        let mut read_buffer: Vec<u8> = vec![0; 0x200];
+        desc_chain
+            .memory()
+            .read_slice(&mut read_buffer, desc_addr)
+            .expect("Failed to read");
+        let read_buffer: Vec<u8> = read_buffer.iter().take(buffer.len()).copied().collect();
+
+        assert_eq!(
+            String::from_utf8(read_buffer).unwrap(),
+            String::from_utf8(buffer).unwrap()
+        );
+
+        assert!(vu_console_backend
+            .process_tx_requests(vec![desc_chain], &vring)
+            .is_ok());
+    }
+
+    #[test]
+    fn test_virtio_console_rx_request() {
+        let console_controller = Arc::new(RwLock::new(ConsoleController::new(BackendType::Nested)));
+        let mut vu_console_backend = VhostUserConsoleBackend::new(console_controller)
+            .expect("Failed create vhuconsole backend");
+
+        // Artificial memory
+        let mem = GuestMemoryMmap::<()>::from_ranges(&[(GuestAddress(0), 0x1000)]).unwrap();
+
+        // Test 1: Empty queue
+        let mem1 = GuestMemoryAtomic::new(mem.clone());
+        let vring = VringRwLock::new(mem1, 0x1000).unwrap();
+        assert!(vu_console_backend
+            .process_rx_requests(vec![], &vring)
+            .is_ok());
+
+        // Test 2: Empty buffer
+        let desc_chain = build_desc_chain(&mem, 1, vec![0], 0x200);
+        let mem1 = GuestMemoryAtomic::new(mem.clone());
+        let vring = VringRwLock::new(mem1.clone(), 0x1000).unwrap();
+        vu_console_backend.update_memory(mem1).unwrap();
+        assert!(!vu_console_backend
+            .process_rx_requests(vec![desc_chain], &vring)
+            .unwrap());
+
+        // Test 3: Fill message to the buffer. The descriptor should be write-only
+        let desc_chain = build_desc_chain(&mem, 1, vec![0], 0x200);
+        let mem1 = GuestMemoryAtomic::new(mem.clone());
+        let vring = VringRwLock::new(mem1.clone(), 0x1000).unwrap();
+        vu_console_backend.update_memory(mem1).unwrap();
+
+        let input_buffer = "Hello!".to_string();
+        let _ = vu_console_backend.rx_data_fifo.add(input_buffer.clone());
+        assert_eq!(
+            vu_console_backend
+                .process_rx_requests(vec![desc_chain], &vring)
+                .unwrap_err(),
+            Error::DescriptorWriteFailed
+        );
+
+        // Test 4: Fill message to the buffer. Everything should work!
+        let desc_chain = build_desc_chain(&mem, 1, vec![VRING_DESC_F_WRITE as u16], 0x200);
+        let mem1 = GuestMemoryAtomic::new(mem.clone());
+        let vring = VringRwLock::new(mem1.clone(), 0x1000).unwrap();
+        vu_console_backend.update_memory(mem1).unwrap();
+
+        let input_buffer = "Hello!".to_string();
+        let _ = vu_console_backend.rx_data_fifo.add(input_buffer.clone());
+        assert!(vu_console_backend
+            .process_rx_requests(vec![desc_chain.clone()], &vring)
+            .unwrap());
+
+        // Test 5: Verify written data
+        let desc_addr = GuestAddress(0x100);
+        let mut read_buffer: Vec<u8> = vec![0; 0x100];
+        desc_chain
+            .memory()
+            .read_slice(&mut read_buffer, desc_addr)
+            .expect("Failed to read");
+
+        let read_buffer: Vec<u8> = read_buffer
+            .iter()
+            .take(input_buffer.len())
+            .copied()
+            .collect();
+
+        assert_eq!(String::from_utf8(read_buffer).unwrap(), input_buffer);
+    }
+
+    #[test]
+    fn test_virtio_console_start_tcp_console_thread() {
+        let console_controller = Arc::new(RwLock::new(ConsoleController::new(BackendType::Nested)));
+        let vu_console_backend = Arc::new(RwLock::new(
+            VhostUserConsoleBackend::new(console_controller)
+                .expect("Failed create vhuconsole backend"),
+        ));
+        let tcp_addr = "127.0.0.1:12345".to_string();
+
+        let read_handle = VhostUserConsoleBackend::start_tcp_console_thread(
+            &vu_console_backend,
+            tcp_addr.clone(),
+        );
+        vu_console_backend.read().unwrap().kill_console_thread();
+        assert!(read_handle.join().is_ok());
+    }
+
+    #[test]
+    fn test_virtio_console_start_nested_console_thread() {
+        let console_controller = Arc::new(RwLock::new(ConsoleController::new(BackendType::Nested)));
+        let vu_console_backend = Arc::new(RwLock::new(
+            VhostUserConsoleBackend::new(console_controller)
+                .expect("Failed create vhuconsole backend"),
+        ));
+
+        let read_handle = VhostUserConsoleBackend::start_console_thread(&vu_console_backend);
+
+        vu_console_backend.read().unwrap().kill_console_thread();
+        assert!(read_handle.join().is_ok());
     }
 }
diff --git a/meta-egvirt/recipes-extended/vhost-device-console/vhost-device-console-0.1.0/src/virtio_console.rs b/meta-egvirt/recipes-extended/vhost-device-console/vhost-device-console-0.1.0/src/virtio_console.rs
new file mode 100755
index 00000000..323473f6
--- /dev/null
+++ b/meta-egvirt/recipes-extended/vhost-device-console/vhost-device-console-0.1.0/src/virtio_console.rs
@@ -0,0 +1,60 @@
+// Console virtio bindings
+//
+// Copyright 2023-2024 VIRTUAL OPEN SYSTEMS SAS. All Rights Reserved.
+//          Timos Ampelikiotis <t.ampelikiotis@virtualopensystems.com>
+//
+// SPDX-License-Identifier: Apache-2.0 or BSD-3-Clause
+
+use vm_memory::{ByteValued, Le16, Le32};
+
+/// Feature bit numbers
+#[allow(dead_code)]
+pub const VIRTIO_CONSOLE_F_SIZE: u16 = 0;
+pub const VIRTIO_CONSOLE_F_MULTIPORT: u16 = 1;
+#[allow(dead_code)]
+pub const VIRTIO_CONSOLE_F_EMERG_WRITE: u16 = 2;
+
+/// Console virtio control messages
+pub const VIRTIO_CONSOLE_DEVICE_READY: u16 = 0;
+pub const VIRTIO_CONSOLE_PORT_ADD: u16 = 1;
+#[allow(dead_code)]
+pub const VIRTIO_CONSOLE_PORT_REMOVE: u16 = 2;
+pub const VIRTIO_CONSOLE_PORT_READY: u16 = 3;
+pub const VIRTIO_CONSOLE_CONSOLE_PORT: u16 = 4;
+#[allow(dead_code)]
+pub const VIRTIO_CONSOLE_RESIZE: u16 = 5;
+pub const VIRTIO_CONSOLE_PORT_OPEN: u16 = 6;
+pub const VIRTIO_CONSOLE_PORT_NAME: u16 = 7;
+
+/// Virtio Console Config
+#[derive(Copy, Clone, Debug, Default, PartialEq)]
+#[repr(C)]
+pub(crate) struct VirtioConsoleConfig {
+    pub cols: Le16,
+    pub rows: Le16,
+    pub max_nr_ports: Le32,
+    pub emerg_wr: Le32,
+}
+
+// SAFETY: The layout of the structure is fixed and can be initialized by
+// reading its content from byte array.
+unsafe impl ByteValued for VirtioConsoleConfig {}
+
+#[derive(Copy, Clone, Debug, Default, PartialEq)]
+#[repr(C)]
+pub(crate) struct VirtioConsoleControl {
+    pub id: Le32,
+    pub event: Le16,
+    pub value: Le16,
+}
+
+impl VirtioConsoleControl {
+    pub fn to_le_bytes(self) -> Vec<u8> {
+        let mut buffer = Vec::new();
+
+        buffer.extend_from_slice(&self.id.to_native().to_le_bytes());
+        buffer.extend_from_slice(&self.event.to_native().to_le_bytes());
+        buffer.extend_from_slice(&self.value.to_native().to_le_bytes());
+        buffer
+    }
+}