From db838c4eb5f966af4f522350f0779a5d41601202 Mon Sep 17 00:00:00 2001 From: yyjeqhc <1772413353@qq.com> Date: Sun, 18 May 2025 16:57:42 +0800 Subject: [PATCH 1/3] add updates the dir's hash in the db and use dashmap replace hashmap. --- scorpio/Cargo.toml | 1 + scorpio/src/dicfuse/store.rs | 50 ++++++++++++++++++------------- scorpio/src/dicfuse/tree_store.rs | 12 ++++++++ scorpio/src/util/config.rs | 11 +++++++ 4 files changed, 54 insertions(+), 20 deletions(-) diff --git a/scorpio/Cargo.toml b/scorpio/Cargo.toml index 56c9e4703..2956b8ce2 100644 --- a/scorpio/Cargo.toml +++ b/scorpio/Cargo.toml @@ -49,6 +49,7 @@ fjall = "2.8.0" thiserror = "2.0.12" crossbeam = "0.8.4" fs_extra = "1.2" +dashmap = "6.1.0" [features] async-io = [] diff --git a/scorpio/src/dicfuse/store.rs b/scorpio/src/dicfuse/store.rs index 050027ba7..dddf4abe9 100644 --- a/scorpio/src/dicfuse/store.rs +++ b/scorpio/src/dicfuse/store.rs @@ -1,4 +1,5 @@ use async_recursion::async_recursion; +use dashmap::DashMap; /// Read only file system for obtaining and displaying monorepo directory information use core::panic; use crossbeam::queue::SegQueue; @@ -266,7 +267,8 @@ pub struct DirItem { } pub struct DictionaryStore { inodes: Arc>>>, - dirs: Arc>>, //save all the dirs. + // dirs: Arc>>, //save all the dirs. + dirs: Arc>, // save all the dirs. next_inode: AtomicU64, radix_trie: Arc>>, persistent_path_store: Arc, // persistent path store for saving and retrieving file paths @@ -281,7 +283,7 @@ impl DictionaryStore { inodes: Arc::new(Mutex::new(HashMap::new())), radix_trie: Arc::new(Mutex::new(radix_trie::Trie::new())), persistent_path_store: Arc::new(tree_store), - dirs: Arc::new(Mutex::new(HashMap::new())), + dirs: Arc::new(DashMap::new()), } } @@ -337,7 +339,7 @@ impl DictionaryStore { #[async_recursion] async fn load_dirs(&self, path: PathBuf, parent_inode: u64) -> Result<(), io::Error> { let root_item = self.persistent_path_store.get_item(parent_inode)?; - self.dirs.lock().await.insert( + self.dirs.insert( path.to_string_lossy().to_string(), DirItem { hash: root_item.hash.to_owned(), @@ -350,8 +352,6 @@ impl DictionaryStore { let child_item = self.persistent_path_store.get_item(child)?; let child_path = path.join(child_item.get_name()); self.dirs - .lock() - .await .get_mut(&path.to_string_lossy().to_string()) .unwrap() .file_list @@ -457,7 +457,7 @@ impl DictionaryStore { let item = self.get_inode(parent).await?; // current_dictionary let mut parent_path = self.find_path(parent).await.unwrap(); parent_path.pop(); - + let parent_item = self.get_by_path(&parent_path.to_string()).await?; let mut re = vec![item.clone(), parent_item.clone()]; @@ -487,7 +487,7 @@ pub async fn import_arc(store: Arc) { if store.load_db().await.is_ok() { tokio::spawn(async move { loop { - tokio::time::sleep(tokio::time::Duration::from_secs(60)).await; + tokio::time::sleep(tokio::time::Duration::from_secs(20)).await; watch_dir(store.clone()).await; } }); @@ -520,8 +520,6 @@ pub async fn import_arc(store: Arc) { store.inodes.lock().await.insert(1, root_item.into()); store .dirs - .lock() - .await .insert("/".to_string(), root_dir_item); } // use the unlock queue instead of mpsc Mutex @@ -531,7 +529,7 @@ pub async fn import_arc(store: Arc) { let items = fetch_dir("").await.unwrap().data; let active_producers = Arc::new(AtomicUsize::new(items.len())); { - let mut locks = store.dirs.lock().await; + let locks = store.dirs.clone(); // let dir_item = locks.get_mut("/").unwrap(); for it in items { let is_dir = it.item.is_dir(); @@ -591,8 +589,6 @@ pub async fn import_arc(store: Arc) { let tmp_path = newit.item.path.to_owned(); store .dirs - .lock() - .await .get_mut(&path) .unwrap() .file_list @@ -603,7 +599,7 @@ pub async fn import_arc(store: Arc) { // If it's a directory, push it to the queue and add the producer count producers.fetch_add(1, Ordering::Relaxed); queue.push(new_inode); - store.dirs.lock().await.insert( + store.dirs.insert( tmp_path, DirItem { hash: newit.hash, @@ -638,12 +634,16 @@ pub async fn import_arc(store: Arc) { join_all(workers).await; tokio::spawn(async move { loop { - tokio::time::sleep(tokio::time::Duration::from_secs(60)).await; + tokio::time::sleep(tokio::time::Duration::from_secs(20)).await; watch_dir(store.clone()).await; } }); } +/// Load directories of a specified depth under the parent folder. +pub async fn load_dir_depth(_store: Arc,_parent: &str) { + +} /// Watch the directory and update the dictionary pub async fn watch_dir(store: Arc) { //TODO: rename a file or dic @@ -654,7 +654,7 @@ pub async fn watch_dir(store: Arc) { let items = fetch_dir("").await.unwrap().data; { - let mut locks = store.dirs.lock().await; + let locks = store.dirs.clone(); for it in items { let is_dir = it.item.is_dir(); let path = it.item.path.to_owned(); @@ -670,8 +670,12 @@ pub async fn watch_dir(store: Arc) { //when the dir's hash changed,fetch the dir. if locks.get_mut(&path).unwrap().hash != it.hash { // If the path already exists, update the hash - let dir_it = locks.get_mut(&path).unwrap(); - dir_it.hash = it.hash; + let mut dir_it = locks.get_mut(&path).unwrap(); + dir_it.hash = it.hash.to_owned(); + + //also update the hash in the db. + let inode = store.get_inode_from_path(&path).await.unwrap(); + tree_db.update_item_hash(inode, it.hash).unwrap(); queue.push(path); } } @@ -745,7 +749,7 @@ pub async fn watch_dir(store: Arc) { Ok(new_items) => { let new_items = new_items.data; println!("{:?}", new_items.len()); - let mut locks = store.dirs.lock().await; + let locks = store.dirs.clone(); for newit in new_items { let is_dir = newit.item.is_dir(); let tmp_path = newit.item.path.to_owned(); @@ -765,9 +769,15 @@ pub async fn watch_dir(store: Arc) { && locks.get_mut(&tmp_path).unwrap().hash != newit.hash { // If the path already exists, update the hash - let dir_it = locks.get_mut(&tmp_path).unwrap(); - dir_it.hash = newit.hash; + let mut dir_it = locks.get_mut(&tmp_path).unwrap(); + dir_it.hash = newit.hash.to_owned(); + //also update the hash in the db. + let inode = + store.get_inode_from_path(&tmp_path).await.unwrap(); + tree_db.update_item_hash(inode, newit.hash).unwrap(); + queue.push(tmp_path); + producers.fetch_add(1, Ordering::Relaxed); } } else { diff --git a/scorpio/src/dicfuse/tree_store.rs b/scorpio/src/dicfuse/tree_store.rs index 5e8a26e0f..aa7961f89 100644 --- a/scorpio/src/dicfuse/tree_store.rs +++ b/scorpio/src/dicfuse/tree_store.rs @@ -178,6 +178,18 @@ impl TreeStorage { names.reverse(); // reverse the names to get the all path of this item. Ok(GPath { path: names }) } + /// when the dir's hash changes,we need to update the hash value in the db. + pub fn update_item_hash(&self, inode: u64, hash: String) -> io::Result<()> { + let mut item = self.get_storage_item(inode)?; + item.hash = hash; + self.db + .insert( + inode.to_be_bytes(), + bincode::serialize(&item).map_err(Error::other)?, + ) + .map_err(Error::other)?; + Ok(()) + } } #[cfg(test)] mod tests { diff --git a/scorpio/src/util/config.rs b/scorpio/src/util/config.rs index 64a5df7d1..2a6abc63b 100644 --- a/scorpio/src/util/config.rs +++ b/scorpio/src/util/config.rs @@ -147,6 +147,7 @@ fn get_config() -> &'static ScorpioConfig { config.insert("git_author".to_string(), "MEGA".to_string()); config.insert("git_email".to_string(), "admin@mega.org".to_string()); config.insert("config_file".to_string(), "config.toml".to_string()); + config.insert("load_dir_depth".to_string(), "3".to_string()); config.insert( "lfs_url".to_string(), "http://localhost:8000/lfs".to_string(), @@ -191,6 +192,7 @@ fn validate(config: &mut HashMap) -> ConfigResult<()> { "config_file", "lfs_url", "dicfuse_readable", + "load_dir_depth", ]; for key in required_keys { @@ -240,6 +242,15 @@ pub fn lfs_url() -> &'static str { pub fn dicfuse_readable() -> bool { get_config().config["dicfuse_readable"] == "true" } + +///get the depth of directory loading +pub fn load_dir_depth() -> usize { + get_config() + .config + .get("load_dir_depth") + .and_then(|s| s.parse().ok()) + .unwrap_or(3) +} #[cfg(test)] mod tests { use super::*; From c5527c2be3cbb6eb268c77a0b65925026c20667b Mon Sep 17 00:00:00 2001 From: yyjeqhc <1772413353@qq.com> Date: Mon, 26 May 2025 21:59:29 +0800 Subject: [PATCH 2/3] add preload the dir and config the depth to load. --- ceres/src/api_service/mod.rs | 59 ++- mono/src/api/api_router.rs | 34 ++ scorpio/scorpio.toml | 7 +- scorpio/src/dicfuse/async_io.rs | 14 +- scorpio/src/dicfuse/store.rs | 625 ++++++++++++++++++------------ scorpio/src/dicfuse/tree_store.rs | 4 + scorpio/src/util/config.rs | 2 +- 7 files changed, 482 insertions(+), 263 deletions(-) diff --git a/ceres/src/api_service/mod.rs b/ceres/src/api_service/mod.rs index 3ba3d606b..82c38205b 100644 --- a/ceres/src/api_service/mod.rs +++ b/ceres/src/api_service/mod.rs @@ -189,7 +189,7 @@ pub trait ApiHandler: Send + Sync { let commit = if let Some(commit) = commit_map.get(commit_id) { commit.to_owned() } else { - tracing::warn!("failed fecth from commit map: {}", commit_id); + tracing::warn!("failed fetch from commit map: {}", commit_id); if root_commit.is_none() { root_commit = Some(self.get_root_commit().await); } @@ -216,6 +216,63 @@ pub trait ApiHandler: Send + Sync { } } + /// return the dir's hash only + async fn get_tree_dir_hash( + &self, + path: PathBuf, + dir_name: &str, + ) -> Result, GitError> { + let mut cache = GitObjectCache::default(); + match self.search_tree_by_path(&path).await? { + Some(tree) => { + let mut item_to_commit = HashMap::new(); + + self.add_trees_to_map( + &mut item_to_commit, + tree.tree_items + .iter() + .filter(|x| x.mode == TreeItemMode::Tree && x.name == dir_name) + .map(|x| x.id.to_string()) + .collect(), + ) + .await; + + let commit_ids: HashSet = item_to_commit.values().cloned().collect(); + let commits = self + .get_commits_by_hashes(commit_ids.into_iter().collect()) + .await + .unwrap(); + let commit_map: HashMap = + commits.into_iter().map(|x| (x.id.to_string(), x)).collect(); + + let mut root_commit: Option = None; + let mut item_to_commit_map: HashMap> = HashMap::new(); + for item in tree.tree_items { + if let Some(commit_id) = item_to_commit.get(&item.id.to_string()) { + let commit = if let Some(commit) = commit_map.get(commit_id) { + commit.to_owned() + } else { + tracing::warn!("failed fetch from commit map: {}", commit_id); + if root_commit.is_none() { + root_commit = Some(self.get_root_commit().await); + } + let root_commit = root_commit.as_ref().unwrap().clone(); + self.traverse_commit_history(&path, &root_commit, &item, &mut cache) + .await + }; + item_to_commit_map.insert(item, Some(commit)); + } + } + let items: Vec = item_to_commit_map + .into_iter() + .map(TreeCommitItem::from) + .collect(); + Ok(items) + } + None => Ok(Vec::new()), + } + } + fn convert_commit_to_info(&self, commit: Commit) -> Result { let message = commit.format_message(); let committer = UserInfo { diff --git a/mono/src/api/api_router.rs b/mono/src/api/api_router.rs index 0836d8158..9a36edc3d 100644 --- a/mono/src/api/api_router.rs +++ b/mono/src/api/api_router.rs @@ -31,6 +31,7 @@ pub fn routers() -> OpenApiRouter { .routes(routes!(create_file)) .routes(routes!(get_latest_commit)) .routes(routes!(get_tree_commit_info)) + .routes(routes!(get_tree_dir_hash)) .routes(routes!(path_can_be_cloned)) .routes(routes!(get_tree_info)) .routes(routes!(get_blob_string)) @@ -172,6 +173,39 @@ async fn get_tree_commit_info( Ok(Json(CommonResult::success(Some(data)))) } +/// return the dir's hash +#[utoipa::path( + get, + path = "/tree/dir-hash", + params( + CodePreviewQuery + ), + responses( + (status = 200, body = CommonResult>, content_type = "application/json") + ), + tag = GIT_TAG +)] +async fn get_tree_dir_hash( + Query(query): Query, + state: State, +) -> Result>>, ApiError> { + let path = std::path::Path::new(&query.path); + let parent_path = path + .parent() + .and_then(|p| p.to_str()) + .unwrap_or("") + .to_string(); + let target_name = path.file_name().and_then(|n| n.to_str()).unwrap_or(""); + + let data = state + .api_handler(parent_path.clone().into()) + .await? + .get_tree_dir_hash(parent_path.into(), target_name) + .await?; + + Ok(Json(CommonResult::success(Some(data)))) +} + pub async fn get_blob_file( state: State, Path(oid): Path, diff --git a/scorpio/scorpio.toml b/scorpio/scorpio.toml index 8ab5bb95d..0b9ffd8d7 100644 --- a/scorpio/scorpio.toml +++ b/scorpio/scorpio.toml @@ -1,8 +1,9 @@ -lfs_url = "http://47.79.35.136:8000" +lfs_url = "http://localhost:8000" store_path = "/home/luxian/megadir/store" config_file = "config.toml" git_author = "MEGA" git_email = "admin@mega.org" workspace = "/home/luxian/megadir/mount" -base_url = "http://47.79.35.136:8000" -dicfuse_readable = "true" \ No newline at end of file +base_url = "http://localhost:8000" +dicfuse_readable = "true" +load_dir_depth = "3" \ No newline at end of file diff --git a/scorpio/src/dicfuse/async_io.rs b/scorpio/src/dicfuse/async_io.rs index 29de4285d..ce3c83cc5 100644 --- a/scorpio/src/dicfuse/async_io.rs +++ b/scorpio/src/dicfuse/async_io.rs @@ -6,11 +6,12 @@ use fuse3::raw::prelude::*; use fuse3::raw::reply::DirectoryEntry; use fuse3::{Errno, Inode, Result}; +use super::Dicfuse; +use crate::dicfuse::store::load_dir; +use crate::util::config; use futures::stream::{iter, Iter}; use std::vec::IntoIter; -use super::Dicfuse; - impl Filesystem for Dicfuse { /// dir entry stream given by [`readdir`][Filesystem::readdir]. type DirEntryStream<'a> @@ -32,6 +33,7 @@ impl Filesystem for Dicfuse { .ok_or_else(|| std::io::Error::from_raw_os_error(libc::ENODATA))?; ppath.push(name.to_string_lossy().into_owned()); + println!("learn lookup {:?}", ppath.clone().to_string()); let chil = store.get_by_path(&ppath.to_string()).await?; if self.open_buff.read().await.get(&chil.get_inode()).is_none() { let _ = self.load_one_file(parent, name).await; @@ -136,6 +138,12 @@ impl Filesystem for Dicfuse { } async fn access(&self, _req: Request, inode: Inode, _mask: u32) -> Result<()> { self.store.get_inode(inode).await?; + let load_parent = "/".to_string() + &self.store.find_path(inode).await.unwrap().to_string(); + let max_depth = config::load_dir_depth() + load_parent.matches('/').count(); + let hash_change = load_dir(self.store.clone(), load_parent, max_depth).await; + if hash_change { + self.store.update_ancestors_hash(inode).await; + } Ok(()) } @@ -213,7 +221,7 @@ impl Filesystem for Dicfuse { })); } } - println!("{:?}", d); + // println!("{:?}", d); Ok(ReplyDirectoryPlus { entries: iter(d.into_iter()), }) diff --git a/scorpio/src/dicfuse/store.rs b/scorpio/src/dicfuse/store.rs index dddf4abe9..0fef28db5 100644 --- a/scorpio/src/dicfuse/store.rs +++ b/scorpio/src/dicfuse/store.rs @@ -1,8 +1,9 @@ +use crate::READONLY_INODE; use async_recursion::async_recursion; -use dashmap::DashMap; /// Read only file system for obtaining and displaying monorepo directory information use core::panic; use crossbeam::queue::SegQueue; +use dashmap::DashMap; use fuse3::raw::reply::ReplyEntry; use fuse3::FileType; use futures::future::join_all; @@ -16,8 +17,6 @@ use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::sync::Arc; use tokio::sync::Mutex; -use crate::READONLY_INODE; - use super::abi::{default_dic_entry, default_file_entry}; use super::tree_store::{StorageItem, TreeStorage}; use crate::util::{config, GPath}; @@ -42,6 +41,23 @@ pub struct ItemExt { pub item: Item, pub hash: String, } + +#[derive(Serialize, Deserialize, Debug, Default)] + struct CommitInfoResponse { + req_result: bool, + data: Vec, + err_message: String, + } + + #[derive(Serialize, Deserialize, Debug, Default)] + struct CommitInfo { + oid: String, + name: String, + content_type: String, + message: String, + date: String, + } + #[allow(unused)] pub struct DicItem { inode: u64, @@ -194,22 +210,76 @@ async fn fetch_dir(path: &str) -> Result { } }; - #[derive(Serialize, Deserialize, Debug, Default)] - struct CommitInfoResponse { - req_result: bool, - data: Vec, - err_message: String, + let commit_info: CommitInfoResponse = match response.json().await { + Ok(info) => info, + Err(e) => { + return Err(DictionaryError { + message: format!("Failed to parse commit info: {}", e), + }); + } + }; + + if !commit_info.req_result { + return Err(DictionaryError { + message: commit_info.err_message, + }); } - #[derive(Serialize, Deserialize, Debug, Default)] - struct CommitInfo { - oid: String, - name: String, - content_type: String, - message: String, - date: String, + let mut data = Vec::with_capacity(commit_info.data.len()); + + let base_path = if path.is_empty() || path == "/" { + "".to_string() + } else if path.ends_with('/') { + path.to_string() + } else { + format!("{}/", path) + }; + + for info in commit_info.data { + let full_path = if base_path.is_empty() { + format!("/{}", info.name) + } else { + format!("/{}{}", base_path.trim_start_matches('/'), info.name) + }; + + data.push(ItemExt { + item: Item { + name: info.name, + path: full_path, + content_type: info.content_type, + }, + hash: info.oid, + }); } + Ok(ApiResponseExt { + _req_result: true, + data, + _err_message: String::new(), + }) +} + +///get the directory hash from the server +async fn fetch_get_dir_hash(path: &str) -> Result { + static CLIENT: Lazy = Lazy::new(Client::new); + let client = CLIENT.clone(); + + let clean_path = path.trim_start_matches('/'); + let url = format!( + "{}/api/v1/tree/dir-hash?path=/{}", + config::base_url(), + clean_path + ); + + let response = match client.get(&url).send().await { + Ok(resp) => resp, + Err(_) => { + return Err(DictionaryError { + message: "Failed to fetch tree".to_string(), + }); + } + }; + let commit_info: CommitInfoResponse = match response.json().await { Ok(info) => info, Err(e) => { @@ -258,6 +328,7 @@ async fn fetch_dir(path: &str) -> Result { _err_message: String::new(), }) } + /// Represents a directory with its metadata /// - hash: represents the hash of the last commit that modified this directory /// - file_list: represents the list of files and subdirectories in this directory, with boolean values indicating if they still exist @@ -336,6 +407,29 @@ impl DictionaryStore { ) .await } + + pub async fn update_ancestors_hash(&self, inode: u64) { + let item = self.persistent_path_store.get_item(inode).unwrap(); + let mut parent_inode = item.get_parent(); + while parent_inode != 1 { + let path = "/".to_string() + + &self + .persistent_path_store + .get_all_path(parent_inode) + .unwrap() + .to_string(); + println!("update hash {:?}", path); + let hash = get_dir_hash(&path).await; + if hash.is_empty() { + return; + } + self.persistent_path_store + .update_item_hash(parent_inode, hash.to_owned()) + .unwrap(); + self.dirs.get_mut(&path).unwrap().hash = hash; + parent_inode = item.get_parent(); + } + } #[async_recursion] async fn load_dirs(&self, path: PathBuf, parent_inode: u64) -> Result<(), io::Error> { let root_item = self.persistent_path_store.get_item(parent_inode)?; @@ -372,6 +466,7 @@ impl DictionaryStore { self.load_dirs(path, 1).await?; Ok(()) } + pub async fn import(&self) { let items = fetch_dir("").await.unwrap().data; @@ -457,7 +552,7 @@ impl DictionaryStore { let item = self.get_inode(parent).await?; // current_dictionary let mut parent_path = self.find_path(parent).await.unwrap(); parent_path.pop(); - + let parent_item = self.get_by_path(&parent_path.to_string()).await?; let mut re = vec![item.clone(), parent_item.clone()]; @@ -482,65 +577,27 @@ impl DictionaryStore { } } -pub async fn import_arc(store: Arc) { - //first load the db. - if store.load_db().await.is_ok() { - tokio::spawn(async move { - loop { - tokio::time::sleep(tokio::time::Duration::from_secs(20)).await; - watch_dir(store.clone()).await; - } - }); - return; - } else { - //if the db is null,then init the store and load from mono. - let _ = store.persistent_path_store.insert_item( - 1, - UNKNOW_INODE, - ItemExt { - item: Item { - name: "".to_string(), - path: "/".to_string(), - content_type: INODE_DICTIONARY.to_string(), - }, - hash: String::new(), - }, - ); - let root_item = DicItem { - inode: 1, - path_name: GPath::new(), - content_type: Arc::new(Mutex::new(ContentType::Directory(false))), - children: Mutex::new(HashMap::new()), - parent: UNKNOW_INODE, // root dictory has no parent - }; - let root_dir_item = DirItem { - hash: String::new(), - file_list: HashMap::new(), - }; - store.inodes.lock().await.insert(1, root_item.into()); - store - .dirs - .insert("/".to_string(), root_dir_item); - } - // use the unlock queue instead of mpsc Mutex +/// This function loads subdirectories of a specified directory up to a given depth. +pub async fn load_dir_depth(store: Arc, parent_path: String, max_depth: usize) { + println!("load_dir_depth {:?}", parent_path); let queue = Arc::new(SegQueue::new()); - - // init root path - let items = fetch_dir("").await.unwrap().data; - let active_producers = Arc::new(AtomicUsize::new(items.len())); + let items = fetch_dir(&parent_path).await.unwrap().data; + // only count the directories. + let dir_count = items.iter().filter(|it| it.item.is_dir()).count(); + let active_producers = Arc::new(AtomicUsize::new(dir_count)); + // let active_producers = Arc::new(AtomicUsize::new(items.len())); { let locks = store.dirs.clone(); - // let dir_item = locks.get_mut("/").unwrap(); for it in items { let is_dir = it.item.is_dir(); let path = it.item.path.to_owned(); - // dir_item.file_list.insert(path.to_owned()); locks - .get_mut("/") + .get_mut(&parent_path) .unwrap() .file_list .insert(path.to_owned(), false); - let it_inode = store.update_inode(1, it.clone()).await.unwrap(); + let parent_node = store.get_inode_from_path(&parent_path).await.unwrap(); + let it_inode = store.update_inode(parent_node, it.clone()).await.unwrap(); if is_dir { queue.push(it_inode); locks.insert( @@ -574,7 +631,6 @@ pub async fn import_arc(store: Arc) { producers.load(Ordering::Acquire) > 0 || !queue.is_empty() } { if let Some(inode) = queue.pop() { - // get path from path store. //get the whole path. let path = "/".to_string() + &path_store.get_all_path(inode).unwrap().to_string(); @@ -583,7 +639,6 @@ pub async fn import_arc(store: Arc) { match fetch_dir(&path).await { Ok(new_items) => { let new_items = new_items.data; - for newit in new_items { let is_dir = newit.item.is_dir(); let tmp_path = newit.item.path.to_owned(); @@ -597,8 +652,12 @@ pub async fn import_arc(store: Arc) { store.update_inode(inode, newit.clone()).await.unwrap(); if is_dir { // If it's a directory, push it to the queue and add the producer count - producers.fetch_add(1, Ordering::Relaxed); - queue.push(new_inode); + if tmp_path.matches('/').count() < max_depth { + producers.fetch_add(1, Ordering::Relaxed); + queue.push(new_inode); + } else { + println!("max_depth reach path = {:?}", tmp_path); + } store.dirs.insert( tmp_path, DirItem { @@ -632,219 +691,275 @@ pub async fn import_arc(store: Arc) { // worker.await.expect("Worker panicked"); // } join_all(workers).await; +} +pub async fn import_arc(store: Arc) { + //first load the db. + if store.load_db().await.is_ok() { + tokio::spawn(async move { + loop { + tokio::time::sleep(tokio::time::Duration::from_secs(60)).await; + watch_dir(store.clone()).await; + } + }); + return; + } else { + //if the db is null,then init the store and load from mono. + let _ = store.persistent_path_store.insert_item( + 1, + UNKNOW_INODE, + ItemExt { + item: Item { + name: "".to_string(), + path: "/".to_string(), + content_type: INODE_DICTIONARY.to_string(), + }, + hash: String::new(), + }, + ); + let root_item = DicItem { + inode: 1, + path_name: GPath::new(), + content_type: Arc::new(Mutex::new(ContentType::Directory(false))), + children: Mutex::new(HashMap::new()), + parent: UNKNOW_INODE, // root dictory has no parent + }; + let root_dir_item = DirItem { + hash: String::new(), + file_list: HashMap::new(), + }; + store.inodes.lock().await.insert(1, root_item.into()); + store.dirs.insert("/".to_string(), root_dir_item); + } + + let max_depth = config::load_dir_depth() + 2; + load_dir_depth(store.clone(), "/".to_string(), max_depth).await; + + // use the unlock queue instead of mpsc Mutex tokio::spawn(async move { loop { - tokio::time::sleep(tokio::time::Duration::from_secs(20)).await; + tokio::time::sleep(tokio::time::Duration::from_secs(60)).await; watch_dir(store.clone()).await; } }); } -/// Load directories of a specified depth under the parent folder. -pub async fn load_dir_depth(_store: Arc,_parent: &str) { - +///get the directory hash from the server +async fn get_dir_hash(path: &str) -> String { + let data = fetch_get_dir_hash(path).await.unwrap().data; + // no need to filter by name, just return the first item.the server ensure the name is unique. + // data.retain(|item| item.item.name == target_name && item.item.is_dir()); + if data.len() == 1 { + data[0].hash.to_owned() + } else { + String::new() + } } -/// Watch the directory and update the dictionary -pub async fn watch_dir(store: Arc) { - //TODO: rename a file or dic - // use the unlock queue instead of mpsc Mutex - //save the dir to be updated. - let queue = Arc::new(SegQueue::new()); - let tree_db = store.persistent_path_store.clone(); +#[async_recursion] +/// This function loads subdirectories of a specified depth when a user accesses a given path. +pub async fn load_dir(store: Arc, parent_path: String, max_depth: usize) -> bool { + if parent_path.matches('/').count() >= max_depth { + println!("max depth reached for path: {}", parent_path); + return false; + } + if max_depth < config::load_dir_depth() + 2 { + println!("max depth is less than config, skipping: {}", parent_path); + return false; + } - let items = fetch_dir("").await.unwrap().data; - { - let locks = store.dirs.clone(); - for it in items { - let is_dir = it.item.is_dir(); - let path = it.item.path.to_owned(); - //old files care about the update. - if locks.get_mut("/").unwrap().file_list.contains_key(&path) { - //set true means this file is still in the dir. - locks - .get_mut("/") - .unwrap() - .file_list - .insert(path.to_owned(), true); - if is_dir { - //when the dir's hash changed,fetch the dir. - if locks.get_mut(&path).unwrap().hash != it.hash { - // If the path already exists, update the hash - let mut dir_it = locks.get_mut(&path).unwrap(); - dir_it.hash = it.hash.to_owned(); - - //also update the hash in the db. - let inode = store.get_inode_from_path(&path).await.unwrap(); - tree_db.update_item_hash(inode, it.hash).unwrap(); - queue.push(path); - } - } - } else { - locks - .get_mut("/") - .unwrap() - .file_list - .insert(path.to_owned(), true); - let _ = store.update_inode(1, it.clone()).await.unwrap(); - //fetch a new dir. - if is_dir { - queue.push(path.to_owned()); - locks.insert( - path, - DirItem { - hash: it.hash, - file_list: HashMap::new(), - }, - ); - } + let parent_inode = store.get_inode_from_path(&parent_path).await.unwrap(); + + let tree_db = store.persistent_path_store.clone(); + let dirs = store.dirs.clone(); + let self_hash = get_dir_hash(&parent_path).await; + + //the dir may be deleted. + if self_hash.is_empty() { + println!("Directory {} is empty, no items to load.", parent_path); + return true; + } + println!("load_dir parent_path {:?}", parent_path); + + //empty dir,load the dir to the max_depth. + if dirs.get(&parent_path).unwrap().file_list.is_empty() { + load_dir_depth(store.clone(), parent_path.to_owned(), max_depth).await; + + if dirs.get(&parent_path).unwrap().hash != self_hash { + dirs.get_mut(&parent_path).unwrap().hash = self_hash.to_owned(); + let inode = store.get_inode_from_path(&parent_path).await.unwrap(); + tree_db.update_item_hash(inode, self_hash).unwrap(); + return true; + } + return false; + } + // if the dir's hash is same as the parent dir's hash, + //then check the subdir from the db,no need to get from the server.. + if dirs.get(&parent_path).unwrap().hash == self_hash { + let item = store.persistent_path_store.get_item(parent_inode).unwrap(); + let children = item.get_children(); + for child in children { + let child_item = store.persistent_path_store.get_item(child).unwrap(); + if child_item.is_dir() { + println!( + "handle dir /{:?}", + tree_db.get_all_path(child).unwrap().to_string() + ); + load_dir( + store.clone(), + "/".to_string() + &tree_db.get_all_path(child).unwrap().to_string(), + max_depth, + ) + .await; } } - - let mut remove_items = Vec::new(); - locks.get_mut("/").unwrap().file_list.retain(|path, v| { + return false; + } + //last, if the dir's hash is different from the parent dir's hash, + //then fetch the dir from the server. + let items = fetch_dir(&parent_path).await.unwrap().data; + dirs.get_mut(&parent_path).unwrap().hash = self_hash.to_owned(); + let inode = store.get_inode_from_path(&parent_path).await.unwrap(); + tree_db.update_item_hash(inode, self_hash).unwrap(); + for it in items { + let is_dir = it.item.is_dir(); + let path = it.item.path.to_owned(); + + // the item already exists in the parent directory. + if dirs + .get(&parent_path) + .unwrap() + .file_list + .contains_key(&path) + { + dirs.get_mut(&parent_path) + .unwrap() + .file_list + .insert(path.to_owned(), true); + if is_dir { + println!("hash changes dir {:?}", path); + load_dir(store.clone(), path.to_owned(), max_depth).await; + } + } else { + dirs.get_mut(&parent_path) + .unwrap() + .file_list + .insert(path.to_owned(), true); + let _ = store.update_inode(parent_inode, it.clone()).await.unwrap(); + //fetch a new dir. + if is_dir { + println!("add dir {:?}", path); + dirs.insert( + path.to_owned(), + DirItem { + hash: it.hash, + file_list: HashMap::new(), + }, + ); + load_dir_depth(store.clone(), path.to_owned(), max_depth).await; + } + } + } + let mut remove_items = Vec::new(); + dirs.get_mut(&parent_path) + .unwrap() + .file_list + .retain(|path, v| { let result = *v; if !(*v) { remove_items.push(path.clone()); - //delete storageItem - // let inode = store.get_inode_from_path(&path).await.unwrap(); - // tree_db.remove_item(inode); } else { *v = false; } result }); - for item in remove_items { - let inode = store.get_inode_from_path(&item).await.unwrap(); - println!("delete {:?} {} ", inode, item); - tree_db.remove_item(inode).unwrap(); - } + for item in remove_items { + let inode = store.get_inode_from_path(&item).await.unwrap(); + println!("delete {:?} {} ", inode, item); + tree_db.remove_item(inode).unwrap(); } + return true; +} - let worker_count = 10; - let mut workers = Vec::with_capacity(worker_count); - - // clone shared resource. - let queue = Arc::clone(&queue); - let active_producers = Arc::new(AtomicUsize::new(queue.len())); - - // Init mulity work thraed - for _ in 0..worker_count { - let queue = Arc::clone(&queue); - // let path_store = persistent_path_store.clone(); - let store = store.clone(); - let producers = Arc::clone(&active_producers); - let tree_db = store.persistent_path_store.clone(); - workers.push(tokio::spawn(async move { - while { - // If there are active producers or the queue is not empty, continue - producers.load(Ordering::Acquire) > 0 || !queue.is_empty() - } { - if let Some(parent) = queue.pop() { - // get path from path store. - //get the whole path. - let parent_inode = store.get_inode_from_path(&parent).await.unwrap(); - println!("Worker processing path: {} {}", parent, parent_inode); - // get all children inode - match fetch_dir(&parent).await { - Ok(new_items) => { - let new_items = new_items.data; - println!("{:?}", new_items.len()); - let locks = store.dirs.clone(); - for newit in new_items { - let is_dir = newit.item.is_dir(); - let tmp_path = newit.item.path.to_owned(); - if locks - .get_mut(&parent) - .unwrap() - .file_list - .contains_key(&tmp_path) - { - locks - .get_mut(&parent) - .unwrap() - .file_list - .insert(tmp_path.to_owned(), true); - //should care the file'hash? - if is_dir - && locks.get_mut(&tmp_path).unwrap().hash != newit.hash - { - // If the path already exists, update the hash - let mut dir_it = locks.get_mut(&tmp_path).unwrap(); - dir_it.hash = newit.hash.to_owned(); - //also update the hash in the db. - let inode = - store.get_inode_from_path(&tmp_path).await.unwrap(); - tree_db.update_item_hash(inode, newit.hash).unwrap(); - - queue.push(tmp_path); - - producers.fetch_add(1, Ordering::Relaxed); - } - } else { - locks - .get_mut(&parent) - .unwrap() - .file_list - .insert(tmp_path.to_owned(), true); - let _ = store - .update_inode(parent_inode, newit.clone()) - .await - .unwrap(); - //insert new dir - if is_dir { - producers.fetch_add(1, Ordering::Relaxed); - queue.push(tmp_path.to_owned()); - locks.insert( - tmp_path, - DirItem { - hash: newit.hash, - file_list: HashMap::new(), - }, - ); - } - } - } +#[async_recursion] +///this function is only used to update the directory which has been loaded. +/// it will update the directory but do not load the new directory. +pub async fn update_dir(store: Arc, parent_path: String) { + let tree_db = store.persistent_path_store.clone(); + let items = fetch_dir(&parent_path).await.unwrap().data; + let dirs = store.dirs.clone(); + + for it in items { + let is_dir = it.item.is_dir(); + let path = it.item.path.to_owned(); + + // the item already exists in the parent directory. + if dirs + .get(&parent_path) + .unwrap() + .file_list + .contains_key(&path) + { + dirs.get_mut(&parent_path) + .unwrap() + .file_list + .insert(path.to_owned(), true); + if is_dir && dirs.get(&path).unwrap().hash != it.hash { + //when the dir's hash changed,fetch the dir. + // If the path already exists, update the hash + update_dir(store.clone(), path.to_owned()).await; + + let mut dir_it = dirs.get_mut(&path).unwrap(); + dir_it.hash = it.hash.to_owned(); + //also update the hash in the db. + let inode = store.get_inode_from_path(&path).await.unwrap(); + tree_db.update_item_hash(inode, it.hash).unwrap(); + println!("modify dir {:?}", path); + } + } else { + dirs.get_mut(&parent_path) + .unwrap() + .file_list + .insert(path.to_owned(), true); + let parent_inode = store.get_inode_from_path(&parent_path).await.unwrap(); - let mut remove_items = Vec::new(); - locks.get_mut(&parent).unwrap().file_list.retain(|path, v| { - let result = *v; - if !(*v) { - remove_items.push(path.clone()); - // let inode = store.get_inode_from_path(&path).await.unwrap(); - // tree_db.remove_item(inode); - } else { - *v = false; - } - result - }); - for item in remove_items { - let inode = store.get_inode_from_path(&item).await.unwrap(); - println!("delete {:?} {}", inode, item); + let _ = store.update_inode(parent_inode, it.clone()).await.unwrap(); + //fetch a new dir. + if is_dir { + println!("add dir {:?}", path); - tree_db.remove_item(inode).unwrap(); - } - } - Err(_) => { - // Continue to the next iteration if there was an error - } - }; + dirs.insert( + path, + DirItem { + hash: it.hash, + file_list: HashMap::new(), + }, + ); + } + } + } - producers.fetch_sub(1, Ordering::Release); - } else { - // If there are no active producers and the queue is empty, exit the loop - if producers.load(Ordering::Acquire) == 0 { - return; - } - // yield to wait unfinished tasks - tokio::task::yield_now().await; - } + let mut remove_items = Vec::new(); + dirs.get_mut(&parent_path) + .unwrap() + .file_list + .retain(|path, v| { + let result = *v; + if !(*v) { + remove_items.push(path.clone()); + } else { + *v = false; } - })); + result + }); + for item in remove_items { + let inode = store.get_inode_from_path(&item).await.unwrap(); + println!("delete {:?} {} ", inode, item); + tree_db.remove_item(inode).unwrap(); } +} - // wait for all workers to complete - join_all(workers).await; - println!("finish"); +/// Watch the directory and update the dictionary has loaded. +pub async fn watch_dir(store: Arc) { + update_dir(store, "/".to_string()).await; } #[cfg(test)] diff --git a/scorpio/src/dicfuse/tree_store.rs b/scorpio/src/dicfuse/tree_store.rs index aa7961f89..bbecc2eb1 100644 --- a/scorpio/src/dicfuse/tree_store.rs +++ b/scorpio/src/dicfuse/tree_store.rs @@ -51,6 +51,10 @@ impl StorageItem { pub fn get_name(&self) -> String { self.name.clone() } + + pub fn get_parent(&self) -> u64 { + self.parent + } } use toml::Value; #[allow(unused)] diff --git a/scorpio/src/util/config.rs b/scorpio/src/util/config.rs index 2a6abc63b..0a4fd6711 100644 --- a/scorpio/src/util/config.rs +++ b/scorpio/src/util/config.rs @@ -243,7 +243,7 @@ pub fn dicfuse_readable() -> bool { get_config().config["dicfuse_readable"] == "true" } -///get the depth of directory loading +///Get the depth of directory loading pub fn load_dir_depth() -> usize { get_config() .config From 87e1fad751294d136589917de22a28fdad3e29cb Mon Sep 17 00:00:00 2001 From: yyjeqhc <1772413353@qq.com> Date: Tue, 27 May 2025 18:11:32 +0800 Subject: [PATCH 3/3] add test for scorpio and add fix some clippy. --- Cargo.toml | 3 +- scorpio/Cargo.toml | 13 +- scorpio/README.md | 9 +- scorpio/src/dicfuse/async_io.rs | 12 +- scorpio/src/dicfuse/mod.rs | 2 +- scorpio/src/dicfuse/store.rs | 173 +++++++-- scorpio/src/dicfuse/tree_store.rs | 6 +- scorpio/src/lib.rs | 2 +- scorpio/src/util/config.rs | 11 +- scorpio/tests/dir_test.rs | 570 ++++++++++++++++++++++++++++++ 10 files changed, 757 insertions(+), 44 deletions(-) create mode 100644 scorpio/tests/dir_test.rs diff --git a/Cargo.toml b/Cargo.toml index a98618b21..57e5fd75b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,8 +16,9 @@ members = [ "orion-server", "third-party", "observatory", + "scorpio", ] -default-members = ["mega", "mono", "libra", "aries", "orion", "orion-server"] +default-members = ["mega", "mono", "libra", "aries", "orion", "orion-server","scorpio"] resolver = "1" [workspace.dependencies] diff --git a/scorpio/Cargo.toml b/scorpio/Cargo.toml index 2956b8ce2..171c8436e 100644 --- a/scorpio/Cargo.toml +++ b/scorpio/Cargo.toml @@ -51,11 +51,20 @@ crossbeam = "0.8.4" fs_extra = "1.2" dashmap = "6.1.0" + + [features] async-io = [] -[workspace] - +[dev-dependencies] +tempfile = { workspace = true } +serial_test = { workspace = true } +lazy_static = { workspace = true } +assert_cmd = { workspace = true } +scopeguard = { workspace = true } +testcontainers = { workspace = true, features = ["http_wait","reusable-containers"] } +reqwest = { version = "0.12.12", features = ["blocking"] } +http = { workspace = true } [package.metadata.docs.rs] all-features = true diff --git a/scorpio/README.md b/scorpio/README.md index 1cb929233..9b7b29833 100644 --- a/scorpio/README.md +++ b/scorpio/README.md @@ -78,6 +78,8 @@ workspace = "/home/luxian/megadir/mount" config_file = "config.toml" git_author = "MEGA" git_email = "admin@mega.org" +dicfuse_readable = "true" +load_dir_depth = "3" ``` ### `scorpio.toml` Configuration Guide: @@ -97,8 +99,13 @@ git_email = "admin@mega.org" Extended configuration filename (default: `config.toml`). - **`git_author`** / **`git_email`** - Default Git author metadata (for version tracking). + Default Git author metadata (for version tracking). +- **`dicfuse_readable`** + Allow reading file contents from a read-only directory. + +- **`load_dir_depth`** + Specifies how deep the file system should load and preload directories during initialization. ### How to Contribute? diff --git a/scorpio/src/dicfuse/async_io.rs b/scorpio/src/dicfuse/async_io.rs index ce3c83cc5..75dcdb95b 100644 --- a/scorpio/src/dicfuse/async_io.rs +++ b/scorpio/src/dicfuse/async_io.rs @@ -8,7 +8,6 @@ use fuse3::{Errno, Inode, Result}; use super::Dicfuse; use crate::dicfuse::store::load_dir; -use crate::util::config; use futures::stream::{iter, Iter}; use std::vec::IntoIter; @@ -33,7 +32,6 @@ impl Filesystem for Dicfuse { .ok_or_else(|| std::io::Error::from_raw_os_error(libc::ENODATA))?; ppath.push(name.to_string_lossy().into_owned()); - println!("learn lookup {:?}", ppath.clone().to_string()); let chil = store.get_by_path(&ppath.to_string()).await?; if self.open_buff.read().await.get(&chil.get_inode()).is_none() { let _ = self.load_one_file(parent, name).await; @@ -138,8 +136,14 @@ impl Filesystem for Dicfuse { } async fn access(&self, _req: Request, inode: Inode, _mask: u32) -> Result<()> { self.store.get_inode(inode).await?; - let load_parent = "/".to_string() + &self.store.find_path(inode).await.unwrap().to_string(); - let max_depth = config::load_dir_depth() + load_parent.matches('/').count(); + let load_parent = "/".to_string() + + &self + .store + .find_path(inode) + .await + .ok_or(libc::ENOENT)? + .to_string(); + let max_depth = self.store.max_depth() + load_parent.matches('/').count(); let hash_change = load_dir(self.store.clone(), load_parent, max_depth).await; if hash_change { self.store.update_ancestors_hash(inode).await; diff --git a/scorpio/src/dicfuse/mod.rs b/scorpio/src/dicfuse/mod.rs index 9cfac1e74..7606da542 100644 --- a/scorpio/src/dicfuse/mod.rs +++ b/scorpio/src/dicfuse/mod.rs @@ -1,6 +1,6 @@ mod abi; mod async_io; -mod store; +pub mod store; mod tree_store; use crate::manager::fetch::fetch_tree; use crate::util::config; diff --git a/scorpio/src/dicfuse/store.rs b/scorpio/src/dicfuse/store.rs index 0fef28db5..0f544d796 100644 --- a/scorpio/src/dicfuse/store.rs +++ b/scorpio/src/dicfuse/store.rs @@ -10,12 +10,13 @@ use futures::future::join_all; use once_cell::sync::Lazy; use reqwest::Client; use serde::{Deserialize, Serialize}; -use std::collections::{HashMap, VecDeque}; +use std::collections::{BTreeSet, HashMap, VecDeque}; use std::io; use std::path::PathBuf; use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::sync::Arc; use tokio::sync::Mutex; +use tokio::sync::Notify; use super::abi::{default_dic_entry, default_file_entry}; use super::tree_store::{StorageItem, TreeStorage}; @@ -43,20 +44,20 @@ pub struct ItemExt { } #[derive(Serialize, Deserialize, Debug, Default)] - struct CommitInfoResponse { - req_result: bool, - data: Vec, - err_message: String, - } +struct CommitInfoResponse { + req_result: bool, + data: Vec, + err_message: String, +} - #[derive(Serialize, Deserialize, Debug, Default)] - struct CommitInfo { - oid: String, - name: String, - content_type: String, - message: String, - date: String, - } +#[derive(Serialize, Deserialize, Debug, Default)] +struct CommitInfo { + oid: String, + name: String, + content_type: String, + message: String, + date: String, +} #[allow(unused)] pub struct DicItem { @@ -144,7 +145,6 @@ impl Iterator for ApiResponse { self.data.pop() } } - struct ApiResponseExt { _req_result: bool, data: Vec, @@ -259,7 +259,7 @@ async fn fetch_dir(path: &str) -> Result { }) } -///get the directory hash from the server +/// Get the directory hash from the server async fn fetch_get_dir_hash(path: &str) -> Result { static CLIENT: Lazy = Lazy::new(Client::new); let client = CLIENT.clone(); @@ -343,6 +343,8 @@ pub struct DictionaryStore { next_inode: AtomicU64, radix_trie: Arc>>, persistent_path_store: Arc, // persistent path store for saving and retrieving file paths + max_depth: Arc, // max depth for loading directories + init_notify: Arc, // used in dir_test to notify the start of the test.. } #[allow(unused)] @@ -355,9 +357,20 @@ impl DictionaryStore { radix_trie: Arc::new(Mutex::new(radix_trie::Trie::new())), persistent_path_store: Arc::new(tree_store), dirs: Arc::new(DashMap::new()), + max_depth: Arc::new(config::load_dir_depth()), + init_notify: Arc::new(Notify::new()), } } + #[inline(always)] + pub fn max_depth(&self) -> usize { + *self.max_depth + } + + pub async fn wait_for_ready(&self) { + // Wait for the store to be initialized + self.init_notify.notified().await; + } async fn update_inode(&self, parent: u64, item: ItemExt) -> std::io::Result { let alloc_inode = self .next_inode @@ -385,6 +398,66 @@ impl DictionaryStore { Ok(alloc_inode) } + #[async_recursion] + async fn traverse_directory( + &self, + current_path: &str, + base_path: &str, + depth_items: &mut HashMap>, + ) -> Result<(), io::Error> { + let current_inode = match self.get_inode_from_path(current_path).await { + Ok(inode) => inode, + Err(_) => return Ok(()), + }; + + let current_item = self.persistent_path_store.get_item(current_inode)?; + + if !current_item.is_dir() { + return Ok(()); + } + + let children = current_item.get_children(); + + for child_inode in children { + let child_item = self.persistent_path_store.get_item(child_inode)?; + let child_name = child_item.get_name(); + + let child_full_path = if current_path == "/" { + format!("/{}", child_name) + } else { + format!("{}/{}", current_path, child_name) + }; + + let relative_path = if base_path == "/" { + child_full_path + .strip_prefix('/') + .unwrap_or(&child_full_path) + .to_string() + } else if child_full_path == base_path { + ".".to_string() + } else if child_full_path.starts_with(&format!("{}/", base_path)) { + child_full_path[base_path.len() + 1..].to_string() + } else { + continue; + }; + + let depth = if relative_path == "." { + 0 + } else { + relative_path.chars().filter(|&c| c == '/').count() as i32 + }; + + depth_items.entry(depth).or_default().insert(relative_path); + + if child_item.is_dir() { + self.traverse_directory(&child_full_path, base_path, depth_items) + .await?; + } + } + + Ok(()) + } + pub async fn add_temp_point(&self, path: &str) -> Result { let item_path = path.to_string(); let mut path = GPath::from(path.to_string()); @@ -408,6 +481,37 @@ impl DictionaryStore { .await } + /// Recursively traverses and returns all files and directories under the specified base directory, grouped by depth relative to base_dir + /// Returns HashMap where depth 0 = direct children, depth 1 = grandchildren, etc. + pub async fn get_dir_by_path(&self, base_dir: &str) -> HashMap> { + let mut depth_items: HashMap> = HashMap::new(); + + let normalized_base_dir = if base_dir.is_empty() || base_dir == "." { + "/".to_string() + } else if !base_dir.starts_with('/') { + format!("/{}", base_dir) + } else { + base_dir.to_string() + }; + + if (self.get_inode_from_path(&normalized_base_dir).await).is_ok() { + if let Err(e) = self + .traverse_directory(&normalized_base_dir, &normalized_base_dir, &mut depth_items) + .await + { + println!("Error traversing directory {}: {}", normalized_base_dir, e); + } + + if normalized_base_dir != "/" { + depth_items.entry(0).or_default(); + } + } else { + println!("Base directory {} not found", normalized_base_dir); + } + + depth_items + } + pub async fn update_ancestors_hash(&self, inode: u64) { let item = self.persistent_path_store.get_item(inode).unwrap(); let mut parent_inode = item.get_parent(); @@ -577,7 +681,11 @@ impl DictionaryStore { } } -/// This function loads subdirectories of a specified directory up to a given depth. +/// Loads subdirectories from a remote server into an empty parent directory up to a specified depth. +/// +/// # Arguments +/// * `parent_path` - The path to an empty directory where subdirectories will be loaded. +/// * `max_depth` - The maximum absolute depth of subdirectories to load, relative to the root. pub async fn load_dir_depth(store: Arc, parent_path: String, max_depth: usize) { println!("load_dir_depth {:?}", parent_path); let queue = Arc::new(SegQueue::new()); @@ -692,9 +800,11 @@ pub async fn load_dir_depth(store: Arc, parent_path: String, ma // } join_all(workers).await; } + pub async fn import_arc(store: Arc) { //first load the db. if store.load_db().await.is_ok() { + store.init_notify.notify_waiters(); tokio::spawn(async move { loop { tokio::time::sleep(tokio::time::Duration::from_secs(60)).await; @@ -731,9 +841,9 @@ pub async fn import_arc(store: Arc) { store.dirs.insert("/".to_string(), root_dir_item); } - let max_depth = config::load_dir_depth() + 2; + let max_depth = store.max_depth() + 2; load_dir_depth(store.clone(), "/".to_string(), max_depth).await; - + store.init_notify.notify_waiters(); // use the unlock queue instead of mpsc Mutex tokio::spawn(async move { loop { @@ -743,25 +853,34 @@ pub async fn import_arc(store: Arc) { }); } -///get the directory hash from the server +/// Get the directory hash from the server async fn get_dir_hash(path: &str) -> String { let data = fetch_get_dir_hash(path).await.unwrap().data; // no need to filter by name, just return the first item.the server ensure the name is unique. - // data.retain(|item| item.item.name == target_name && item.item.is_dir()); if data.len() == 1 { data[0].hash.to_owned() } else { String::new() } } + #[async_recursion] -/// This function loads subdirectories of a specified depth when a user accesses a given path. +/// Preloads a directory and its subdirectories up to a specified depth from a remote server. +/// +/// The function fetches the directory's hash to verify its existence. If the directory is empty, +/// it loads subdirectories up to `max_depth` (absolute depth relative to the root directory). +/// If non-empty, it compares the hash to detect changes: if unchanged, it processes the local +/// directory; if changed, it fetches and loads the updated directory from the remote server. +/// +/// # Arguments +/// * `parent_path` - The path to the directory to preload (must be a valid, existing path). +/// * `max_depth` - The maximum absolute depth of subdirectories to load, relative to the root. pub async fn load_dir(store: Arc, parent_path: String, max_depth: usize) -> bool { if parent_path.matches('/').count() >= max_depth { println!("max depth reached for path: {}", parent_path); return false; } - if max_depth < config::load_dir_depth() + 2 { + if max_depth < store.max_depth() + 2 { println!("max depth is less than config, skipping: {}", parent_path); return false; } @@ -778,7 +897,7 @@ pub async fn load_dir(store: Arc, parent_path: String, max_dept return true; } println!("load_dir parent_path {:?}", parent_path); - + //empty dir,load the dir to the max_depth. if dirs.get(&parent_path).unwrap().file_list.is_empty() { load_dir_depth(store.clone(), parent_path.to_owned(), max_depth).await; @@ -791,7 +910,7 @@ pub async fn load_dir(store: Arc, parent_path: String, max_dept } return false; } - // if the dir's hash is same as the parent dir's hash, + // if the dir's hash is same as the parent dir's hash, //then check the subdir from the db,no need to get from the server.. if dirs.get(&parent_path).unwrap().hash == self_hash { let item = store.persistent_path_store.get_item(parent_inode).unwrap(); @@ -880,8 +999,8 @@ pub async fn load_dir(store: Arc, parent_path: String, max_dept } #[async_recursion] -///this function is only used to update the directory which has been loaded. -/// it will update the directory but do not load the new directory. +/// This function is only used to update the directory which has been loaded. +/// It will update the directory but do not load the new directory. pub async fn update_dir(store: Arc, parent_path: String) { let tree_db = store.persistent_path_store.clone(); let items = fetch_dir(&parent_path).await.unwrap().data; diff --git a/scorpio/src/dicfuse/tree_store.rs b/scorpio/src/dicfuse/tree_store.rs index bbecc2eb1..a2326404d 100644 --- a/scorpio/src/dicfuse/tree_store.rs +++ b/scorpio/src/dicfuse/tree_store.rs @@ -13,7 +13,7 @@ use super::store::ItemExt; pub struct TreeStorage { db: Db, } -const CONFIG_PATH: &str = "config.toml"; + #[derive(Serialize, Deserialize, Clone)] pub struct StorageItem { inode: u64, @@ -56,15 +56,13 @@ impl StorageItem { self.parent } } -use toml::Value; +// use toml::Value; #[allow(unused)] impl TreeStorage { pub fn new_from_db(db: Db) -> Self { TreeStorage { db } } pub fn new() -> io::Result { - let config_content = std::fs::read_to_string(CONFIG_PATH).map_err(Error::other)?; - let config: Value = toml::de::from_str(&config_content).map_err(Error::other)?; let store_path = config::store_path(); let path = format!("{}/path.db", store_path); let db = sled::open(path)?; diff --git a/scorpio/src/lib.rs b/scorpio/src/lib.rs index 15f38dcc9..b994ec9cc 100644 --- a/scorpio/src/lib.rs +++ b/scorpio/src/lib.rs @@ -2,7 +2,7 @@ extern crate log; pub mod daemon; -mod dicfuse; +pub mod dicfuse; pub mod fuse; pub mod manager; mod scolfs; diff --git a/scorpio/src/util/config.rs b/scorpio/src/util/config.rs index 0a4fd6711..66ea7f771 100644 --- a/scorpio/src/util/config.rs +++ b/scorpio/src/util/config.rs @@ -15,7 +15,7 @@ pub type ConfigResult = Result; pub struct ScorpioConfig { config: HashMap, } - +const DEFAULT_LOAD_DIR_DEPTH: usize = 3; // Global configuration management static SCORPIO_CONFIG: OnceLock = OnceLock::new(); @@ -147,7 +147,10 @@ fn get_config() -> &'static ScorpioConfig { config.insert("git_author".to_string(), "MEGA".to_string()); config.insert("git_email".to_string(), "admin@mega.org".to_string()); config.insert("config_file".to_string(), "config.toml".to_string()); - config.insert("load_dir_depth".to_string(), "3".to_string()); + config.insert( + "load_dir_depth".to_string(), + DEFAULT_LOAD_DIR_DEPTH.to_string(), + ); config.insert( "lfs_url".to_string(), "http://localhost:8000/lfs".to_string(), @@ -249,7 +252,7 @@ pub fn load_dir_depth() -> usize { .config .get("load_dir_depth") .and_then(|s| s.parse().ok()) - .unwrap_or(3) + .unwrap_or(DEFAULT_LOAD_DIR_DEPTH) } #[cfg(test)] mod tests { @@ -265,6 +268,7 @@ mod tests { config_file = "config.toml" lfs_url = "http://localhost:8000/lfs" dicfuse_readable = "true" + load_dir_depth = "3" "#; let config_path = "/tmp/scorpio.toml"; std::fs::write(config_path, config_content).expect("Failed to write test config file"); @@ -284,6 +288,7 @@ mod tests { file_blob_endpoint(), "http://localhost:8000/api/v1/file/blob" ); + assert_eq!(load_dir_depth(), 3); assert_eq!(config_file(), "config.toml"); } } diff --git a/scorpio/tests/dir_test.rs b/scorpio/tests/dir_test.rs new file mode 100644 index 000000000..73b186519 --- /dev/null +++ b/scorpio/tests/dir_test.rs @@ -0,0 +1,570 @@ +use core::panic; +use http::Method; +use lazy_static::lazy_static; +use libfuse_fs::passthrough::newlogfs::LoggingFileSystem; +use scorpio::dicfuse::store; +use scorpio::fuse::MegaFuse; +use scorpio::manager::fetch::CheckHash; +use scorpio::manager::ScorpioManager; +use scorpio::server::mount_filesystem; +use scorpio::util::config; +use serde::{Deserialize, Serialize}; +use std::collections::BTreeSet; +use std::collections::HashMap; +use std::net::TcpStream; +use std::path::{Path, PathBuf}; +use std::process::Command; +use std::time::Duration; +use std::time::SystemTime; +use std::time::UNIX_EPOCH; +use std::{env, fs, io}; +use std::{ffi::OsStr, sync::Arc}; +use testcontainers::core::wait::HttpWaitStrategy; +use testcontainers::{ + core::{IntoContainerPort, Mount, ReuseDirective, WaitFor}, + runners::AsyncRunner, + ContainerAsync, GenericImage, ImageExt, +}; +use tokio::sync::{mpsc, oneshot}; + +#[derive(Debug, Clone, Serialize, Deserialize)] +enum SCORCommand { + ImportArc(), // test init the scorpio directory structure + WatchDir(), // update the directory structure + LoadDir(String), // test cd/ls and preload the directory structure + GitAddFile(String, String), // add a new file and update to check the watch_dir + GitDeleteFile(String), // remove a new file and update to check the watch_dir + Shutdown, // finish and close the file system service +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +enum CommandResult { + StoreDirectoryStructure(HashMap>), //used to return the directory structure + Success, + Error(String), + InitFinish(usize), // used to indicate the initialization is finished +} + +lazy_static! { + static ref TARGET: String = { + let mut manifest = PathBuf::from(env!("CARGO_MANIFEST_DIR")); // Get env at compile time + manifest.pop(); + manifest.to_str().unwrap().to_string() + }; + static ref MONO: PathBuf = { + let path = if cfg!(target_os = "windows") { + format!("{}/target/debug/mono.exe", TARGET.as_str()) + } else { + format!("{}/target/debug/mono", TARGET.as_str()) + }; + PathBuf::from(path) + }; + + static ref SCOR_DIR: PathBuf = { + Path::new("/tmp/scorpio_dir_test").to_path_buf() + }; + +} +fn run_cmd(program: &str, args: &[&str], stdin: Option<&str>, envs: Option>) { + let mut cmd = assert_cmd::Command::new(program); + let mut cmd = cmd.args(args); + if let Some(stdin) = stdin { + cmd = cmd.write_stdin(stdin); + } + if let Some(envs) = envs { + cmd = cmd.envs(envs); + } + let assert = cmd.assert().success(); + let output = assert.get_output(); + + println!( + "Command success: {} {}\nStatus: {}\nStdout: {}", + program, + args.join(" "), + output.status, + String::from_utf8_lossy(&output.stdout), + ); +} + +fn run_git_cmd(args: &[&str]) { + run_cmd("git", args, None, None); +} + +fn is_port_in_use(port: u16) -> bool { + TcpStream::connect_timeout( + &format!("127.0.0.1:{}", port).parse().unwrap(), + Duration::from_millis(1000), + ) + .is_ok() +} + +///clone the git repository and push to the mono server +fn git_clone(url: &str, mono_server_url: &str) -> io::Result>> { + use std::collections::{BTreeSet, HashMap}; + + fs::create_dir_all(SCOR_DIR.to_owned())?; + + let is_valid_git_repo = || -> bool { + let git_config_path = SCOR_DIR.join("dir_test").join(".git"); + if !git_config_path.exists() { + return false; + } + true + }; + + env::set_current_dir(SCOR_DIR.to_owned())?; + + if !is_valid_git_repo() { + println!("No valid git repo found, cloning from {}", url); + run_git_cmd(&["clone", url]); + + let repo_name = url.split('/').next_back().unwrap().trim_end_matches(".git"); + let repo_dir = SCOR_DIR.join(repo_name); + env::set_current_dir(&repo_dir)?; + + let mono_url = format!("{}/third-party/dir_test.git", mono_server_url); + run_git_cmd(&["remote", "add", "mono", mono_url.as_str()]); + run_git_cmd(&["push", "--all", "mono"]); + } else { + println!("Using existing git repository"); + let repo_dir = SCOR_DIR.join("dir_test"); + + env::set_current_dir(&repo_dir)?; + let mono_url = format!("{}/third-party/dir_test.git", mono_server_url); + run_git_cmd(&["remote", "remove", "mono"]); + run_git_cmd(&["remote", "add", "mono", mono_url.as_str()]); + run_git_cmd(&["push", "--all", "mono"]); + } + + let mut depth_items: HashMap> = HashMap::new(); + + let output = Command::new("git").args(["ls-files"]).output()?; + let files = String::from_utf8_lossy(&output.stdout); + + for file in files.lines() { + let depth = file.chars().filter(|&c| c == '/').count() as i32; + depth_items + .entry(depth) + .or_default() + .insert(file.to_string()); + + let parts: Vec<&str> = file.split('/').collect(); + for i in 0..parts.len() { + if i == 0 { + depth_items + .entry(0) + .or_default() + .insert(parts[0].to_string()); + } else { + let parent_path = parts[0..i].join("/"); + let parent_depth = (i - 1) as i32; + depth_items + .entry(parent_depth) + .or_default() + .insert(parent_path); + } + } + } + + let config = format!( + r#" +lfs_url = "{}" +store_path = "{}/store" +config_file = "config.toml" +git_author = "MEGA" +git_email = "admin@mega.org" +workspace = "{}/mount" +base_url = "{}" +dicfuse_readable = "true" +load_dir_depth = "4" + "#, + mono_server_url, + SCOR_DIR.to_str().unwrap(), + SCOR_DIR.to_str().unwrap(), + mono_server_url, + ); + + let store_path = SCOR_DIR.join("store"); + let _ = fs::remove_dir_all(&store_path); // Clear old store + let mount_path = SCOR_DIR.join("mount"); + let _ = fs::create_dir_all(&mount_path); + let umount_result = Command::new("umount") + .args(["-f", mount_path.to_str().unwrap()]) + .output(); + + match umount_result { + Ok(output) => { + if !output.status.success() { + println!( + "Umount warning: {}", + String::from_utf8_lossy(&output.stderr) + ); + } + } + Err(e) => { + println!("Umount command failed (this is usually okay): {}", e); + } + } + + fs::write(SCOR_DIR.join("scorpio.toml"), config)?; + Ok(depth_items) +} +async fn mono_container(mapping_port: u16) -> ContainerAsync { + println!("MONO {:?} ", MONO.to_str().unwrap()); + if !MONO.exists() { + panic!("MONO binary not found in \"target/debug/\", skip lfs test"); + } + if is_port_in_use(mapping_port) { + panic!("port {} is already in use", mapping_port); + } + let port_str = mapping_port.to_string(); + let cmd = vec![ + "/root/mono", + "service", + "multi", + "http", + "-p", + &port_str, + "--host", + "0.0.0.0", + ]; + + GenericImage::new("ubuntu", "latest") + .with_exposed_port(mapping_port.tcp()) + .with_wait_for(WaitFor::Http(Box::new( + HttpWaitStrategy::new("/") + .with_method(Method::GET) + .with_expected_status_code(404_u16), + ))) + .with_mapped_port(mapping_port, mapping_port.tcp()) + .with_mount(Mount::bind_mount(MONO.to_str().unwrap(), "/root/mono")) + .with_working_dir("/root") + .with_reuse(ReuseDirective::Never) + .with_cmd(cmd) + .start() + .await + .expect("Failed to start mono_server") +} + +pub async fn mono_bootstrap_servers(mapping_port: u16) -> (ContainerAsync, String) { + let container = mono_container(mapping_port).await; + let mega_ip = container.get_bridge_ip_address().await.unwrap(); + let mega_port: u16 = container.get_host_port_ipv4(mapping_port).await.unwrap(); + (container, format!("http://{}:{}", mega_ip, mega_port)) +} + +#[tokio::test] +///Use container to run mono server and test the scorpio service +async fn test_scorpio_service_with_containers() { + let (_container, mono_server_url) = mono_bootstrap_servers(12001).await; + println!("container: {}", mono_server_url); + let dir_list = git_clone("https://github.com/yyjeqhc/dir_test.git", &mono_server_url).unwrap(); + + let (cmd_tx, cmd_rx) = mpsc::channel(32); + let (result_tx, mut result_rx) = mpsc::channel(32); + + let scorpio_handle = tokio::spawn(test_scorpio_dir(cmd_rx, result_tx)); + + // This is the preload's relative depth for the dir to load. + let mut max_depth = 0; + + // Wait for the store:init_notify: Arc + tokio::select! { + _ = scorpio_handle => { + panic!("start the scorpio service failed"); + } + success = result_rx.recv() => { + if let Some(CommandResult::InitFinish(depth)) = success { + println!("scorpio service started successfully, max depth: {}", depth); + max_depth = depth; + } + } + } + println!("\n===== ImportArc ====="); + + cmd_tx.send(SCORCommand::ImportArc()).await.unwrap(); + + if let Some(result) = result_rx.recv().await { + match result { + CommandResult::StoreDirectoryStructure(store_items) => { + for i in 0..max_depth as i32 { + assert_eq!( + dir_list.get(&i).unwrap(), + store_items.get(&i).unwrap(), + "dir structure at depth {} does not match.", + i + ); + } + println!("import_arc,load dir success"); + } + _ => { + cmd_tx.send(SCORCommand::Shutdown).await.unwrap(); + let _ = result_rx.recv().await; + // let _ = scorpio_handle.await; + panic!("ImportArc failed to load dir."); + } + } + } + + println!("\n===== add a file and WatchDir ====="); + let test_file = "test_file.txt"; + let test_content = format!( + "now: {}", + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs() + ); + + cmd_tx + .send(SCORCommand::GitAddFile(test_file.to_string(), test_content)) + .await + .unwrap(); + + if let Some(result) = result_rx.recv().await { + match result { + CommandResult::Success => { + println!("git add success."); + } + _ => { + cmd_tx.send(SCORCommand::Shutdown).await.unwrap(); + // let _ = scorpio_handle.await; + let _ = result_rx.recv().await; + + panic!("git add file error."); + } + } + } + + cmd_tx.send(SCORCommand::WatchDir()).await.unwrap(); + + if let Some(result) = result_rx.recv().await { + match result { + CommandResult::StoreDirectoryStructure(result) => { + assert!( + result.get(&0).unwrap().contains(&test_file.to_string()), + "WatchDir fail: did not find the added file" + ); + } + _ => { + cmd_tx.send(SCORCommand::Shutdown).await.unwrap(); + // let _ = scorpio_handle.await; + let _ = result_rx.recv().await; + + panic!("WatchDir error."); + } + } + } + + println!("\n===== remove ad file and WatchDir ====="); + + cmd_tx + .send(SCORCommand::GitDeleteFile(test_file.to_string())) + .await + .unwrap(); + + if let Some(result) = result_rx.recv().await { + match result { + CommandResult::Success => { + println!("git remove success."); + } + _ => { + cmd_tx.send(SCORCommand::Shutdown).await.unwrap(); + // let _ = scorpio_handle.await; + let _ = result_rx.recv().await; + + panic!("git remove file error."); + } + } + } + + cmd_tx.send(SCORCommand::WatchDir()).await.unwrap(); + + if let Some(result) = result_rx.recv().await { + match result { + CommandResult::StoreDirectoryStructure(result) => { + assert!( + !result.get(&0).unwrap().contains(&test_file.to_string()), + "WatchDir fail: did not remove the file" + ); + } + _ => { + cmd_tx.send(SCORCommand::Shutdown).await.unwrap(); + // let _ = scorpio_handle.await; + let _ = result_rx.recv().await; + panic!("WatchDir: delete file error."); + } + } + } + + cmd_tx + .send(SCORCommand::LoadDir( + "/third-party/dir_test/1/1/2/3".to_string(), + )) + .await + .unwrap(); + + if let Some(result) = result_rx.recv().await { + match result { + CommandResult::StoreDirectoryStructure(store_items) => { + let mut expected_items: HashMap> = HashMap::new(); + let test_path = "1/1/2/3/".to_string(); + for git_files in dir_list.values() { + for file in git_files { + if file.starts_with(&test_path) { + let relative_path = file.trim_start_matches(&test_path); + expected_items + .entry(relative_path.matches('/').count() as i32) + .or_default() + .insert(relative_path.to_string()); + } + } + } + for i in 0..max_depth as i32 { + assert_eq!( + expected_items.get(&i).unwrap(), + store_items.get(&i).unwrap(), + "dir structure at depth {} does not match.", + i + ); + } + println!("load_dir,preload dir success"); + } + _ => { + cmd_tx.send(SCORCommand::Shutdown).await.unwrap(); + let _ = result_rx.recv().await; + panic!("load_dir fail"); + } + } + } + + cmd_tx.send(SCORCommand::Shutdown).await.unwrap(); + + // let _ = scorpio_handle.await; + let _ = result_rx.recv().await; + + println!("success to finish the test."); +} + +async fn test_scorpio_dir( + mut cmd_rx: mpsc::Receiver, + result_tx: mpsc::Sender, +) { + if let Err(e) = config::init_config(SCOR_DIR.join("scorpio.toml").to_str().unwrap()) { + eprintln!("init config fail {:?}", e); + let _ = result_tx + .send(CommandResult::Error("load config error".to_string())) + .await; + return; + } + + let mut manager = ScorpioManager { works: vec![] }; + manager.check().await; + //init scorpio configuration + let fuse_interface = MegaFuse::new_from_manager(&manager).await; + let workspace = config::workspace(); + let mountpoint = OsStr::new(workspace); + let lgfs = LoggingFileSystem::new(fuse_interface.clone()); + + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + + let mut mount_handle = mount_filesystem(lgfs, mountpoint).await; + let handle = &mut mount_handle; + + let arc_fuse = Arc::new(fuse_interface); + let repo_dir = SCOR_DIR.join("dir_test"); + + let shutdown_tx = shutdown_tx; + let fuse_interface = arc_fuse.clone(); + let store = fuse_interface.dic.clone().store.clone(); + + store.wait_for_ready().await; + result_tx + .send(CommandResult::InitFinish(store.max_depth())) + .await + .expect("Failed to send success signal"); + let cmd_handle = { + let result_tx = result_tx.clone(); + + tokio::spawn(async move { + while let Some(cmd) = cmd_rx.recv().await { + match cmd { + SCORCommand::ImportArc() => { + let base_path = "/third-party/dir_test"; + let depth_items = store.get_dir_by_path(base_path).await; + if depth_items.is_empty() { + let _ = result_tx + .send(CommandResult::Error("ImportArc fail".to_string())) + .await; + } else { + let _ = result_tx + .send(CommandResult::StoreDirectoryStructure(depth_items)) + .await; + } + } + SCORCommand::WatchDir() => { + store::watch_dir(store.clone()).await; + let dir_items = store.get_dir_by_path("/third-party/dir_test").await; + let _ = result_tx + .send(CommandResult::StoreDirectoryStructure(dir_items)) + .await; + } + SCORCommand::LoadDir(path) => { + let max_depth = path.matches('/').count() + config::load_dir_depth(); + store::load_dir(store.clone(), path.to_owned(), max_depth).await; + let dir_items = store.get_dir_by_path(&path).await; + if dir_items.is_empty() { + let _ = result_tx + .send(CommandResult::Error(format!("LoadDir fail: {}", path))) + .await; + } else { + let _ = result_tx + .send(CommandResult::StoreDirectoryStructure(dir_items)) + .await; + } + } + SCORCommand::GitAddFile(path, content) => { + env::set_current_dir(&repo_dir).unwrap(); + let _ = fs::write(repo_dir.join(&path), content); + run_git_cmd(&["add", &path]); + run_git_cmd(&["commit", "-m", "add file"]); + run_git_cmd(&["push", "--all", "mono"]); + result_tx.send(CommandResult::Success).await.unwrap(); + } + SCORCommand::GitDeleteFile(path) => { + env::set_current_dir(&repo_dir).unwrap(); + fs::remove_file(repo_dir.join(&path)).unwrap(); + run_git_cmd(&["add", &path]); + run_git_cmd(&["commit", "-m", "remove file"]); + run_git_cmd(&["push", "--all", "mono"]); + result_tx.send(CommandResult::Success).await.unwrap(); + } + SCORCommand::Shutdown => { + let _ = shutdown_tx.send(()); + break; + } + } + } + }) + }; + + tokio::select! { + res = handle => res.unwrap(), + _ = cmd_handle => { + println!("unmount...."); + mount_handle.unmount().await.unwrap(); + let _ = result_tx.send(CommandResult::Success).await; + } + _ = shutdown_rx => { + println!("unmount...."); + mount_handle.unmount().await.unwrap(); + let _ = result_tx.send(CommandResult::Success).await; + + } + } + + // let _ = cmd_handle.await; + // let _ = daemon_handle.await; + + println!("success to close the scorpio service."); +}