Add close file

Author: diqiu50

clients/filesystem-fuse/src/filesystem.rs

@@ -54,6 +54,8 @@ pub trait RawFileSystem: Send + Sync {
 
     async fn open_file(&self, file_id: u64, flags: u32) -> Result<FileHandle>;
 
+    async fn open_dir(&self, file_id: u64, flags: u32) -> Result<FileHandle>;
+
     async fn create_file(&self, parent_file_id: u64, name: &str, flags: u32) -> Result<FileHandle>;
 
     async fn create_dir(&self, parent_file_id: u64, name: &str) -> Result<FileHandle>;
@@ -85,6 +87,8 @@ pub trait PathFileSystem: Send + Sync {
 
     async fn open_file(&self, name: &str, flags: OpenFileFlags) -> Result<OpenedFile>;
 
+    async fn open_dir(&self, name: &str, flags: OpenFileFlags) -> Result<OpenedFile>;
+
     async fn create_file(
         &self,
         parent: &str,
@@ -250,6 +254,24 @@ impl OpenedFile {
         self.writer.as_mut().unwrap().write(offset, data).await
     }
 
+    async fn close(&mut self) -> Result<()> {
+        if let Some(mut reader) = self.reader.take() {
+            reader.close().await?;
+        }
+        if let Some(mut writer) = self.writer.take() {
+            self.flush().await?;
+            writer.close().await?
+        }
+        Ok(())
+    }
+
+    async fn flush(&mut self) -> Result<()> {
+        if let Some(writer) = &mut self.writer {
+            writer.flush().await?;
+        }
+        Ok(())
+    }
+
     fn file_handle(&self) -> FileHandle {
         FileHandle {
             file_id: self.file_stat.inode,
@@ -266,12 +288,21 @@ impl OpenedFile {
 #[async_trait]
 pub trait FileReader: Sync + Send {
     async fn read(&mut self, offset: u64, size: u32) -> Result<Bytes>;
+    async fn close(&mut self) -> Result<()> {
+        Ok(())
+    }
 }
 
 /// File writer interface  for write file content
 #[async_trait]
 pub trait FileWriter: Sync + Send {
     async fn write(&mut self, offset: u64, data: &[u8]) -> Result<u32>;
+    async fn close(&mut self) -> Result<()> {
+        Ok(())
+    }
+    async fn flush(&mut self) -> Result<()> {
+        Ok(())
+    }
 }
 
 #[derive(Debug, Clone)]
@@ -409,6 +440,35 @@ impl<T: PathFileSystem> SimpleFileSystem<T> {
         let file = file.lock().await;
         Ok(file.file_handle())
     }
+
+    async fn open_file_internal(
+        &self,
+        file_id: u64,
+        flags: u32,
+        kind: FileType,
+    ) -> Result<FileHandle> {
+        let file_node = self.get_file_node(file_id)?;
+
+        let mut file = {
+            match kind {
+                FileType::Directory => {
+                    self.fs
+                        .open_dir(&file_node.file_name, OpenFileFlags(flags))
+                        .await?
+                }
+                FileType::RegularFile => {
+                    self.fs
+                        .open_file(&file_node.file_name, OpenFileFlags(flags))
+                        .await?
+                }
+                _ => return Err(Errno::from(libc::EINVAL)),
+            }
+        };
+        file.set_inode(file_node.parent_file_id, file_id);
+        let file = self.opened_file_manager.put_file(file);
+        let file = file.lock().await;
+        Ok(file.file_handle())
+    }
 }
 
 #[async_trait]
@@ -467,15 +527,13 @@ impl<T: PathFileSystem> RawFileSystem for SimpleFileSystem<T> {
     }
 
     async fn open_file(&self, file_id: u64, flags: u32) -> Result<FileHandle> {
-        let file_node = self.get_file_node(file_id)?;
-        let mut file = self
-            .fs
-            .open_file(&file_node.file_name, OpenFileFlags(flags))
-            .await?;
-        file.set_inode(file_node.parent_file_id, file_id);
-        let file = self.opened_file_manager.put_file(file);
-        let file = file.lock().await;
-        Ok(file.file_handle())
+        self.open_file_internal(file_id, flags, FileType::RegularFile)
+            .await
+    }
+
+    async fn open_dir(&self, file_id: u64, flags: u32) -> Result<FileHandle> {
+        self.open_file_internal(file_id, flags, FileType::Directory)
+            .await
     }
 
     async fn create_file(&self, parent_file_id: u64, name: &str, flags: u32) -> Result<FileHandle> {
@@ -522,7 +580,12 @@ impl<T: PathFileSystem> RawFileSystem for SimpleFileSystem<T> {
     }
 
     async fn close_file(&self, _file_id: u64, fh: u64) -> Result<()> {
-        self.opened_file_manager.remove_file(fh);
+        let file = self
+            .opened_file_manager
+            .remove_file(fh)
+            .ok_or(Errno::from(libc::EBADF))?;
+        let mut file = file.lock().await;
+        file.close().await?;
         Ok(())
     }
 

clients/filesystem-fuse/src/fuse_api_handle.rs

@@ -231,19 +231,21 @@ impl<T: RawFileSystem> Filesystem for FuseApiHandle<T> {
         lock_owner: u64,
         flush: bool,
     ) -> fuse3::Result<()> {
-        let _ = self.local_fs.close_file(inode, fh);
-        Ok(())
+        self.local_fs.close_file(inode, fh).await
     }
 
     async fn opendir(&self, req: Request, inode: Inode, flags: u32) -> fuse3::Result<ReplyOpen> {
-        let file_handle = self.local_fs.open_file(inode, flags).await?;
+        let file_handle = self.local_fs.open_dir(inode, flags).await?;
         Ok(ReplyOpen {
             fh: file_handle.handle_id,
             flags: flags,
         })
     }
 
-    type DirEntryStream<'a> = BoxStream<'a, fuse3::Result<DirectoryEntry>> where T: 'a;
+    type DirEntryStream<'a>
+        = BoxStream<'a, fuse3::Result<DirectoryEntry>>
+    where
+        T: 'a;
 
     async fn readdir<'a>(
         &'a self,
@@ -321,7 +323,10 @@ impl<T: RawFileSystem> Filesystem for FuseApiHandle<T> {
         })
     }
 
-    type DirEntryPlusStream<'a> = BoxStream<'a, fuse3::Result<DirectoryEntryPlus>> where T: 'a;
+    type DirEntryPlusStream<'a>
+        = BoxStream<'a, fuse3::Result<DirectoryEntryPlus>>
+    where
+        T: 'a;
 
     async fn readdirplus<'a>(
         &'a self,

clients/filesystem-fuse/src/memory_filesystem.rs

@@ -126,6 +126,10 @@ impl PathFileSystem for MemoryFileSystem {
         }
     }
 
+    async fn open_dir(&self, name: &str, flags: OpenFileFlags) -> Result<OpenedFile> {
+        self.open_file(name, flags).await
+    }
+
     async fn create_file(
         &self,
         parent: &str,

clients/filesystem-fuse/src/opened_file_manager.rs

@@ -59,7 +59,7 @@ impl OpenedFileManager {
             .map(|x| x.value().clone())
     }
 
-    pub(crate) fn remove_file(&self, handle_id: u64) {
-        self.file_handle_map.remove(&handle_id);
+    pub(crate) fn remove_file(&self, handle_id: u64) -> Option<Arc<Mutex<OpenedFile>>> {
+        self.file_handle_map.remove(&handle_id).map(|x| x.1)
     }
 }