Add

Author: diqiu50

clients/filesystem-fuse/Cargo.toml

@@ -42,4 +42,5 @@ libc = "0.2.168"
 log = "0.4.22"
 tokio = { version = "1.38.0", features = ["full"] }
 tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
+regex = "1.11.1"
 

clients/filesystem-fuse/src/filesystem.rs

@@ -94,39 +94,38 @@ pub(crate) trait PathFileSystem: Send + Sync {
     async fn init(&self) -> Result<()>;
 
     /// Get the file stat by file path, if the file is exist, return the file stat
-    async fn stat(&self, name: &str) -> Result<FileStat>;
+    async fn stat(&self, path: &str) -> Result<FileStat>;
 
-    /// Get the file stat by parent file path and file name, if the file is exist, return the file stat
-    async fn lookup(&self, parent: &str, name: &str) -> Result<FileStat>;
+    /// Get the file stat by file path, if the file is exist, return the file stat
+    async fn lookup(&self, path: &str) -> Result<FileStat>;
 
     /// Read the directory by file path, if the file is a valid directory, return the file stat list
-    async fn read_dir(&self, name: &str) -> Result<Vec<FileStat>>;
+    async fn read_dir(&self, path: &str) -> Result<Vec<FileStat>>;
 
     /// Open the file by file path and flags, if the file is exist, return the opened file
-    async fn open_file(&self, name: &str, flags: OpenFileFlags) -> Result<OpenedFile>;
+    async fn open_file(&self, path: &str, flags: OpenFileFlags) -> Result<OpenedFile>;
 
     /// Open the directory by file path and flags, if the file is exist, return the opened file
-    async fn open_dir(&self, name: &str, flags: OpenFileFlags) -> Result<OpenedFile>;
+    async fn open_dir(&self, path: &str, flags: OpenFileFlags) -> Result<OpenedFile>;
 
-    /// Create the file by parent file path and file name and flags, if successful, return the opened file
+    /// Create the file by file path and flags, if successful, return the opened file
     async fn create_file(
         &self,
-        parent: &str,
-        name: &str,
+        path: &str,
         flags: OpenFileFlags,
     ) -> Result<OpenedFile>;
 
-    /// Create the directory by parent file path and file name, if successful, return the file stat
-    async fn create_dir(&self, parent: &str, name: &str) -> Result<FileStat>;
+    /// Create the directory by file path , if successful, return the file stat
+    async fn create_dir(&self, path: &str) -> Result<FileStat>;
 
     /// Set the file attribute by file path and file stat
-    async fn set_attr(&self, name: &str, file_stat: &FileStat, flush: bool) -> Result<()>;
+    async fn set_attr(&self, path: &str, file_stat: &FileStat, flush: bool) -> Result<()>;
 
-    /// Remove the file by parent file path and file name
-    async fn remove_file(&self, parent: &str, name: &str) -> Result<()>;
+    /// Remove the file by file path
+    async fn remove_file(&self, path: &str) -> Result<()>;
 
-    /// Remove the directory by parent file path and file name
-    async fn remove_dir(&self, parent: &str, name: &str) -> Result<()>;
+    /// Remove the directory by file path
+    async fn remove_dir(&self, path: &str) -> Result<()>;
 }
 
 // FileSystemContext is the system environment for the fuse file system.
