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/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/Cargo.toml b/scorpio/Cargo.toml index 56c9e4703..171c8436e 100644 --- a/scorpio/Cargo.toml +++ b/scorpio/Cargo.toml @@ -49,12 +49,22 @@ fjall = "2.8.0" thiserror = "2.0.12" 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/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..75dcdb95b 100644 --- a/scorpio/src/dicfuse/async_io.rs +++ b/scorpio/src/dicfuse/async_io.rs @@ -6,11 +6,11 @@ use fuse3::raw::prelude::*; use fuse3::raw::reply::DirectoryEntry; use fuse3::{Errno, Inode, Result}; +use super::Dicfuse; +use crate::dicfuse::store::load_dir; 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> @@ -136,6 +136,18 @@ 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 + .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; + } Ok(()) } @@ -213,7 +225,7 @@ impl Filesystem for Dicfuse { })); } } - println!("{:?}", d); + // println!("{:?}", d); Ok(ReplyDirectoryPlus { entries: iter(d.into_iter()), }) 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 050027ba7..0f544d796 100644 --- a/scorpio/src/dicfuse/store.rs +++ b/scorpio/src/dicfuse/store.rs @@ -1,21 +1,22 @@ +use crate::READONLY_INODE; use async_recursion::async_recursion; /// 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; 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 crate::READONLY_INODE; +use tokio::sync::Notify; use super::abi::{default_dic_entry, default_file_entry}; use super::tree_store::{StorageItem, TreeStorage}; @@ -41,6 +42,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, @@ -127,7 +145,6 @@ impl Iterator for ApiResponse { self.data.pop() } } - struct ApiResponseExt { _req_result: bool, data: Vec, @@ -193,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) => { @@ -257,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 @@ -266,10 +338,13 @@ 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 + max_depth: Arc, // max depth for loading directories + init_notify: Arc, // used in dir_test to notify the start of the test.. } #[allow(unused)] @@ -281,10 +356,21 @@ 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()), + 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 @@ -312,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()); @@ -334,10 +480,64 @@ 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(); + 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)?; - self.dirs.lock().await.insert( + self.dirs.insert( path.to_string_lossy().to_string(), DirItem { hash: root_item.hash.to_owned(), @@ -350,8 +550,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 @@ -372,6 +570,7 @@ impl DictionaryStore { self.load_dirs(path, 1).await?; Ok(()) } + pub async fn import(&self) { let items = fetch_dir("").await.unwrap().data; @@ -482,67 +681,31 @@ 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(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 - .lock() - .await - .insert("/".to_string(), root_dir_item); - } - // use the unlock queue instead of mpsc Mutex +/// 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()); - - // 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 mut locks = store.dirs.lock().await; - // let dir_item = locks.get_mut("/").unwrap(); + let locks = store.dirs.clone(); 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( @@ -576,7 +739,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(); @@ -585,14 +747,11 @@ 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(); store .dirs - .lock() - .await .get_mut(&path) .unwrap() .file_list @@ -601,9 +760,13 @@ 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); - store.dirs.lock().await.insert( + 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 { hash: newit.hash, @@ -636,6 +799,52 @@ 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() { + store.init_notify.notify_waiters(); + 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 = 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 { tokio::time::sleep(tokio::time::Duration::from_secs(60)).await; @@ -644,197 +853,232 @@ pub async fn import_arc(store: Arc) { }); } -/// 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()); +/// 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. + if data.len() == 1 { + data[0].hash.to_owned() + } else { + String::new() + } +} + +#[async_recursion] +/// 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 < store.max_depth() + 2 { + println!("max depth is less than config, skipping: {}", parent_path); + return false; + } + + 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; - let items = fetch_dir("").await.unwrap().data; - { - let mut locks = store.dirs.lock().await; - 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 dir_it = locks.get_mut(&path).unwrap(); - dir_it.hash = it.hash; - 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(), - }, - ); - } + //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 mut locks = store.dirs.lock().await; - 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 dir_it = locks.get_mut(&tmp_path).unwrap(); - dir_it.hash = newit.hash; - 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 5e8a26e0f..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, @@ -51,16 +51,18 @@ impl StorageItem { pub fn get_name(&self) -> String { self.name.clone() } + + pub fn get_parent(&self) -> u64 { + 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)?; @@ -178,6 +180,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/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 64a5df7d1..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,6 +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(), + DEFAULT_LOAD_DIR_DEPTH.to_string(), + ); config.insert( "lfs_url".to_string(), "http://localhost:8000/lfs".to_string(), @@ -191,6 +195,7 @@ fn validate(config: &mut HashMap) -> ConfigResult<()> { "config_file", "lfs_url", "dicfuse_readable", + "load_dir_depth", ]; for key in required_keys { @@ -240,6 +245,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(DEFAULT_LOAD_DIR_DEPTH) +} #[cfg(test)] mod tests { use super::*; @@ -254,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"); @@ -273,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."); +}