From d53dd97dfeef8fbf96834bd028b4f16fe254ebc7 Mon Sep 17 00:00:00 2001 From: yyjeqhc <1772413353@qq.com> Date: Fri, 16 May 2025 21:20:24 +0800 Subject: [PATCH 1/2] add watch mono's dir. --- scorpio/src/dicfuse/store.rs | 390 +++++++++++++++++++++++++++++++++-- 1 file changed, 376 insertions(+), 14 deletions(-) diff --git a/scorpio/src/dicfuse/store.rs b/scorpio/src/dicfuse/store.rs index 58b850906..c2bc5582e 100644 --- a/scorpio/src/dicfuse/store.rs +++ b/scorpio/src/dicfuse/store.rs @@ -8,6 +8,7 @@ use std::sync::Arc; use crossbeam::queue::SegQueue; 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}; @@ -33,6 +34,12 @@ impl Item { self.content_type == INODE_DICTIONARY } } +#[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq)] +//use hash to get the dir's status. +pub struct ItemExt { + pub item: Item, + pub hash: String, +} #[allow(unused)] pub struct DicItem { inode: u64, @@ -119,6 +126,13 @@ impl Iterator for ApiResponse { self.data.pop() } } + +struct ApiResponseExt { + _req_result: bool, + data: Vec, + _err_message: String, +} + #[derive(Debug)] pub struct DictionaryError { pub message: String, @@ -157,9 +171,102 @@ async fn fetch_tree(path: &str) -> Result { Err(e) => Err(e.into()), } } +async fn fetch_dir(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/commit-info?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(), + }); + } + }; + + #[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, + } + + 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, + }); + } + + 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(), + }) +} +/// 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 +pub struct DirItem { + hash: String, + file_list: HashMap, +} pub struct DictionaryStore { inodes: Arc>>>, + 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 @@ -183,6 +290,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())), }; let root_item = DicItem { inode: 1, @@ -192,7 +300,15 @@ impl DictionaryStore { parent: UNKNOW_INODE, // root dictory has no parent }; + let root_dir_item = DirItem { + hash: String::new(), + file_list: HashMap::new(), + }; init.inodes.lock().await.insert(1, root_item.into()); + init.dirs + .lock() + .await + .insert("/".to_string(), root_dir_item); init } async fn update_inode(&self, parent: u64, item: Item) -> std::io::Result { @@ -304,6 +420,19 @@ impl DictionaryStore { self.get_inode(inode).await } + /// get the inode from path + pub async fn get_inode_from_path(&self, path: &str) -> Result { + let inode = if path.is_empty() || path == "/" { + 1 + } else { + let binding = self.radix_trie.lock().await; + *binding + .get(&GPath::from(path.to_owned()).to_string()) + .ok_or(io::Error::new(io::ErrorKind::NotFound, "path not found"))? + }; + + Ok(inode) + } pub async fn do_readdir( &self, @@ -341,17 +470,35 @@ impl DictionaryStore { } pub async fn import_arc(store: Arc) { - // use the unlock queue instead of mpsc + Mutex + // use the unlock queue instead of mpsc Mutex let queue = Arc::new(SegQueue::new()); // init root path - let items = fetch_tree("").await.unwrap().data; + let items = fetch_dir("").await.unwrap().data; let active_producers = Arc::new(AtomicUsize::new(items.len())); - for it in items { - let is_dir = it.content_type == INODE_DICTIONARY; - let it_inode = store.update_inode(1, it).await.unwrap(); - if is_dir { - queue.push(it_inode); + { + let mut locks = store.dirs.lock().await; + // 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("/") + .unwrap() + .file_list + .insert(path.to_owned(), false); + let it_inode = store.update_inode(1, it.item).await.unwrap(); + if is_dir { + queue.push(it_inode); + locks.insert( + path, + DirItem { + hash: it.hash, + file_list: HashMap::new(), + }, + ); + } } } @@ -376,21 +523,39 @@ pub async fn import_arc(store: Arc) { } { if let Some(inode) = queue.pop() { // get path from path store. - let path = path_store.get_all_path(inode).unwrap().to_string(); + //get the whole path. + let path = + "/".to_string() + &path_store.get_all_path(inode).unwrap().to_string(); println!("Worker processing path: {}", path); - // get all children inode - match fetch_tree(&path).await { + match fetch_dir(&path).await { Ok(new_items) => { let new_items = new_items.data; for newit in new_items { - let is_dir = newit.is_dir(); - let new_inode = store.update_inode(inode, newit).await.unwrap(); + 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 + .insert(tmp_path.to_owned(), false); + let new_inode = + store.update_inode(inode, newit.item).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( + tmp_path, + DirItem { + hash: newit.hash, + file_list: HashMap::new(), + }, + ); } } } @@ -413,9 +578,206 @@ pub async fn import_arc(store: Arc) { } // wait for all workers to complete - while let Some(worker) = workers.pop() { - worker.await.expect("Worker panicked"); + // while let Some(worker) = workers.pop() { + // worker.await.expect("Worker panicked"); + // } + join_all(workers).await; + tokio::spawn(async move { + loop { + tokio::time::sleep(tokio::time::Duration::from_secs(60)).await; + watch_dir(store.clone()).await; + } + }); +} + +/// Watch the directory and update the dictionary +pub async fn watch_dir(store: Arc) { + // 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(); + + 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.item).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 mut remove_items = Vec::new(); + locks.get_mut("/").unwrap().file_list.retain(|path, v| { + let result = *v; + if *v == false { + 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(); + } } + + 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 { + if 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.item).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(), + }, + ); + } + } + } + + let mut remove_items = Vec::new(); + locks.get_mut(&parent).unwrap().file_list.retain(|path, v| { + let result = *v; + if *v == false { + 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); + + tree_db.remove_item(inode).unwrap(); + } + } + Err(_) => { + // Continue to the next iteration if there was an error + } + }; + + 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; + } + } + })); + } + + // wait for all workers to complete + join_all(workers).await; + println!("finish"); } #[cfg(test)] From 72b4b31827c31aafd7f2e6261579494a2450eb61 Mon Sep 17 00:00:00 2001 From: yyjeqhc <1772413353@qq.com> Date: Sat, 17 May 2025 17:01:11 +0800 Subject: [PATCH 2/2] add load dirs from db and fix some clippy. --- mercury/src/internal/pack/cache_object.rs | 2 +- mercury/src/internal/pack/utils.rs | 2 +- mercury/zstdelta/src/zstdelta.rs | 12 +- scorpio/src/dicfuse/store.rs | 189 ++++++++++++++-------- scorpio/src/dicfuse/tree_store.rs | 60 ++++--- scorpio/src/manager/fetch.rs | 5 +- scorpio/src/manager/store.rs | 10 +- 7 files changed, 172 insertions(+), 108 deletions(-) diff --git a/mercury/src/internal/pack/cache_object.rs b/mercury/src/internal/pack/cache_object.rs index d86acf6b2..94ed7c602 100644 --- a/mercury/src/internal/pack/cache_object.rs +++ b/mercury/src/internal/pack/cache_object.rs @@ -27,7 +27,7 @@ impl Deserialize<'a>> FileLoadStore for T { fn f_load(path: &Path) -> Result { let data = fs::read(path)?; let obj: T = bincode::serde::decode_from_slice(&data, bincode::config::standard()) - .map_err(|e| io::Error::new(io::ErrorKind::Other, e))? + .map_err(io::Error::other)? .0; Ok(obj) } diff --git a/mercury/src/internal/pack/utils.rs b/mercury/src/internal/pack/utils.rs index 2cede0a5c..fb8fe5740 100644 --- a/mercury/src/internal/pack/utils.rs +++ b/mercury/src/internal/pack/utils.rs @@ -351,7 +351,7 @@ mod tests { struct BrokenReader; impl Read for BrokenReader { fn read(&mut self, _: &mut [u8]) -> io::Result { - Err(io::Error::new(io::ErrorKind::Other, "error")) + Err(io::Error::other("error")) } } diff --git a/mercury/zstdelta/src/zstdelta.rs b/mercury/zstdelta/src/zstdelta.rs index 9e75fe8e5..2a2f928f7 100644 --- a/mercury/zstdelta/src/zstdelta.rs +++ b/mercury/zstdelta/src/zstdelta.rs @@ -86,7 +86,7 @@ pub fn diff(base: &[u8], data: &[u8]) -> io::Result> { unsafe { let cctx = ZSTD_createCCtx(); if cctx.is_null() { - return Err(io::Error::new(io::ErrorKind::Other, "cannot create CCtx")); + return Err(io::Error::other("cannot create CCtx")); } let max_outsize = ZSTD_compressBound(data.len()); @@ -107,7 +107,7 @@ pub fn diff(base: &[u8], data: &[u8]) -> io::Result> { if ZSTD_isError(outsize) != 0 { let msg = format!("cannot compress ({})", explain_error(outsize)); - Err(io::Error::new(io::ErrorKind::Other, msg)) + Err(io::Error::other(msg)) } else { buf.set_len(outsize); Ok(buf) @@ -120,7 +120,7 @@ pub fn apply(base: &[u8], delta: &[u8]) -> io::Result> { unsafe { let dctx = ZSTD_createDCtx(); if dctx.is_null() { - return Err(io::Error::new(io::ErrorKind::Other, "cannot create DCtx")); + return Err(io::Error::other("cannot create DCtx")); } ZSTD_DCtx_setMaxWindowSize(dctx, 1 << ZSTD_WINDOWLOG_MAX); @@ -128,7 +128,7 @@ pub fn apply(base: &[u8], delta: &[u8]) -> io::Result> { if size == ZSTD_CONTENTSIZE_ERROR as usize || size == ZSTD_CONTENTSIZE_UNKNOWN as usize { ZSTD_freeDCtx(dctx); let msg = "cannot get decompress size"; - return Err(io::Error::new(io::ErrorKind::Other, msg)); + return Err(io::Error::other(msg)); } let mut buf = vec![0u8; size]; @@ -146,13 +146,13 @@ pub fn apply(base: &[u8], delta: &[u8]) -> io::Result> { if ZSTD_isError(outsize) != 0 { let msg = format!("cannot decompress ({})", explain_error(outsize)); - Err(io::Error::new(io::ErrorKind::Other, msg)) + Err(io::Error::other(msg)) } else if outsize != size { let msg = format!( "decompress size mismatch (expected {}, got {})", size, outsize ); - Err(io::Error::new(io::ErrorKind::Other, msg)) + Err(io::Error::other(msg)) } else { Ok(buf) } diff --git a/scorpio/src/dicfuse/store.rs b/scorpio/src/dicfuse/store.rs index c2bc5582e..050027ba7 100644 --- a/scorpio/src/dicfuse/store.rs +++ b/scorpio/src/dicfuse/store.rs @@ -1,10 +1,6 @@ +use async_recursion::async_recursion; /// Read only file system for obtaining and displaying monorepo directory information use core::panic; -use std::collections::{HashMap, VecDeque}; -use std::io; -use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; -use std::sync::Arc; - use crossbeam::queue::SegQueue; use fuse3::raw::reply::ReplyEntry; use fuse3::FileType; @@ -12,6 +8,11 @@ use futures::future::join_all; use once_cell::sync::Lazy; use reqwest::Client; use serde::{Deserialize, Serialize}; +use std::collections::{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; @@ -154,6 +155,7 @@ impl From for DictionaryError { } // Get Mega dictionary tree from server +#[allow(unused)] async fn fetch_tree(path: &str) -> Result { static CLIENT: Lazy = Lazy::new(Client::new); let client = CLIENT.clone(); @@ -226,12 +228,10 @@ async fn fetch_dir(path: &str) -> Result { let base_path = if path.is_empty() || path == "/" { "".to_string() + } else if path.ends_with('/') { + path.to_string() } else { - if path.ends_with('/') { - path.to_string() - } else { - format!("{}/", path) - } + format!("{}/", path) }; for info in commit_info.data { @@ -276,47 +276,20 @@ pub struct DictionaryStore { impl DictionaryStore { pub async fn new() -> Self { let tree_store = TreeStorage::new().expect("Failed to create TreeStorage"); - tree_store.insert_item( - 1, - UNKNOW_INODE, - Item { - name: "".to_string(), - path: "/".to_string(), - content_type: INODE_DICTIONARY.to_string(), - }, - ); - let mut init = DictionaryStore { + DictionaryStore { next_inode: AtomicU64::new(2), 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())), - }; - 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(), - }; - init.inodes.lock().await.insert(1, root_item.into()); - init.dirs - .lock() - .await - .insert("/".to_string(), root_dir_item); - init + } } - async fn update_inode(&self, parent: u64, item: Item) -> std::io::Result { + + async fn update_inode(&self, parent: u64, item: ItemExt) -> std::io::Result { let alloc_inode = self .next_inode .fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1; - assert!(alloc_inode < READONLY_INODE); let prw = self.persistent_path_store.clone(); @@ -325,7 +298,7 @@ impl DictionaryStore { self.radix_trie .lock() .await - .insert(GPath::from(item.path.clone()).to_string(), alloc_inode); + .insert(GPath::from(item.item.path.clone()).to_string(), alloc_inode); prw.insert_item(alloc_inode, parent, item); //prw.append_child(parent, alloc_inode); } else { @@ -350,23 +323,63 @@ impl DictionaryStore { }; self.update_inode( parent.get_inode(), - Item { - name, - path: item_path, - content_type: INODE_DICTIONARY.to_string(), + ItemExt { + item: Item { + name, + path: item_path, + content_type: INODE_DICTIONARY.to_string(), + }, + hash: String::new(), }, ) .await } - + #[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( + path.to_string_lossy().to_string(), + DirItem { + hash: root_item.hash.to_owned(), + file_list: HashMap::new(), + }, + ); + let children = root_item.get_children(); + for child in children { + self.next_inode.fetch_max(child, Ordering::Relaxed); + 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 + .insert(child_path.to_string_lossy().to_string(), false); + self.radix_trie.lock().await.insert( + GPath::from(child_path.to_string_lossy().to_string()).to_string(), + child, + ); + + if child_item.is_dir() { + self.load_dirs(child_path, child).await?; + } + } + Ok(()) + } + pub async fn load_db(&self) -> Result<(), io::Error> { + let mut path = PathBuf::from("/"); + self.load_dirs(path, 1).await?; + Ok(()) + } pub async fn import(&self) { - let items = fetch_tree("").await.unwrap().data; + let items = fetch_dir("").await.unwrap().data; //let root_inode = self.inodes.lock().await.get(&1).unwrap().clone(); // deque for bus. let mut queue = VecDeque::::new(); for it in items { - let is_dir = it.content_type == INODE_DICTIONARY; + let is_dir = it.item.content_type == INODE_DICTIONARY; let it_inode = self.update_inode(1, it).await.unwrap(); if is_dir { queue.push_back(it_inode); @@ -385,12 +398,12 @@ impl DictionaryStore { let path = it.to_string(); println!("fetch path :{}", path); // get tree by parent inode. - new_items = fetch_tree(&path).await.unwrap().data; + new_items = fetch_dir(&path).await.unwrap().data; // Insert all new inode. for newit in new_items { //println!("import item :{:?}",newit); - let is_dir = newit.is_dir(); + let is_dir = newit.item.is_dir(); let new_inode = self.update_inode(one_inode, newit).await.unwrap(); // Await the update_inode call // push to queue to BFS. if is_dir { @@ -470,6 +483,47 @@ 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 let queue = Arc::new(SegQueue::new()); @@ -488,7 +542,7 @@ pub async fn import_arc(store: Arc) { .unwrap() .file_list .insert(path.to_owned(), false); - let it_inode = store.update_inode(1, it.item).await.unwrap(); + let it_inode = store.update_inode(1, it.clone()).await.unwrap(); if is_dir { queue.push(it_inode); locks.insert( @@ -544,7 +598,7 @@ pub async fn import_arc(store: Arc) { .file_list .insert(tmp_path.to_owned(), false); let new_inode = - store.update_inode(inode, newit.item).await.unwrap(); + 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); @@ -592,6 +646,7 @@ 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()); @@ -626,7 +681,7 @@ pub async fn watch_dir(store: Arc) { .unwrap() .file_list .insert(path.to_owned(), true); - let _ = store.update_inode(1, it.item).await.unwrap(); + let _ = store.update_inode(1, it.clone()).await.unwrap(); //fetch a new dir. if is_dir { queue.push(path.to_owned()); @@ -644,7 +699,7 @@ pub async fn watch_dir(store: Arc) { let mut remove_items = Vec::new(); locks.get_mut("/").unwrap().file_list.retain(|path, v| { let result = *v; - if *v == false { + if !(*v) { remove_items.push(path.clone()); //delete storageItem // let inode = store.get_inode_from_path(&path).await.unwrap(); @@ -706,14 +761,14 @@ pub async fn watch_dir(store: Arc) { .file_list .insert(tmp_path.to_owned(), true); //should care the file'hash? - if is_dir { - if 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); - } + 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 @@ -721,8 +776,10 @@ pub async fn watch_dir(store: Arc) { .unwrap() .file_list .insert(tmp_path.to_owned(), true); - let _ = - store.update_inode(parent_inode, newit.item).await.unwrap(); + let _ = store + .update_inode(parent_inode, newit.clone()) + .await + .unwrap(); //insert new dir if is_dir { producers.fetch_add(1, Ordering::Relaxed); @@ -741,7 +798,7 @@ pub async fn watch_dir(store: Arc) { let mut remove_items = Vec::new(); locks.get_mut(&parent).unwrap().file_list.retain(|path, v| { let result = *v; - if *v == false { + if !(*v) { remove_items.push(path.clone()); // let inode = store.get_inode_from_path(&path).await.unwrap(); // tree_db.remove_item(inode); diff --git a/scorpio/src/dicfuse/tree_store.rs b/scorpio/src/dicfuse/tree_store.rs index 3761c8d4e..5e8a26e0f 100644 --- a/scorpio/src/dicfuse/tree_store.rs +++ b/scorpio/src/dicfuse/tree_store.rs @@ -7,7 +7,7 @@ use std::io; use std::io::{Error, ErrorKind}; use super::abi::{default_dic_entry, default_file_entry}; -use super::store::Item; +use super::store::ItemExt; /// inode -> StorageItem{ inode, parent, name, is_dir, children } pub struct TreeStorage { @@ -21,6 +21,7 @@ pub struct StorageItem { pub name: String, is_dir: bool, // True for Directory . children: Vec, + pub hash: String, } impl StorageItem { @@ -66,15 +67,16 @@ impl TreeStorage { Ok(TreeStorage { db }) } /// Insert an item and update the parent item's children list. - pub fn insert_item(&self, inode: u64, parent: u64, item: Item) -> io::Result<()> { + pub fn insert_item(&self, inode: u64, parent: u64, item: ItemExt) -> io::Result<()> { // create a StorageItem - let is_dir = item.content_type == "directory"; + let is_dir = item.item.content_type == "directory"; let storage_item = StorageItem { inode, parent, - name: item.name.clone(), + name: item.item.name.clone(), is_dir, children: Vec::new(), + hash: item.hash, }; // Insert an item into db and update the parent item's children list. @@ -179,9 +181,9 @@ impl TreeStorage { } #[cfg(test)] mod tests { - use core::panic; - use super::*; + use crate::dicfuse::store::Item; + use core::panic; fn setup(path: &str) -> io::Result { if std::path::Path::new(path).exists() { @@ -197,24 +199,30 @@ mod tests { #[test] fn test_insert_and_get_item() { let storage = setup("test_insert_and_get_item").unwrap(); - let item = Item { - name: String::from("Test Item"), - path: String::from("/path/to/item"), - content_type: String::from("text/plain"), + let item = ItemExt { + item: Item { + name: String::from("Test Item"), + path: String::from("/path/to/item"), + content_type: String::from("text/plain"), + }, + hash: String::new(), }; storage.insert_item(1, 0, item.clone()).unwrap(); let retrieved_item = storage.get_item(1).unwrap(); - assert_eq!(item.name, retrieved_item.name); + assert_eq!(item.item.name, retrieved_item.name); unset("test_insert_and_get_item"); } #[test] fn test_remove_item() { let storage = setup("test_remove_item").unwrap(); - let item = Item { - name: String::from("Test Item"), - path: String::from("/path/to/item"), - content_type: String::from("text/plain"), + let item = ItemExt { + item: Item { + name: String::from("Test Item"), + path: String::from("/path/to/item"), + content_type: String::from("text/plain"), + }, + hash: String::new(), }; storage.insert_item(2, 0, item.clone()).unwrap(); storage.remove_item(2).unwrap(); @@ -224,15 +232,21 @@ mod tests { #[test] fn test_list_items() { let storage = setup("test_list_items").unwrap(); - let item1 = Item { - name: String::from("Test Item 1"), - path: String::from("/path/to/item1"), - content_type: String::from("text/plain"), + let item1 = ItemExt { + item: Item { + name: String::from("Test Item 1"), + path: String::from("/path/to/item1"), + content_type: String::from("text/plain"), + }, + hash: String::new(), }; - let item2 = Item { - name: String::from("Test Item 2"), - path: String::from("/path/to/item2"), - content_type: String::from("image/png"), + let item2 = ItemExt { + item: Item { + name: String::from("Test Item 2"), + path: String::from("/path/to/item2"), + content_type: String::from("image/png"), + }, + hash: String::new(), }; storage.insert_item(3, 0, item1.clone()).unwrap(); storage.insert_item(4, 0, item2.clone()).unwrap(); diff --git a/scorpio/src/manager/fetch.rs b/scorpio/src/manager/fetch.rs index 5f869f5a6..57b951cf4 100644 --- a/scorpio/src/manager/fetch.rs +++ b/scorpio/src/manager/fetch.rs @@ -294,10 +294,7 @@ async fn set_parent_commit(work_path: &Path) -> std::io::Result<()> { Ok(info) => info, Err(e) => { eprintln!("Failed to fetch parent commit info: {}", e); - return Err(std::io::Error::new( - std::io::ErrorKind::Other, - "Failed to fetch parent commit info", - )); + return Err(std::io::Error::other("Failed to fetch parent commit info")); } }; diff --git a/scorpio/src/manager/store.rs b/scorpio/src/manager/store.rs index c79f9b204..bc15ef017 100644 --- a/scorpio/src/manager/store.rs +++ b/scorpio/src/manager/store.rs @@ -202,7 +202,7 @@ impl ModifiedStore for sled::Db { .to_string(); Ok((PathBuf::from(path), hash)) } - Err(e) => Err(Error::new(ErrorKind::Other, e)), + Err(e) => Err(Error::other(e)), }) .collect::>>() } @@ -255,11 +255,7 @@ impl> TreesStore { } fn get_bypath(&self, path: PathBuf) -> Result { - match self - .db - ._get(&path) - .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))? - { + match self.db._get(&path).map_err(std::io::Error::other)? { Some(encoded_value) => Ok(encoded_value), None => Err(std::io::Error::new( std::io::ErrorKind::NotFound, @@ -373,7 +369,7 @@ mod kv { fn from(e: KvError) -> Self { match e { KvError::IoError(e) => e, - _ => std::io::Error::new(std::io::ErrorKind::Other, e), + _ => std::io::Error::other(e), } } }