feat(ovfs): add filesystem to handle message (#4720)

* add server

* fix conflict

* fix conflict

* typo

* fix conflict

* move to integrations

* polish code

* add comments

* typo

* typo

integrations/virtiofs/src/filesystem.rs

@@ -0,0 +1,133 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::io::Write;
+use std::mem::size_of;
+
+use vm_memory::ByteValued;
+
+use crate::error::*;
+use crate::filesystem_message::*;
+use crate::virtiofs_util::{Reader, Writer};
+
+/// Version number of this interface.
+const KERNEL_VERSION: u32 = 7;
+/// Minor version number of this interface.
+const KERNEL_MINOR_VERSION: u32 = 38;
+/// Minimum Minor version number supported.
+const MIN_KERNEL_MINOR_VERSION: u32 = 27;
+/// The length of the header part of the message.
+const BUFFER_HEADER_SIZE: u32 = 256;
+/// The maximum length of the data part of the message, used for read/write data.
+const MAX_BUFFER_SIZE: u32 = 1 << 20;
+
+/// Filesystem is a filesystem implementation with opendal backend,
+/// and will decode and process messages from VMs.
+pub struct Filesystem {}
+
+#[allow(dead_code)]
+impl Filesystem {
+    pub fn new() -> Filesystem {
+        Filesystem {}
+    }
+
+    pub fn handle_message(&self, mut r: Reader, w: Writer) -> Result<usize> {
+        let in_header: InHeader = r.read_obj().map_err(|e| {
+            new_vhost_user_fs_error("failed to decode protocol messages", Some(e.into()))
+        })?;
+        if in_header.len > (MAX_BUFFER_SIZE + BUFFER_HEADER_SIZE) {
+            // The message is too long here.
+            return Filesystem::reply_error(in_header.unique, w);
+        }
+        if let Ok(opcode) = Opcode::try_from(in_header.opcode) {
+            match opcode {
+                Opcode::Init => self.init(in_header, r, w),
+            }
+        } else {
+            Filesystem::reply_error(in_header.unique, w)
+        }
+    }
+}
+
+impl Filesystem {
+    fn reply_ok<T: ByteValued>(
+        out: Option<T>,
+        data: Option<&[u8]>,
+        unique: u64,
+        mut w: Writer,
+    ) -> Result<usize> {
+        let mut len = size_of::<OutHeader>();
+        if out.is_some() {
+            len += size_of::<T>();
+        }
+        if let Some(data) = data {
+            len += data.len();
+        }
+        let header = OutHeader {
+            unique,
+            error: 0, // Return no error.
+            len: len as u32,
+        };
+        w.write_all(header.as_slice()).map_err(|e| {
+            new_vhost_user_fs_error("failed to encode protocol messages", Some(e.into()))
+        })?;
+        if let Some(out) = out {
+            w.write_all(out.as_slice()).map_err(|e| {
+                new_vhost_user_fs_error("failed to encode protocol messages", Some(e.into()))
+            })?;
+        }
+        if let Some(data) = data {
+            w.write_all(data).map_err(|e| {
+                new_vhost_user_fs_error("failed to encode protocol messages", Some(e.into()))
+            })?;
+        }
+        Ok(w.bytes_written())
+    }
+
+    fn reply_error(unique: u64, mut w: Writer) -> Result<usize> {
+        let header = OutHeader {
+            unique,
+            error: libc::EIO, // Here we simply return I/O error.
+            len: size_of::<OutHeader>() as u32,
+        };
+        w.write_all(header.as_slice()).map_err(|e| {
+            new_vhost_user_fs_error("failed to encode protocol messages", Some(e.into()))
+        })?;
+        Ok(w.bytes_written())
+    }
+}
+
+impl Filesystem {
+    fn init(&self, in_header: InHeader, mut r: Reader, w: Writer) -> Result<usize> {
+        let InitIn { major, minor, .. } = r.read_obj().map_err(|e| {
+            new_vhost_user_fs_error("failed to decode protocol messages", Some(e.into()))
+        })?;
+
+        if major != KERNEL_VERSION || minor < MIN_KERNEL_MINOR_VERSION {
+            return Filesystem::reply_error(in_header.unique, w);
+        }
+
+        // We will directly return ok and do nothing for now.
+        let out = InitOut {
+            major: KERNEL_VERSION,
+            minor: KERNEL_MINOR_VERSION,
+            max_write: MAX_BUFFER_SIZE,
+            ..Default::default()
+        };
+        Filesystem::reply_ok(Some(out), None, in_header.unique, w)
+    }
+}

integrations/virtiofs/src/filesystem_message.rs

@@ -0,0 +1,113 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use vm_memory::ByteValued;
+
+use crate::error::*;
+
+/// Opcode represents the filesystem call that needs to be executed by VMs message.
+/// The corresponding value needs to be aligned with the specification.
+#[non_exhaustive]
+pub enum Opcode {
+    Init = 26,
+}
+
+impl TryFrom<u32> for Opcode {
+    type Error = Error;
+
+    fn try_from(value: u32) -> Result<Self, Self::Error> {
+        match value {
+            26 => Ok(Opcode::Init),
+            _ => Err(new_vhost_user_fs_error("failed to decode opcode", None)),
+        }
+    }
+}
+
+/// InHeader represents the incoming message header in the filesystem call.
+///
+/// The fields of the struct need to conform to the specific format of the virtiofs message.
+/// Currently, we only need to align them exactly with virtiofsd.
+/// Reference: https://gitlab.com/virtio-fs/virtiofsd/-/blob/main/src/fuse.rs?ref_type=heads#L1155
+#[repr(C)]
+#[derive(Debug, Default, Clone, Copy)]
+pub struct InHeader {
+    pub len: u32,
+    pub opcode: u32,
+    pub unique: u64,
+    pub nodeid: u64,
+    pub uid: u32,
+    pub gid: u32,
+    pub pid: u32,
+    pub total_extlen: u16,
+    pub padding: u16,
+}
+
+/// OutHeader represents the message header returned in the filesystem call.
+///
+/// The fields of the struct need to conform to the specific format of the virtiofs message.
+/// Currently, we only need to align them exactly with virtiofsd.
+/// Reference: https://gitlab.com/virtio-fs/virtiofsd/-/blob/main/src/fuse.rs?ref_type=heads#L1170
+#[repr(C)]
+#[derive(Debug, Default, Clone, Copy)]
+pub struct OutHeader {
+    pub len: u32,
+    pub error: i32,
+    pub unique: u64,
+}
+
+/// InitIn is used to parse the parameters passed in the Init filesystem call.
+///
+/// The fields of the struct need to conform to the specific format of the virtiofs message.
+/// Currently, we only need to align them exactly with virtiofsd.
+/// Reference: https://gitlab.com/virtio-fs/virtiofsd/-/blob/main/src/fuse.rs?ref_type=heads#L1030
+#[repr(C)]
+#[derive(Debug, Default, Clone, Copy)]
+pub struct InitIn {
+    pub major: u32,
+    pub minor: u32,
+    pub max_readahead: u32,
+    pub flags: u32,
+}
+
+/// InitOut is used to return the result of the Init filesystem call.
+///
+/// The fields of the struct need to conform to the specific format of the virtiofs message.
+/// Currently, we only need to align them exactly with virtiofsd.
+/// Reference: https://gitlab.com/virtio-fs/virtiofsd/-/blob/main/src/fuse.rs?ref_type=heads#L1048
+#[repr(C)]
+#[derive(Debug, Default, Clone, Copy)]
+pub struct InitOut {
+    pub major: u32,
+    pub minor: u32,
+    pub max_readahead: u32,
+    pub flags: u32,
+    pub max_background: u16,
+    pub congestion_threshold: u16,
+    pub max_write: u32,
+    pub time_gran: u32,
+    pub max_pages: u16,
+    pub map_alignment: u16,
+    pub flags2: u32,
+    pub unused: [u32; 7],
+}
+
+/// We will use ByteValued to implement the encoding and decoding
+/// of these structures in shared memory.
+unsafe impl ByteValued for InHeader {}
+unsafe impl ByteValued for OutHeader {}
+unsafe impl ByteValued for InitIn {}
+unsafe impl ByteValued for InitOut {}

integrations/virtiofs/src/lib.rs

@@ -16,5 +16,7 @@
 // under the License.
 
 mod error;
+mod filesystem;
+mod filesystem_message;
 mod virtiofs;
 mod virtiofs_util;