Update vhost-device-console

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

1 files changed, 960 insertions, 471 deletions

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());
     }
 }