async-ify xenstore and xenclient

libs/xen/xenstore/Cargo.toml

@@ -11,6 +11,11 @@ path = "src/lib.rs"
 thiserror = { workspace = true }
 libc = { workspace = true }
 log = { workspace = true }
+tokio = { workspace = true }
+async-trait = { workspace = true }
+
+[dev-dependencies]
+futures = { workspace = true }
 
 [dependencies.bytemuck]
 workspace = true

libs/xen/xenstore/examples/list.rs

@@ -1,9 +1,10 @@
+use futures::executor::block_on;
 use xenstore::client::{XsdClient, XsdInterface};
 use xenstore::error::Result;
 use xenstore::sys::XSD_ERROR_EINVAL;
 
 fn list_recursive(client: &mut XsdClient, level: usize, path: &str) -> Result<()> {
-    let children = match client.list(path) {
+    let children = match block_on(client.list(path)) {
         Ok(children) => children,
         Err(error) => {
             return if error.to_string() == XSD_ERROR_EINVAL.error {
@@ -16,20 +17,16 @@ fn list_recursive(client: &mut XsdClient, level: usize, path: &str) -> Result<()
 
     for child in children {
         let full = format!("{}/{}", if path == "/" { "" } else { path }, child);
-        let value = client.read(full.as_str())?;
-        println!(
-            "{}{} = {:?}",
-            " ".repeat(level),
-            child,
-            String::from_utf8(value)?
-        );
+        let value = block_on(client.read_string(full.as_str()))?.expect("expected value");
+        println!("{}{} = {:?}", " ".repeat(level), child, value,);
         list_recursive(client, level + 1, full.as_str())?;
     }
     Ok(())
 }
 
-fn main() -> Result<()> {
-    let mut client = XsdClient::open()?;
+#[tokio::main]
+async fn main() -> Result<()> {
+    let mut client = XsdClient::open().await?;
     list_recursive(&mut client, 0, "/")?;
     Ok(())
 }

libs/xen/xenstore/src/bus.rs

@@ -5,7 +5,8 @@ use std::fs::{self, metadata, File};
 use std::io::{Read, Write};
 use std::mem::size_of;
 use std::os::unix::fs::FileTypeExt;
-use std::os::unix::net::UnixStream;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::UnixStream;
 
 const XEN_BUS_PATHS: &[&str] = &["/dev/xen/xenbus", "/var/run/xenstored/socket"];
 
@@ -19,18 +20,21 @@ fn find_bus_path() -> Option<String> {
     None
 }
 
+#[async_trait::async_trait]
 trait XsdTransport {
-    fn xsd_write_all(&mut self, buf: &[u8]) -> Result<()>;
-    fn xsd_read_exact(&mut self, buf: &mut [u8]) -> Result<()>;
+    async fn xsd_write_all(&mut self, buf: &[u8]) -> Result<()>;
+    async fn xsd_read_exact(&mut self, buf: &mut [u8]) -> Result<()>;
 }
 
+#[async_trait::async_trait]
 impl XsdTransport for UnixStream {
-    fn xsd_write_all(&mut self, buf: &[u8]) -> Result<()> {
-        Ok(self.write_all(buf)?)
+    async fn xsd_write_all(&mut self, buf: &[u8]) -> Result<()> {
+        Ok(self.write_all(buf).await?)
     }
 
-    fn xsd_read_exact(&mut self, buf: &mut [u8]) -> Result<()> {
-        Ok(self.read_exact(buf)?)
+    async fn xsd_read_exact(&mut self, buf: &mut [u8]) -> Result<()> {
+        self.read_exact(buf).await?;
+        Ok(())
     }
 }
 
@@ -45,12 +49,13 @@ impl XsdFileTransport {
     }
 }
 
+#[async_trait::async_trait]
 impl XsdTransport for XsdFileTransport {
-    fn xsd_read_exact(&mut self, buf: &mut [u8]) -> Result<()> {
+    async fn xsd_read_exact(&mut self, buf: &mut [u8]) -> Result<()> {
         Ok(self.handle.read_exact(buf)?)
     }
 
-    fn xsd_write_all(&mut self, buf: &[u8]) -> Result<()> {
+    async fn xsd_write_all(&mut self, buf: &[u8]) -> Result<()> {
         self.handle.write_all(buf)?;
         self.handle.flush()?;
         Ok(())
@@ -93,7 +98,7 @@ impl XsdResponse {
 }
 
 impl XsdSocket {
-    pub fn dial() -> Result<XsdSocket> {
+    pub async fn open() -> Result<XsdSocket> {
         let path = match find_bus_path() {
             Some(path) => path,
             None => return Err(Error::BusNotFound),
@@ -102,7 +107,7 @@ impl XsdSocket {
         let metadata = fs::metadata(&path)?;
         let file_type = metadata.file_type();
         if file_type.is_socket() {
-            let stream = UnixStream::connect(&path)?;
+            let stream = UnixStream::connect(&path).await?;
             return Ok(XsdSocket {
                 handle: Box::new(stream),
             });
@@ -113,7 +118,7 @@ impl XsdSocket {
         })
     }
 
-    pub fn send(&mut self, tx: u32, typ: u32, buf: &[u8]) -> Result<XsdResponse> {
+    pub async fn send(&mut self, tx: u32, typ: u32, buf: &[u8]) -> Result<XsdResponse> {
         let header = XsdMessageHeader {
             typ,
             req: 0,
@@ -124,12 +129,14 @@ impl XsdSocket {
         let mut composed: Vec<u8> = Vec::new();
         composed.extend_from_slice(header_bytes);
         composed.extend_from_slice(buf);
-        self.handle.xsd_write_all(&composed)?;
+        self.handle.xsd_write_all(&composed).await?;
         let mut result_buf = vec![0u8; size_of::<XsdMessageHeader>()];
-        self.handle.xsd_read_exact(result_buf.as_mut_slice())?;
+        self.handle
+            .xsd_read_exact(result_buf.as_mut_slice())
+            .await?;
         let result_header = bytemuck::from_bytes::<XsdMessageHeader>(&result_buf);
         let mut payload = vec![0u8; result_header.len as usize];
-        self.handle.xsd_read_exact(payload.as_mut_slice())?;
+        self.handle.xsd_read_exact(payload.as_mut_slice()).await?;
         if result_header.typ == XSD_ERROR {
             let error = CString::from_vec_with_nul(payload)?;
             return Err(Error::ResponseError(error.into_string()?));
@@ -138,18 +145,23 @@ impl XsdSocket {
         Ok(response)
     }
 
-    pub fn send_single(&mut self, tx: u32, typ: u32, string: &str) -> Result<XsdResponse> {
+    pub async fn send_single(&mut self, tx: u32, typ: u32, string: &str) -> Result<XsdResponse> {
         let text = CString::new(string)?;
         let buf = text.as_bytes_with_nul();
-        self.send(tx, typ, buf)
+        self.send(tx, typ, buf).await
     }
 
-    pub fn send_multiple(&mut self, tx: u32, typ: u32, array: &[&str]) -> Result<XsdResponse> {
+    pub async fn send_multiple(
+        &mut self,
+        tx: u32,
+        typ: u32,
+        array: &[&str],
+    ) -> Result<XsdResponse> {
         let mut buf: Vec<u8> = Vec::new();
         for item in array {
             buf.extend_from_slice(item.as_bytes());
             buf.push(0);
         }
-        self.send(tx, typ, buf.as_slice())
+        self.send(tx, typ, buf.as_slice()).await
     }
 }

libs/xen/xenstore/src/client.rs

@@ -35,85 +35,71 @@ impl XsPermission {
     }
 }
 
+#[allow(async_fn_in_trait)]
 pub trait XsdInterface {
-    fn list(&mut self, path: &str) -> Result<Vec<String>>;
-    fn read(&mut self, path: &str) -> Result<Vec<u8>>;
-    fn read_string(&mut self, path: &str) -> Result<String>;
-    fn write(&mut self, path: &str, data: Vec<u8>) -> Result<bool>;
-    fn write_string(&mut self, path: &str, data: &str) -> Result<bool>;
-    fn mkdir(&mut self, path: &str) -> Result<bool>;
-    fn rm(&mut self, path: &str) -> Result<bool>;
-    fn set_perms(&mut self, path: &str, perms: &[XsPermission]) -> Result<bool>;
+    async fn list(&mut self, path: &str) -> Result<Vec<String>>;
+    async fn read(&mut self, path: &str) -> Result<Option<Vec<u8>>>;
+    async fn read_string(&mut self, path: &str) -> Result<Option<String>>;
+    async fn write(&mut self, path: &str, data: Vec<u8>) -> Result<bool>;
+    async fn write_string(&mut self, path: &str, data: &str) -> Result<bool>;
+    async fn mkdir(&mut self, path: &str) -> Result<bool>;
+    async fn rm(&mut self, path: &str) -> Result<bool>;
+    async fn set_perms(&mut self, path: &str, perms: &[XsPermission]) -> Result<bool>;
 
-    fn mknod(&mut self, path: &str, perms: &[XsPermission]) -> Result<bool> {
-        let result1 = self.write_string(path, "")?;
-        let result2 = self.set_perms(path, perms)?;
+    async fn mknod(&mut self, path: &str, perms: &[XsPermission]) -> Result<bool> {
+        let result1 = self.write_string(path, "").await?;
+        let result2 = self.set_perms(path, perms).await?;
         Ok(result1 && result2)
     }
-
-    fn read_string_optional(&mut self, path: &str) -> Result<Option<String>> {
-        Ok(match self.read_string(path) {
-            Ok(value) => Some(value),
-            Err(error) => {
-                if error.is_noent_response() {
-                    None
-                } else {
-                    return Err(error);
-                }
-            }
-        })
-    }
-
-    fn list_any(&mut self, path: &str) -> Result<Vec<String>> {
-        Ok(match self.list(path) {
-            Ok(value) => value,
-            Err(error) => {
-                if error.is_noent_response() {
-                    Vec::new()
-                } else {
-                    return Err(error);
-                }
-            }
-        })
-    }
 }
 
 impl XsdClient {
-    pub fn open() -> Result<XsdClient> {
-        let socket = XsdSocket::dial()?;
+    pub async fn open() -> Result<XsdClient> {
+        let socket = XsdSocket::open().await?;
         Ok(XsdClient { socket })
     }
 
-    fn list(&mut self, tx: u32, path: &str) -> Result<Vec<String>> {
+    async fn list(&mut self, tx: u32, path: &str) -> Result<Vec<String>> {
         trace!("list tx={tx} path={path}");
-        let response = self.socket.send_single(tx, XSD_DIRECTORY, path)?;
+        let response = self.socket.send_single(tx, XSD_DIRECTORY, path).await?;
         response.parse_string_vec()
     }
 
-    fn read(&mut self, tx: u32, path: &str) -> Result<Vec<u8>> {
+    async fn read(&mut self, tx: u32, path: &str) -> Result<Option<Vec<u8>>> {
         trace!("read tx={tx} path={path}");
-        let response = self.socket.send_single(tx, XSD_READ, path)?;
-        Ok(response.payload)
+        match self.socket.send_single(tx, XSD_READ, path).await {
+            Ok(response) => Ok(Some(response.payload)),
+            Err(error) => {
+                if error.is_noent_response() {
+                    Ok(None)
+                } else {
+                    Err(error)
+                }
+            }
+        }
     }
 
-    fn write(&mut self, tx: u32, path: &str, data: Vec<u8>) -> Result<bool> {
+    async fn write(&mut self, tx: u32, path: &str, data: Vec<u8>) -> Result<bool> {
         trace!("write tx={tx} path={path} data={:?}", data);
         let mut buffer = Vec::new();
         let path = CString::new(path)?;
         buffer.extend_from_slice(path.as_bytes_with_nul());
         buffer.extend_from_slice(data.as_slice());
-        let response = self.socket.send(tx, XSD_WRITE, buffer.as_slice())?;
+        let response = self.socket.send(tx, XSD_WRITE, buffer.as_slice()).await?;
         response.parse_bool()
     }
 
-    fn mkdir(&mut self, tx: u32, path: &str) -> Result<bool> {
+    async fn mkdir(&mut self, tx: u32, path: &str) -> Result<bool> {
         trace!("mkdir tx={tx} path={path}");
-        self.socket.send_single(tx, XSD_MKDIR, path)?.parse_bool()
+        self.socket
+            .send_single(tx, XSD_MKDIR, path)
+            .await?
+            .parse_bool()
     }
 
-    fn rm(&mut self, tx: u32, path: &str) -> Result<bool> {
+    async fn rm(&mut self, tx: u32, path: &str) -> Result<bool> {
         trace!("rm tx={tx} path={path}");
-        let result = self.socket.send_single(tx, XSD_RM, path);
+        let result = self.socket.send_single(tx, XSD_RM, path).await;
         if let Err(error) = result {
             if error.is_noent_response() {
                 return Ok(true);
@@ -123,7 +109,7 @@ impl XsdClient {
         result.unwrap().parse_bool()
     }
 
-    fn set_perms(&mut self, tx: u32, path: &str, perms: &[XsPermission]) -> Result<bool> {
+    async fn set_perms(&mut self, tx: u32, path: &str, perms: &[XsPermission]) -> Result<bool> {
         trace!("set_perms tx={tx} path={path} perms={:?}", perms);
         let mut items: Vec<String> = Vec::new();
         items.push(path.to_string());
@@ -131,36 +117,46 @@ impl XsdClient {
             items.push(perm.encode()?);
         }
         let items_str: Vec<&str> = items.iter().map(|x| x.as_str()).collect();
-        let response = self.socket.send_multiple(tx, XSD_SET_PERMS, &items_str)?;
+        let response = self
+            .socket
+            .send_multiple(tx, XSD_SET_PERMS, &items_str)
+            .await?;
         response.parse_bool()
     }
 
-    pub fn transaction(&mut self) -> Result<XsdTransaction> {
+    pub async fn transaction(&mut self) -> Result<XsdTransaction> {
         trace!("transaction start");
-        let response = self.socket.send_single(0, XSD_TRANSACTION_START, "")?;
+        let response = self
+            .socket
+            .send_single(0, XSD_TRANSACTION_START, "")
+            .await?;
         let str = response.parse_string()?;
         let tx = str.parse::<u32>()?;
         Ok(XsdTransaction { client: self, tx })
     }
 
-    pub fn get_domain_path(&mut self, domid: u32) -> Result<String> {
-        let response =
-            self.socket
-                .send_single(0, XSD_GET_DOMAIN_PATH, domid.to_string().as_str())?;
+    pub async fn get_domain_path(&mut self, domid: u32) -> Result<String> {
+        let response = self
+            .socket
+            .send_single(0, XSD_GET_DOMAIN_PATH, domid.to_string().as_str())
+            .await?;
         response.parse_string()
     }
 
-    pub fn introduce_domain(&mut self, domid: u32, mfn: u64, evtchn: u32) -> Result<bool> {
+    pub async fn introduce_domain(&mut self, domid: u32, mfn: u64, evtchn: u32) -> Result<bool> {
         trace!("introduce domain domid={domid} mfn={mfn} evtchn={evtchn}");
-        let response = self.socket.send_multiple(
-            0,
-            XSD_INTRODUCE,
-            &[
-                domid.to_string().as_str(),
-                mfn.to_string().as_str(),
-                evtchn.to_string().as_str(),
-            ],
-        )?;
+        let response = self
+            .socket
+            .send_multiple(
+                0,
+                XSD_INTRODUCE,
+                &[
+                    domid.to_string().as_str(),
+                    mfn.to_string().as_str(),
+                    evtchn.to_string().as_str(),
+                ],
+            )
+            .await?;
         response.parse_bool()
     }
 }
@@ -171,89 +167,104 @@ pub struct XsdTransaction<'a> {
 }
 
 impl XsdInterface for XsdClient {
-    fn list(&mut self, path: &str) -> Result<Vec<String>> {
-        self.list(0, path)
+    async fn list(&mut self, path: &str) -> Result<Vec<String>> {
+        self.list(0, path).await
     }
 
-    fn read(&mut self, path: &str) -> Result<Vec<u8>> {
-        self.read(0, path)
+    async fn read(&mut self, path: &str) -> Result<Option<Vec<u8>>> {
+        self.read(0, path).await
     }
 
-    fn read_string(&mut self, path: &str) -> Result<String> {
-        Ok(String::from_utf8(self.read(0, path)?)?)
+    async fn read_string(&mut self, path: &str) -> Result<Option<String>> {
+        match self.read(0, path).await {
+            Ok(value) => match value {
+                Some(value) => Ok(Some(String::from_utf8(value)?)),
+                None => Ok(None),
+            },
+            Err(error) => Err(error),
+        }
     }
 
-    fn write(&mut self, path: &str, data: Vec<u8>) -> Result<bool> {
-        self.write(0, path, data)
+    async fn write(&mut self, path: &str, data: Vec<u8>) -> Result<bool> {
+        self.write(0, path, data).await
     }
 
-    fn write_string(&mut self, path: &str, data: &str) -> Result<bool> {
-        self.write(0, path, data.as_bytes().to_vec())
+    async fn write_string(&mut self, path: &str, data: &str) -> Result<bool> {
+        self.write(0, path, data.as_bytes().to_vec()).await
     }
 
-    fn mkdir(&mut self, path: &str) -> Result<bool> {
-        self.mkdir(0, path)
+    async fn mkdir(&mut self, path: &str) -> Result<bool> {
+        self.mkdir(0, path).await
     }
 
-    fn rm(&mut self, path: &str) -> Result<bool> {
-        self.rm(0, path)
+    async fn rm(&mut self, path: &str) -> Result<bool> {
+        self.rm(0, path).await
     }
 
-    fn set_perms(&mut self, path: &str, perms: &[XsPermission]) -> Result<bool> {
-        self.set_perms(0, path, perms)
+    async fn set_perms(&mut self, path: &str, perms: &[XsPermission]) -> Result<bool> {
+        self.set_perms(0, path, perms).await
     }
 }
 
 impl XsdInterface for XsdTransaction<'_> {
-    fn list(&mut self, path: &str) -> Result<Vec<String>> {
-        self.client.list(self.tx, path)
+    async fn list(&mut self, path: &str) -> Result<Vec<String>> {
+        self.client.list(self.tx, path).await
     }
 
-    fn read(&mut self, path: &str) -> Result<Vec<u8>> {
-        self.client.read(self.tx, path)
+    async fn read(&mut self, path: &str) -> Result<Option<Vec<u8>>> {
+        self.client.read(self.tx, path).await
     }
 
-    fn read_string(&mut self, path: &str) -> Result<String> {
-        Ok(String::from_utf8(self.client.read(self.tx, path)?)?)
+    async fn read_string(&mut self, path: &str) -> Result<Option<String>> {
+        match self.client.read(self.tx, path).await {
+            Ok(value) => match value {
+                Some(value) => Ok(Some(String::from_utf8(value)?)),
+                None => Ok(None),
+            },
+            Err(error) => Err(error),
+        }
     }
 
-    fn write(&mut self, path: &str, data: Vec<u8>) -> Result<bool> {
-        self.client.write(self.tx, path, data)
+    async fn write(&mut self, path: &str, data: Vec<u8>) -> Result<bool> {
+        self.client.write(self.tx, path, data).await
     }
 
-    fn write_string(&mut self, path: &str, data: &str) -> Result<bool> {
-        self.client.write(self.tx, path, data.as_bytes().to_vec())
+    async fn write_string(&mut self, path: &str, data: &str) -> Result<bool> {
+        self.client
+            .write(self.tx, path, data.as_bytes().to_vec())
+            .await
     }
 
-    fn mkdir(&mut self, path: &str) -> Result<bool> {
-        self.client.mkdir(self.tx, path)
+    async fn mkdir(&mut self, path: &str) -> Result<bool> {
+        self.client.mkdir(self.tx, path).await
     }
 
-    fn rm(&mut self, path: &str) -> Result<bool> {
-        self.client.rm(self.tx, path)
+    async fn rm(&mut self, path: &str) -> Result<bool> {
+        self.client.rm(self.tx, path).await
     }
 
-    fn set_perms(&mut self, path: &str, perms: &[XsPermission]) -> Result<bool> {
-        self.client.set_perms(self.tx, path, perms)
+    async fn set_perms(&mut self, path: &str, perms: &[XsPermission]) -> Result<bool> {
+        self.client.set_perms(self.tx, path, perms).await
     }
 }
 
 impl XsdTransaction<'_> {
-    pub fn end(&mut self, abort: bool) -> Result<bool> {
+    pub async fn end(&mut self, abort: bool) -> Result<bool> {
         let abort_str = if abort { "F" } else { "T" };
 
         trace!("transaction end abort={}", abort);
         self.client
             .socket
-            .send_single(self.tx, XSD_TRANSACTION_END, abort_str)?
+            .send_single(self.tx, XSD_TRANSACTION_END, abort_str)
+            .await?
             .parse_bool()
     }
 
-    pub fn commit(&mut self) -> Result<bool> {
-        self.end(false)
+    pub async fn commit(&mut self) -> Result<bool> {
+        self.end(false).await
     }
 
-    pub fn abort(&mut self) -> Result<bool> {
-        self.end(true)
+    pub async fn abort(&mut self) -> Result<bool> {
+        self.end(true).await
     }
 }