@@ -505,7 +504,8 @@ impl<T: PathFileSystem> RawFileSystem for DefaultRawFileSystem<T> {
 
     async fn lookup(&self, parent_file_id: u64, name: &str) -> Result<FileStat> {
         let parent_file_entry = self.get_file_entry(parent_file_id).await?;
-        let mut file_stat = self.fs.lookup(&parent_file_entry.path, name).await?;
+        let path = join_file_path(&parent_file_entry.path, name);
+        let mut file_stat = self.fs.lookup(&path).await?;
         // fill the file id to file stat
         self.resolve_file_id_to_filestat(&mut file_stat, parent_file_id)
             .await;
@@ -533,9 +533,10 @@ impl<T: PathFileSystem> RawFileSystem for DefaultRawFileSystem<T> {
 
     async fn create_file(&self, parent_file_id: u64, name: &str, flags: u32) -> Result<FileHandle> {
         let parent_file_entry = self.get_file_entry(parent_file_id).await?;
+        let path = join_file_path(&parent_file_entry.path, name);
         let mut opened_file = self
             .fs
-            .create_file(&parent_file_entry.path, name, OpenFileFlags(flags))
+            .create_file(&path, OpenFileFlags(flags))
             .await?;
 
         opened_file.set_file_id(parent_file_id, self.next_file_id());
@@ -558,7 +559,8 @@ impl<T: PathFileSystem> RawFileSystem for DefaultRawFileSystem<T> {
 
     async fn create_dir(&self, parent_file_id: u64, name: &str) -> Result<u64> {
         let parent_file_entry = self.get_file_entry(parent_file_id).await?;
-        let mut filestat = self.fs.create_dir(&parent_file_entry.path, name).await?;
+        let path = join_file_path(&parent_file_entry.path, name);
+        let mut filestat = self.fs.create_dir(&path).await?;
 
         filestat.set_file_id(parent_file_id, self.next_file_id());
 
@@ -577,7 +579,8 @@ impl<T: PathFileSystem> RawFileSystem for DefaultRawFileSystem<T> {
 
     async fn remove_file(&self, parent_file_id: u64, name: &str) -> Result<()> {
         let parent_file_entry = self.get_file_entry(parent_file_id).await?;
-        self.fs.remove_file(&parent_file_entry.path, name).await?;
+        let path = join_file_path(&parent_file_entry.path, name);
+        self.fs.remove_file(&path).await?;
 
         // remove the file from file entry manager
         {
@@ -589,7 +592,8 @@ impl<T: PathFileSystem> RawFileSystem for DefaultRawFileSystem<T> {
 
     async fn remove_dir(&self, parent_file_id: u64, name: &str) -> Result<()> {
         let parent_file_entry = self.get_file_entry(parent_file_id).await?;
-        self.fs.remove_dir(&parent_file_entry.path, name).await?;
+        let path = join_file_path(&parent_file_entry.path, name);
+        self.fs.remove_dir(&path).await?;
 
         // remove the dir from file entry manager
         {

clients/filesystem-fuse/src/fuse_api_handle.rs

@@ -34,7 +34,7 @@ use std::ffi::{OsStr, OsString};
 use std::num::NonZeroU32;
 use std::time::{Duration, SystemTime};
 
-pub(crate) struct FuseApiHandle<T: RawFileSystem> {
+pub struct FuseApiHandle<T: RawFileSystem> {
     fs: T,
     default_ttl: Duration,
     fs_context: FileSystemContext,

clients/filesystem-fuse/src/fuse_server.rs

@@ -0,0 +1,115 @@
+/*
+ * 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 fuse3::raw::{Filesystem, MountHandle, Session};
+use fuse3::{MountOptions, Result};
+use log::{error, info};
+use std::process::exit;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::{Mutex, Notify};
+use tokio::time::timeout;
+
+/// Represents a FUSE server capable of starting and stopping the FUSE filesystem.
+pub struct FuseServer {
+    // Notification for stop
+    close_notify: Arc<Notify>,
+
+    // Shared handle to manage FUSE unmounting
+    mount_handle: Arc<Mutex<Option<MountHandle>>>, // Shared handle to manage FUSE unmounting
+
+    // Mount point of the FUSE filesystem
+    mount_point: String,
+}
+
+impl FuseServer {
+    /// Creates a new instance of `FuseServer`.
+    pub fn new(mount_point: &str) -> Self {
+        Self {
+            close_notify: Arc::new(Notify::new()),
+            mount_handle: Arc::new(Mutex::new(None)),
+            mount_point: mount_point.to_string(),
+        }
+    }
+
+    /// Starts the FUSE filesystem and blocks until it is stopped.
+    pub async fn start(&self, fuse_fs: impl Filesystem + Sync + 'static) -> Result<()> {
+        //check if the mount point exists
+        if !std::path::Path::new(&self.mount_point).exists() {
+            error!("Mount point {} does not exist", self.mount_point);
+            exit(libc::ENOENT);
+        }
+
+        info!(
+            "Starting FUSE filesystem and mounting at {}",
+            self.mount_point
+        );
+
+        let mount_options = MountOptions::default();
+        let mount_handle = Session::new(mount_options)
+            .mount_with_unprivileged(fuse_fs, &self.mount_point)
+            .await?;
+
+        {
+            let mut handle_guard = self.mount_handle.lock().await;
+            *handle_guard = Some(mount_handle);
+        }
+
+        // Wait for stop notification
+        self.close_notify.notified().await;
+
+        info!("Received stop notification, FUSE filesystem will be unmounted.");
+        Ok(())
+    }
+
+    /// Stops the FUSE filesystem and waits for unmounting to complete.
+    pub async fn stop(&self) -> Result<()> {
+        // Notify stop
+        self.close_notify.notify_one();
+
+        info!("Stopping FUSE filesystem...");
+        let timeout_duration = Duration::from_secs(5);
+
+        let handle = {
+            let mut handle_guard = self.mount_handle.lock().await;
+            handle_guard.take() // Take the handle out to unmount
+        };
+
+        if let Some(mount_handle) = handle {
+            let res = timeout(timeout_duration, mount_handle.unmount()).await;
+
+            match res {
+                Ok(Ok(())) => {
+                    info!("FUSE filesystem unmounted successfully.");
+                    Ok(())
+                }
+                Ok(Err(e)) => {
+                    error!("Failed to unmount FUSE filesystem: {:?}", e);
+                    Err(e.into())
+                }
+                Err(_) => {
+                    error!("Unmount timed out.");
+                    Err(libc::ETIMEDOUT.into())
+                }
+            }
+        } else {
+            error!("No active mount handle to unmount.");
+            Err(libc::EBADF.into())
+        }
+    }
+}

clients/filesystem-fuse/src/lib.rs

@@ -20,3 +20,14 @@ mod filesystem;
 mod fuse_api_handle;
 mod opened_file_manager;
 mod utils;
+mod memory_filesystem;
+mod fuse_server;
+mod mount;
+
+pub async fn gvfs_mount() -> fuse3::Result<()> {
+    mount::mount().await
+}
+
+pub fn gvfs_unmount() {
+
+}

clients/filesystem-fuse/src/main.rs

@@ -19,64 +19,16 @@
 mod filesystem;
 mod fuse_api_handle;
 mod opened_file_manager;
+mod memory_filesystem;
+mod fuse_server;
 mod utils;
+mod mount;
 
-use log::debug;
-use log::info;
-use std::process::exit;
+use gvfs_fuse::gvfs_mount;
 
 #[tokio::main]
-async fn main() {
+async fn main() -> fuse3::Result<()> {
     tracing_subscriber::fmt().init();
-    info!("Starting filesystem...");
-    debug!("Shutdown filesystem...");
-    exit(0);
-}
-
-async fn create_gvfs_fuse_filesystem() {
-    // Gvfs-fuse filesystem structure:
-    // FuseApiHandle
-    // ├─ DefaultRawFileSystem (RawFileSystem)
-    // │ └─ FileSystemLog (PathFileSystem)
-    // │    ├─ GravitinoComposedFileSystem (PathFileSystem)
-    // │    │  ├─ GravitinoFilesetFileSystem (PathFileSystem)
-    // │    │  │  └─ S3FileSystem (PathFileSystem)
-    // │    │  │     └─ OpenDALFileSystem (PathFileSystem)
-    // │    │  ├─ GravitinoFilesetFileSystem (PathFileSystem)
-    // │    │  │  └─ HDFSFileSystem (PathFileSystem)
-    // │    │  │     └─ OpenDALFileSystem (PathFileSystem)
-    // │    │  ├─ GravitinoFilesetFileSystem (PathFileSystem)
-    // │    │  │  └─ JuiceFileSystem (PathFileSystem)
-    // │    │  │     └─ NasFileSystem (PathFileSystem)
-    // │    │  ├─ GravitinoFilesetFileSystem (PathFileSystem)
-    // │    │  │  └─ XXXFileSystem (PathFileSystem)
-    //
-    // `SimpleFileSystem` is a low-level filesystem designed to communicate with FUSE APIs.
-    // It manages file and directory relationships, as well as file mappings.
-    // It delegates file operations to the PathFileSystem
-    //
-    // `FileSystemLog` is a decorator that adds extra debug logging functionality to file system APIs.
-    // Similar implementations include permissions, caching, and metrics.
-    //
-    // `GravitinoComposeFileSystem` is a composite file system that can combine multiple `GravitinoFilesetFileSystem`.
-    // It use the part of catalog and schema of fileset path to a find actual GravitinoFilesetFileSystem. delegate the operation to the real storage.
-    // If the user only mounts a fileset, this layer is not present. There will only be one below layer.
-    //
-    // `GravitinoFilesetFileSystem` is a file system that can access a fileset.It translates the fileset path to the real storage path.
-    // and delegate the operation to the real storage.
-    //
-    // `OpenDALFileSystem` is a file system that use the OpenDAL to access real storage.
-    // it can assess the S3, HDFS, gcs, azblob and other storage.
-    //
-    // `S3FileSystem` is a file system that use `OpenDALFileSystem` to access S3 storage.
-    //
-    // `HDFSFileSystem` is a file system that use `OpenDALFileSystem` to access HDFS storage.
-    //
-    // `NasFileSystem` is a filesystem that uses a locally accessible path mounted by NAS tools, such as JuiceFS.
-    //
-    // `JuiceFileSystem` is a file that use `NasFileSystem` to access JuiceFS storage.
-    //
-    // `XXXFileSystem is a filesystem that allows you to implement file access through your own extensions.
-
-    todo!("Implement the createGvfsFuseFileSystem function");
-}
+    gvfs_mount().await?;
+    Ok(())
+}