diff --git a/ceres/src/pack/handler.rs b/ceres/src/pack/handler.rs index 9eb933545..e2c23c9ab 100644 --- a/ceres/src/pack/handler.rs +++ b/ceres/src/pack/handler.rs @@ -62,13 +62,14 @@ pub trait PackHandler: Send + Sync { &self, tree: Tree, exist_objs: &HashSet, + counted_obj: &mut HashSet, obj_num: &AtomicUsize, ) { let mut search_tree_ids = vec![]; let mut search_blob_ids = vec![]; for item in &tree.tree_items { let hash = item.id.to_plain_str(); - if !exist_objs.contains(&hash) { + if !exist_objs.contains(&hash) && counted_obj.insert(hash.clone()) { if item.mode == TreeItemMode::Tree { search_tree_ids.push(hash.clone()) } else { @@ -79,29 +80,47 @@ pub trait PackHandler: Send + Sync { obj_num.fetch_add(search_blob_ids.len(), Ordering::SeqCst); let trees = self.get_trees_by_hashes(search_tree_ids).await.unwrap(); for t in trees { - self.traverse_for_count(t, exist_objs, obj_num).await; + self.traverse_for_count(t, exist_objs, counted_obj, obj_num) + .await; } obj_num.fetch_add(1, Ordering::SeqCst); } + /// Traverse a tree structure asynchronously. + /// + /// This function traverses a given tree, keeps track of processed objects, and optionally sends + /// traversal data to a provided sender. The function will: + /// 1. Traverse the tree and calculate the quantities of tree and blob items. + /// 2. If a sender is provided, send blob and tree data via the sender. + /// + /// # Parameters + /// - `tree`: The tree structure to traverse. + /// - `exist_objs`: A mutable reference to a set containing already processed object IDs. + /// - `sender`: An optional sender for sending traversal data. + /// + /// # Details + /// - The function processes tree items, distinguishing between tree and blob items. + /// - It collects IDs of items that have not been processed yet. + /// - It retrieves and sends blob data if a sender is provided. + /// - It recursively traverses sub-trees. + /// - It sends the entire tree data if a sender is provided. async fn traverse( &self, tree: Tree, exist_objs: &mut HashSet, sender: Option<&tokio::sync::mpsc::Sender>, ) { - exist_objs.insert(tree.id.to_plain_str()); let mut search_tree_ids = vec![]; let mut search_blob_ids = vec![]; + for item in &tree.tree_items { let hash = item.id.to_plain_str(); - if !exist_objs.contains(&hash) { + if exist_objs.insert(hash.clone()) { if item.mode == TreeItemMode::Tree { - search_tree_ids.push(hash.clone()) + search_tree_ids.push(hash); } else { - search_blob_ids.push(hash.clone()); + search_blob_ids.push(hash); } - exist_objs.insert(hash); } } @@ -112,10 +131,12 @@ pub trait PackHandler: Send + Sync { sender.send(blob.into()).await.unwrap(); } } + let trees = self.get_trees_by_hashes(search_tree_ids).await.unwrap(); for t in trees { self.traverse(t, exist_objs, sender).await; } + if let Some(sender) = sender { sender.send(tree.into()).await.unwrap(); } diff --git a/ceres/src/pack/import_repo.rs b/ceres/src/pack/import_repo.rs index bc1fab797..f1b011c2a 100644 --- a/ceres/src/pack/import_repo.rs +++ b/ceres/src/pack/import_repo.rs @@ -1,3 +1,5 @@ +use std::{collections::{HashMap, HashSet}, str::FromStr, sync::atomic::{AtomicUsize, Ordering}}; + use async_trait::async_trait; use bytes::Bytes; use tokio::sync::mpsc; @@ -9,7 +11,7 @@ use jupiter::{ context::Context, storage::{batch_query_by_columns, GitStorageProvider}, }; -use mercury::internal::pack::encode::PackEncoder; +use mercury::{hash::SHA1, internal::pack::encode::PackEncoder}; use mercury::{ errors::GitError, internal::{ @@ -119,10 +121,95 @@ impl PackHandler for ImportRepo { async fn incremental_pack( &self, - _want: Vec, - _have: Vec, + mut want: Vec, + have: Vec, ) -> Result>, GitError> { - unimplemented!() + let pack_config = &self.context.config.pack; + let storage = self.context.services.git_db_storage.clone(); + let obj_num = AtomicUsize::new(0); + + let mut exist_objs = HashSet::new(); + + let mut want_commits: Vec = storage + .get_commits_by_hashes(&self.repo, &want) + .await + .unwrap() + .into_iter() + .map(|x| x.into()) + .collect(); + let mut traversal_list: Vec = want_commits.clone(); + + // traverse commit's all parents to find the commit that client does not have + while let Some(temp) = traversal_list.pop() { + for p_commit_id in temp.parent_commit_ids { + let p_commit_id = p_commit_id.to_plain_str(); + + if !have.contains(&p_commit_id) && !want.contains(&p_commit_id) { + let parent: Commit = storage + .get_commit_by_hash(&self.repo, &p_commit_id) + .await + .unwrap() + .unwrap() + .into(); + want_commits.push(parent.clone()); + want.push(p_commit_id); + traversal_list.push(parent); + } + } + } + + let want_tree_ids = want_commits + .iter() + .map(|c| c.tree_id.to_plain_str()) + .collect(); + let want_trees: HashMap = storage + .get_trees_by_hashes(&self.repo, want_tree_ids) + .await + .unwrap() + .into_iter() + .map(|m| (SHA1::from_str(&m.tree_id).unwrap(), m.into())) + .collect(); + + obj_num.fetch_add(want_commits.len(), Ordering::SeqCst); + + let have_commits = storage.get_commits_by_hashes(&self.repo, &have).await.unwrap(); + let have_trees = storage + .get_trees_by_hashes(&self.repo, have_commits.iter().map(|x| x.tree.clone()).collect()) + .await + .unwrap(); + // traverse to get exist_objs + for have_tree in have_trees { + self.traverse(have_tree.into(), &mut exist_objs, None).await; + } + + let mut counted_obj = HashSet::new(); + // traverse for get obj nums + for c in want_commits.clone() { + self.traverse_for_count( + want_trees.get(&c.tree_id).unwrap().clone(), + &exist_objs, + &mut counted_obj, + &obj_num, + ) + .await; + } + let (entry_tx, entry_rx) = mpsc::channel(pack_config.channel_message_size); + let (stream_tx, stream_rx) = mpsc::channel(pack_config.channel_message_size); + let encoder = PackEncoder::new(obj_num.into_inner(), 0, stream_tx); + encoder.encode_async(entry_rx).await.unwrap(); + + for c in want_commits { + self.traverse( + want_trees.get(&c.tree_id).unwrap().clone(), + &mut exist_objs, + Some(&entry_tx), + ) + .await; + entry_tx.send(c.into()).await.unwrap(); + } + drop(entry_tx); + + Ok(ReceiverStream::new(stream_rx)) } async fn get_trees_by_hashes(&self, hashes: Vec) -> Result, MegaError> { diff --git a/ceres/src/pack/monorepo.rs b/ceres/src/pack/monorepo.rs index cb8d79802..a32adab6f 100644 --- a/ceres/src/pack/monorepo.rs +++ b/ceres/src/pack/monorepo.rs @@ -188,7 +188,7 @@ impl PackHandler for MonoRepo { .unwrap() .unwrap() .into(); - self.traverse_for_count(tree.clone(), &HashSet::new(), &obj_num) + self.traverse_for_count(tree.clone(), &HashSet::new(), &mut HashSet::new(), &obj_num) .await; obj_num.fetch_add(1, Ordering::SeqCst); @@ -267,11 +267,13 @@ impl PackHandler for MonoRepo { self.traverse(have_tree.into(), &mut exist_objs, None).await; } + let mut counted_obj = HashSet::new(); // traverse for get obj nums for c in want_commits.clone() { self.traverse_for_count( want_trees.get(&c.tree_id).unwrap().clone(), &exist_objs, + &mut counted_obj, &obj_num, ) .await; diff --git a/jupiter/src/storage/git_db_storage.rs b/jupiter/src/storage/git_db_storage.rs index 52304dcce..0c23f1f6d 100644 --- a/jupiter/src/storage/git_db_storage.rs +++ b/jupiter/src/storage/git_db_storage.rs @@ -215,6 +215,19 @@ impl GitDbStorage { .unwrap()) } + pub async fn get_commits_by_hashes( + &self, + repo: &Repo, + hashes: &Vec, + ) -> Result, MegaError> { + Ok(git_commit::Entity::find() + .filter(git_commit::Column::RepoId.eq(repo.repo_id)) + .filter(git_commit::Column::CommitId.is_in(hashes)) + .all(self.get_connection()) + .await + .unwrap()) + } + pub async fn get_commits_by_repo_id( &self, repo: &Repo, diff --git a/mercury/src/internal/pack/encode.rs b/mercury/src/internal/pack/encode.rs index 2b1185d92..402cfe49d 100644 --- a/mercury/src/internal/pack/encode.rs +++ b/mercury/src/internal/pack/encode.rs @@ -120,7 +120,10 @@ impl PackEncoder { } None => { if self.process_index != self.object_number { - panic!("not all objects are encoded"); + panic!( + "not all objects are encoded, process:{}, total:{}", + self.process_index, self.object_number + ); } break; } @@ -296,5 +299,4 @@ mod tests { assert_eq!(data[0], 0b_1101_0101); assert_eq!(data[1], 0b_0000_0101); } - }