fix(idm): reimplement packet processing algorithm (#330)

* chore(xen): rewrite event channel code

* fix(idm): repair idm bugs on the file backend

Cargo.lock

@@ -1527,6 +1527,7 @@ dependencies = [
 name = "krata-xenevtchn"
 version = "0.0.15"
 dependencies = [
+ "byteorder",
  "libc",
  "log",
  "nix 0.29.0",

crates/ctl/src/cli/zone/attach.rs

@@ -26,7 +26,7 @@ impl ZoneAttachCommand {
         let input = StdioConsoleStream::stdin_stream(zone_id.clone()).await;
         let output = client.attach_zone_console(input).await?.into_inner();
         let stdout_handle =
-            tokio::task::spawn(async move { StdioConsoleStream::stdout(output).await });
+            tokio::task::spawn(async move { StdioConsoleStream::stdout(output, true).await });
         let exit_hook_task = StdioConsoleStream::zone_exit_hook(zone_id.clone(), events).await?;
         let code = select! {
             x = stdout_handle => {

crates/ctl/src/cli/zone/launch.rs

@@ -155,7 +155,7 @@ impl ZoneLaunchCommand {
             let input = StdioConsoleStream::stdin_stream(id.clone()).await;
             let output = client.attach_zone_console(input).await?.into_inner();
             let stdout_handle =
-                tokio::task::spawn(async move { StdioConsoleStream::stdout(output).await });
+                tokio::task::spawn(async move { StdioConsoleStream::stdout(output, true).await });
             let exit_hook_task = StdioConsoleStream::zone_exit_hook(id.clone(), events).await?;
             select! {
                 x = stdout_handle => {

crates/ctl/src/cli/zone/logs.rs

@@ -43,7 +43,7 @@ impl ZoneLogsCommand {
         };
         let output = client.attach_zone_console(input).await?.into_inner();
         let stdout_handle =
-            tokio::task::spawn(async move { StdioConsoleStream::stdout(output).await });
+            tokio::task::spawn(async move { StdioConsoleStream::stdout(output, false).await });
         let exit_hook_task = StdioConsoleStream::zone_exit_hook(zone_id.clone(), events).await?;
         let code = select! {
             x = stdout_handle => {

crates/ctl/src/console.rs

@@ -73,8 +73,8 @@ impl StdioConsoleStream {
         }
     }
 
-    pub async fn stdout(mut stream: Streaming<ZoneConsoleReply>) -> Result<()> {
+    pub async fn stdout(mut stream: Streaming<ZoneConsoleReply>, raw: bool) -> Result<()> {
-        if stdin().is_tty() {
+        if raw && stdin().is_tty() {
             enable_raw_mode()?;
             StdioConsoleStream::register_terminal_restore_hook()?;
         }

crates/daemon/src/idm.rs

@@ -254,9 +254,9 @@ pub struct IdmDaemonBackend {
 
 #[async_trait::async_trait]
 impl IdmBackend for IdmDaemonBackend {
-    async fn recv(&mut self) -> Result<IdmTransportPacket> {
+    async fn recv(&mut self) -> Result<Vec<IdmTransportPacket>> {
         if let Some(packet) = self.rx_receiver.recv().await {
-            Ok(packet)
+            Ok(vec![packet])
         } else {
             Err(anyhow!("idm receive channel closed"))
         }

crates/krata/src/idm/client.rs

@@ -9,13 +9,13 @@ use std::{
 };
 
 use anyhow::{anyhow, Result};
-use bytes::{BufMut, BytesMut};
+use bytes::{Buf, BufMut, BytesMut};
 use log::{debug, error};
 use nix::sys::termios::{cfmakeraw, tcgetattr, tcsetattr, SetArg};
 use prost::Message;
 use tokio::{
     fs::File,
-    io::{unix::AsyncFd, AsyncReadExt, AsyncWriteExt},
+    io::{AsyncReadExt, AsyncWriteExt},
     select,
     sync::{
         broadcast,
@@ -43,12 +43,13 @@ const IDM_PACKET_MAX_SIZE: usize = 20 * 1024 * 1024;
 
 #[async_trait::async_trait]
 pub trait IdmBackend: Send {
-    async fn recv(&mut self) -> Result<IdmTransportPacket>;
+    async fn recv(&mut self) -> Result<Vec<IdmTransportPacket>>;
     async fn send(&mut self, packet: IdmTransportPacket) -> Result<()>;
 }
 
 pub struct IdmFileBackend {
-    read_fd: Arc<Mutex<AsyncFd<File>>>,
+    read: Arc<Mutex<File>>,
+    read_buffer: BytesMut,
     write: Arc<Mutex<File>>,
 }
 
@@ -57,7 +58,8 @@ impl IdmFileBackend {
         IdmFileBackend::set_raw_port(&read_file)?;
         IdmFileBackend::set_raw_port(&write_file)?;
         Ok(IdmFileBackend {
-            read_fd: Arc::new(Mutex::new(AsyncFd::new(read_file)?)),
+            read: Arc::new(Mutex::new(read_file)),
+            read_buffer: BytesMut::new(),
             write: Arc::new(Mutex::new(write_file)),
         })
     }
@@ -72,26 +74,58 @@ impl IdmFileBackend {
 
 #[async_trait::async_trait]
 impl IdmBackend for IdmFileBackend {
-    async fn recv(&mut self) -> Result<IdmTransportPacket> {
+    async fn recv(&mut self) -> Result<Vec<IdmTransportPacket>> {
-        let mut fd = self.read_fd.lock().await;
+        let mut data = vec![0; 8192];
-        let mut guard = fd.readable_mut().await?;
+        let mut first = true;
-        let b1 = guard.get_inner_mut().read_u8().await?;
+        'read_more: loop {
-        if b1 != 0xff {
+            let mut packets = Vec::new();
-            return Ok(IdmTransportPacket::default());
+            if !first {
+                if !packets.is_empty() {
+                    return Ok(packets);
                 }
-        let b2 = guard.get_inner_mut().read_u8().await?;
+                let size = self.read.lock().await.read(&mut data).await?;
-        if b2 != 0xff {
+                self.read_buffer.extend_from_slice(&data[0..size]);
-            return Ok(IdmTransportPacket::default());
             }
-        let size = guard.get_inner_mut().read_u32_le().await?;
+            first = false;
-        if size == 0 {
+            loop {
-            return Ok(IdmTransportPacket::default());
+                if self.read_buffer.len() < 6 {
+                    continue 'read_more;
                 }
-        let mut buffer = vec![0u8; size as usize];
+
-        guard.get_inner_mut().read_exact(&mut buffer).await?;
+                let b1 = self.read_buffer[0];
-        match IdmTransportPacket::decode(buffer.as_slice()) {
+                let b2 = self.read_buffer[1];
-            Ok(packet) => Ok(packet),
+
-            Err(error) => Err(anyhow!("received invalid idm packet: {}", error)),
+                if b1 != 0xff || b2 != 0xff {
+                    self.read_buffer.clear();
+                    continue 'read_more;
+                }
+
+                let size = (self.read_buffer[2] as u32
+                    | (self.read_buffer[3] as u32) << 8
+                    | (self.read_buffer[4] as u32) << 16
+                    | (self.read_buffer[5] as u32) << 24) as usize;
+                let needed = size + 6;
+                if self.read_buffer.len() < needed {
+                    continue 'read_more;
+                }
+
+                let mut packet = self.read_buffer.split_to(needed);
+                packet.advance(6);
+
+                match IdmTransportPacket::decode(packet) {
+                    Ok(packet) => {
+                        packets.push(packet);
+                    }
+                    Err(error) => {
+                        return Err(anyhow!("received invalid idm packet: {}", error));
+                    }
+                }
+
+                if self.read_buffer.is_empty() {
+                    break;
+                }
+            }
+            return Ok(packets);
         }
     }
 
@@ -403,7 +437,8 @@ impl<R: IdmRequest, E: IdmSerializable> IdmClient<R, E> {
         loop {
             select! {
                 x = backend.recv() => match x {
-                    Ok(packet) => {
+                    Ok(packets) => {
+                        for packet in packets {
                          if packet.channel != channel {
                             continue;
                         }
@@ -478,6 +513,7 @@ impl<R: IdmRequest, E: IdmSerializable> IdmClient<R, E> {
 
                             _ => {},
                         }
+                        }
                     },
 
                     Err(error) => {

crates/runtime/src/channel.rs

@@ -8,14 +8,11 @@ use anyhow::{anyhow, Result};
 use log::{debug, error};
 use tokio::{
     select,
-    sync::{
+    sync::mpsc::{channel, Receiver, Sender},
-        broadcast,
-        mpsc::{channel, Receiver, Sender},
-    },
     task::JoinHandle,
     time::sleep,
 };
-use xenevtchn::EventChannel;
+use xenevtchn::EventChannelService;
 use xengnt::{sys::GrantRef, GrantTab, MappedMemory};
 use xenstore::{XsdClient, XsdInterface};
 
@@ -43,7 +40,7 @@ pub struct ChannelService {
     typ: String,
     use_reserved_ref: Option<u64>,
     backends: HashMap<u32, ChannelBackend>,
-    evtchn: EventChannel,
+    evtchn: EventChannelService,
     store: XsdClient,
     gnttab: GrantTab,
     input_receiver: Receiver<(u32, Vec<u8>)>,
@@ -64,7 +61,7 @@ impl ChannelService {
         let (output_sender, output_receiver) = channel(GROUPED_CHANNEL_QUEUE_LEN);
 
         debug!("opening Xen event channel");
-        let evtchn = EventChannel::open().await?;
+        let evtchn = EventChannelService::open().await?;
         debug!("opening XenStore");
         let store = XsdClient::open().await?;
         debug!("opening GrantTab");
@@ -234,7 +231,7 @@ impl ChannelBackend {
         domid: u32,
         id: u32,
         store: XsdClient,
-        evtchn: EventChannel,
+        evtchn: EventChannelService,
         gnttab: GrantTab,
         output_sender: Sender<(u32, Option<Vec<u8>>)>,
         use_reserved_ref: Option<u64>,
@@ -273,7 +270,7 @@ pub struct KrataChannelBackendProcessor {
     id: u32,
     domid: u32,
     store: XsdClient,
-    evtchn: EventChannel,
+    evtchn: EventChannelService,
     gnttab: GrantTab,
 }
 
@@ -492,25 +489,18 @@ impl KrataChannelBackendProcessor {
                 },
 
                 x = channel.receiver.recv() => match x {
-                    Ok(_) => {
+                    Some(_) => {
                         unsafe {
                             let buffer = self.read_output_buffer(channel.local_port, &memory).await?;
                             if !buffer.is_empty() {
                                 sender.send((self.domid, Some(buffer))).await?;
                             }
                         };
-                        channel.unmask_sender.send(channel.local_port).await?;
+                        channel.unmask().await?;
                     },
 
-                    Err(error) => {
+                    None => {
-                        match error {
-                            broadcast::error::RecvError::Closed => {
                         break;
-                            },
-                            error => {
-                                return Err(anyhow!("failed to receive event notification: {}", error));
-                            }
-                        }
                     }
                 }
             };

crates/runtime/src/ip.rs

@@ -7,7 +7,7 @@ use std::{
 
 use anyhow::{anyhow, Result};
 use ipnetwork::{Ipv4Network, Ipv6Network};
-use log::{debug, error, trace};
+use log::{debug, error};
 use tokio::sync::RwLock;
 use uuid::Uuid;
 use xenstore::{XsdClient, XsdInterface};

crates/runtime/src/power.rs

@@ -167,7 +167,10 @@ impl PowerManagementContext {
             .set_cpufreq_gov(CpuId::All, policy)
             .await
             .unwrap_or_else(|error| {
-                info!("non-fatal error while setting scheduler policy: {:?}", error);
+                info!(
+                    "non-fatal error while setting scheduler policy: {:?}",
+                    error
+                );
             });
         Ok(())
     }

crates/xen/xenevtchn/Cargo.toml

@@ -9,6 +9,7 @@ edition = "2021"
 resolver = "2"
 
 [dependencies]
+byteorder = { workspace = true }
 libc = { workspace = true }
 log = { workspace = true }
 thiserror = { workspace = true }

crates/xen/xenevtchn/examples/simple.rs

@@ -1,9 +1,9 @@
 use xenevtchn::error::Result;
-use xenevtchn::EventChannel;
+use xenevtchn::EventChannelService;
 
 #[tokio::main]
 async fn main() -> Result<()> {
-    let channel = EventChannel::open().await?;
+    let channel = EventChannelService::open().await?;
     println!("channel opened");
     let port = channel.bind_unbound_port(0).await?;
     println!("port: {}", port);

crates/xen/xenevtchn/src/error.rs

@@ -8,6 +8,10 @@ pub enum Error {
     Io(#[from] io::Error),
     #[error("failed to send event channel wake: {0}")]
     WakeSend(tokio::sync::broadcast::error::SendError<u32>),
+    #[error("failed to acquire lock")]
+    LockAcquireFailed,
+    #[error("event port already in use")]
+    PortInUse,
 }
 
 pub type Result<T> = std::result::Result<T, Error>;

crates/xen/xenevtchn/src/lib.rs

@@ -1,82 +1,78 @@
 pub mod error;
+pub mod raw;
 pub mod sys;
 
 use crate::error::{Error, Result};
 use crate::sys::{BindInterdomain, BindUnboundPort, BindVirq, Notify, UnbindPort};
 
-use log::error;
+use crate::raw::EVENT_CHANNEL_DEVICE;
+use byteorder::{LittleEndian, ReadBytesExt};
+use log::warn;
 use std::collections::hash_map::Entry;
 use std::collections::HashMap;
 use std::mem::size_of;
 use std::os::fd::AsRawFd;
 use std::os::raw::c_void;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use tokio::fs::{File, OpenOptions};
 use tokio::io::AsyncReadExt;
-use tokio::select;
-use tokio::sync::broadcast::{
-    channel as broadcast_channel, Receiver as BroadcastReceiver, Sender as BroadastSender,
-};
 use tokio::sync::mpsc::{channel, Receiver, Sender};
-use tokio::sync::Mutex;
+use tokio::sync::{Mutex, RwLock};
-use tokio::task::JoinHandle;
 
-const UNBIND_CHANNEL_QUEUE_LEN: usize = 30;
+const CHANNEL_QUEUE_LEN: usize = 30;
-const UNMASK_CHANNEL_QUEUE_LEN: usize = 30;
-const BROADCAST_CHANNEL_QUEUE_LEN: usize = 30;
 
-type WakeMap = Arc<Mutex<HashMap<u32, BroadastSender<u32>>>>;
+type WakeMap = Arc<RwLock<HashMap<u32, Sender<u32>>>>;
 
 #[derive(Clone)]
-pub struct EventChannel {
+pub struct EventChannelService {
     handle: Arc<Mutex<File>>,
     wakes: WakeMap,
-    unbind_sender: Sender<u32>,
+    process_flag: Arc<AtomicBool>,
-    unmask_sender: Sender<u32>,
-    task: Arc<JoinHandle<()>>,
 }
 
 pub struct BoundEventChannel {
     pub local_port: u32,
-    pub receiver: BroadcastReceiver<u32>,
+    pub receiver: Receiver<u32>,
-    unbind_sender: Sender<u32>,
+    pub service: EventChannelService,
-    pub unmask_sender: Sender<u32>,
+}
+
+impl BoundEventChannel {
+    pub async fn unmask(&self) -> Result<()> {
+        self.service.unmask(self.local_port).await
+    }
 }
 
 impl Drop for BoundEventChannel {
     fn drop(&mut self) {
-        let _ = self.unbind_sender.try_send(self.local_port);
+        let service = self.service.clone();
+        let port = self.local_port;
+        tokio::task::spawn(async move {
+            let _ = service.unbind(port).await;
+        });
     }
 }
 
-impl EventChannel {
+impl EventChannelService {
-    pub async fn open() -> Result<EventChannel> {
+    pub async fn open() -> Result<EventChannelService> {
-        let file = OpenOptions::new()
+        let handle = OpenOptions::new()
             .read(true)
             .write(true)
-            .open("/dev/xen/evtchn")
+            .open(EVENT_CHANNEL_DEVICE)
             .await?;
-
+        let wakes = Arc::new(RwLock::new(HashMap::new()));
-        let wakes = Arc::new(Mutex::new(HashMap::new()));
+        let flag = Arc::new(AtomicBool::new(false));
-        let (unbind_sender, unbind_receiver) = channel(UNBIND_CHANNEL_QUEUE_LEN);
+        let processor = EventChannelProcessor {
-        let (unmask_sender, unmask_receiver) = channel(UNMASK_CHANNEL_QUEUE_LEN);
+            flag: flag.clone(),
-        let task = {
+            handle: handle.try_clone().await?.into_std().await,
-            let file = file.try_clone().await?;
+            wakes: wakes.clone(),
-            let wakes = wakes.clone();
-            tokio::task::spawn(async move {
-                if let Err(error) =
-                    EventChannel::process(file, wakes, unmask_receiver, unbind_receiver).await
-                {
-                    error!("event channel processor failed: {}", error);
-                }
-            })
         };
-        Ok(EventChannel {
+        processor.launch()?;
-            handle: Arc::new(Mutex::new(file)),
+
+        Ok(EventChannelService {
+            handle: Arc::new(Mutex::new(handle)),
             wakes,
-            unbind_sender,
+            process_flag: flag,
-            unmask_sender,
-            task: Arc::new(task),
         })
     }
 
@@ -109,11 +105,29 @@ impl EventChannel {
         }
     }
 
+    pub async fn unmask(&self, port: u32) -> Result<()> {
+        let handle = self.handle.lock().await;
+        let mut port = port;
+        let result = unsafe {
+            libc::write(
+                handle.as_raw_fd(),
+                &mut port as *mut u32 as *mut c_void,
+                size_of::<u32>(),
+            )
+        };
+        if result != size_of::<u32>() as isize {
+            return Err(Error::Io(std::io::Error::from_raw_os_error(result as i32)));
+        }
+        Ok(())
+    }
+
     pub async fn unbind(&self, port: u32) -> Result<u32> {
         let handle = self.handle.lock().await;
         unsafe {
             let mut request = UnbindPort { port };
-            Ok(sys::unbind(handle.as_raw_fd(), &mut request)? as u32)
+            let result = sys::unbind(handle.as_raw_fd(), &mut request)? as u32;
+            self.wakes.write().await.remove(&port);
+            Ok(result)
         }
     }
 
@@ -132,95 +146,65 @@ impl EventChannel {
 
     pub async fn bind(&self, domid: u32, port: u32) -> Result<BoundEventChannel> {
         let local_port = self.bind_interdomain(domid, port).await?;
-        let (receiver, unmask_sender) = self.subscribe(local_port).await?;
+        let receiver = self.subscribe(local_port).await?;
         let bound = BoundEventChannel {
             local_port,
             receiver,
-            unbind_sender: self.unbind_sender.clone(),
+            service: self.clone(),
-            unmask_sender,
         };
         Ok(bound)
     }
 
-    pub async fn subscribe(&self, port: u32) -> Result<(BroadcastReceiver<u32>, Sender<u32>)> {
+    pub async fn subscribe(&self, port: u32) -> Result<Receiver<u32>> {
-        let mut wakes = self.wakes.lock().await;
+        let mut wakes = self.wakes.write().await;
         let receiver = match wakes.entry(port) {
-            Entry::Occupied(entry) => entry.get().subscribe(),
+            Entry::Occupied(_) => {
+                return Err(Error::PortInUse);
+            }
 
             Entry::Vacant(entry) => {
-                let (sender, receiver) = broadcast_channel::<u32>(BROADCAST_CHANNEL_QUEUE_LEN);
+                let (sender, receiver) = channel::<u32>(CHANNEL_QUEUE_LEN);
                 entry.insert(sender);
                 receiver
             }
         };
-        Ok((receiver, self.unmask_sender.clone()))
+        Ok(receiver)
     }
+}
 
-    async fn process(
+pub struct EventChannelProcessor {
-        mut file: File,
+    flag: Arc<AtomicBool>,
-        wakers: WakeMap,
+    handle: std::fs::File,
-        mut unmask_receiver: Receiver<u32>,
+    wakes: WakeMap,
-        mut unbind_receiver: Receiver<u32>,
+}
-    ) -> Result<()> {
-        loop {
-            select! {
-                result = file.read_u32_le() => {
-                    match result {
-                        Ok(port) => {
-                            if let Some(sender) = wakers.lock().await.get(&port) {
-                                if let Err(error) = sender.send(port) {
-                                    return Err(Error::WakeSend(error));
-                                }
-                            }
-                        }
 
-                        Err(error) => return Err(Error::Io(error))
+impl EventChannelProcessor {
-                    }
+    pub fn launch(mut self) -> Result<()> {
-                }
+        std::thread::spawn(move || {
-
+            while let Err(error) = self.process() {
-                result = unmask_receiver.recv() => {
+                if self.flag.load(Ordering::Acquire) {
-                    match result {
-                        Some(port) => {
-                            unsafe {
-                                let mut port = port;
-                                let result = libc::write(file.as_raw_fd(), &mut port as *mut u32 as *mut c_void, size_of::<u32>());
-                                if result != size_of::<u32>() as isize {
-                                    return Err(Error::Io(std::io::Error::from_raw_os_error(result as i32)));
-                                }
-                            }
-                        }
-
-                        None => {
                     break;
                 }
+                warn!("failed to process event channel notifications: {}", error);
             }
-                }
+        });
-
-                result = unbind_receiver.recv() => {
-                    match result {
-                        Some(port) => {
-                            unsafe {
-                                let mut request = UnbindPort { port };
-                                sys::unbind(file.as_raw_fd(), &mut request)?;
-                            }
-                        }
-
-                        None => {
-                            break;
-                        }
-                    }
-                }
-            };
-        }
-
         Ok(())
     }
+
+    pub fn process(&mut self) -> Result<()> {
+        loop {
+            let port = self.handle.read_u32::<LittleEndian>()?;
+            if let Some(wake) = self.wakes.blocking_read().get(&port) {
+                let _ = wake.try_send(port);
+            }
+        }
+    }
 }
 
-impl Drop for EventChannel {
+impl Drop for EventChannelService {
     fn drop(&mut self) {
-        if Arc::strong_count(&self.task) <= 1 {
+        if Arc::strong_count(&self.handle) <= 1 {
-            self.task.abort();
+            self.process_flag.store(true, Ordering::Release);
         }
     }
 }

crates/xen/xenevtchn/src/raw.rs

@@ -0,0 +1,84 @@
+use std::fs::{File, OpenOptions};
+use std::os::fd::AsRawFd;
+use std::sync::{Arc, Mutex};
+
+use byteorder::{LittleEndian, ReadBytesExt};
+
+use crate::error::{Error, Result};
+use crate::sys;
+
+pub const EVENT_CHANNEL_DEVICE: &str = "/dev/xen/evtchn";
+
+#[derive(Clone)]
+pub struct RawEventChannelService {
+    handle: Arc<Mutex<File>>,
+}
+
+impl RawEventChannelService {
+    pub fn open() -> Result<RawEventChannelService> {
+        let handle = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .open(EVENT_CHANNEL_DEVICE)?;
+        let handle = Arc::new(Mutex::new(handle));
+        Ok(RawEventChannelService { handle })
+    }
+
+    pub fn from_handle(handle: File) -> Result<RawEventChannelService> {
+        Ok(RawEventChannelService {
+            handle: Arc::new(Mutex::new(handle)),
+        })
+    }
+
+    pub fn bind_virq(&self, virq: u32) -> Result<u32> {
+        let handle = self.handle.lock().map_err(|_| Error::LockAcquireFailed)?;
+        let mut request = sys::BindVirq { virq };
+        Ok(unsafe { sys::bind_virq(handle.as_raw_fd(), &mut request)? as u32 })
+    }
+
+    pub fn bind_interdomain(&self, domid: u32, port: u32) -> Result<u32> {
+        let handle = self.handle.lock().map_err(|_| Error::LockAcquireFailed)?;
+        let mut request = sys::BindInterdomain {
+            remote_domain: domid,
+            remote_port: port,
+        };
+        Ok(unsafe { sys::bind_interdomain(handle.as_raw_fd(), &mut request)? as u32 })
+    }
+
+    pub fn bind_unbound_port(&self, domid: u32) -> Result<u32> {
+        let handle = self.handle.lock().map_err(|_| Error::LockAcquireFailed)?;
+        let mut request = sys::BindUnboundPort {
+            remote_domain: domid,
+        };
+        Ok(unsafe { sys::bind_unbound_port(handle.as_raw_fd(), &mut request)? as u32 })
+    }
+
+    pub fn unbind(&self, port: u32) -> Result<u32> {
+        let handle = self.handle.lock().map_err(|_| Error::LockAcquireFailed)?;
+        let mut request = sys::UnbindPort { port };
+        Ok(unsafe { sys::unbind(handle.as_raw_fd(), &mut request)? as u32 })
+    }
+
+    pub fn notify(&self, port: u32) -> Result<u32> {
+        let handle = self.handle.lock().map_err(|_| Error::LockAcquireFailed)?;
+        let mut request = sys::Notify { port };
+        Ok(unsafe { sys::notify(handle.as_raw_fd(), &mut request)? as u32 })
+    }
+
+    pub fn reset(&self) -> Result<u32> {
+        let handle = self.handle.lock().map_err(|_| Error::LockAcquireFailed)?;
+        Ok(unsafe { sys::reset(handle.as_raw_fd())? as u32 })
+    }
+
+    pub fn pending(&self) -> Result<u32> {
+        let mut handle = self.handle.lock().map_err(|_| Error::LockAcquireFailed)?;
+        Ok(handle.read_u32::<LittleEndian>()?)
+    }
+
+    pub fn into_handle(self) -> Result<File> {
+        Arc::into_inner(self.handle)
+            .ok_or(Error::LockAcquireFailed)?
+            .into_inner()
+            .map_err(|_| Error::LockAcquireFailed)
+    }
+}

crates/xen/xenstore/src/bus.rs

@@ -203,8 +203,7 @@ impl XsdSocketProcessor {
         let mut header_buffer: Vec<u8> = vec![0u8; XsdMessageHeader::SIZE];
         let mut buffer: Vec<u8> = vec![0u8; XEN_BUS_MAX_PACKET_SIZE - XsdMessageHeader::SIZE];
         loop {
-            let message =
+            let message = XsdSocketProcessor::read_message(&mut header_buffer, &mut buffer, read)?;
-                XsdSocketProcessor::read_message(&mut header_buffer, &mut buffer, read)?;
             rx_sender.blocking_send(message)?;
         }
     }