Add multi_ack_detailed capability for Libra fetching with Mega server & improve performance

libra/Cargo.toml

@@ -47,6 +47,7 @@ async_static = "0.1.3"
 once_cell = "1.19.0"
 byte-unit = "5.1.4"
 scopeguard = "1.2.0"
+lru-mem = "0.3.0"
 
 [target.'cfg(unix)'.dependencies] # only on Unix
 pager = "0.16.0"

libra/src/command/push.rs

@@ -225,10 +225,12 @@ fn incremental_objs(local_ref: SHA1, remote_ref: SHA1) -> HashSet<Entry> {
 
 
     let mut objs = HashSet::new();
+    let mut visit = HashSet::new(); // avoid duplicate commit visit
     let exist_commits = collect_history_commits(&remote_ref);
     let mut queue = VecDeque::new();
     if !exist_commits.contains(&local_ref) {
         queue.push_back(local_ref);
+        visit.insert(local_ref);
     }
     let mut root_commit = None;
 
@@ -238,15 +240,16 @@ fn incremental_objs(local_ref: SHA1, remote_ref: SHA1) -> HashSet<Entry> {
         if parents.is_empty() {
             if root_commit.is_none() {
                 root_commit = Some(commit.id);
-            } else {
-                eprintln!("fatal: multiple root commits");
+            } else if root_commit != Some(commit.id) {
+                eprintln!("{}", "fatal: multiple root commits".red());
             }
         }
         for parent in parents.iter() {
             let parent_tree = Commit::load(parent).tree_id;
             objs.extend(diff_tree_objs(Some(&parent_tree), &commit.tree_id));
-            if !exist_commits.contains(parent) {
+            if !exist_commits.contains(parent) && !visit.contains(parent) {
                 queue.push_back(*parent);
+                visit.insert(*parent);
             }
         }
         objs.insert(commit.into());

libra/src/internal/protocol/https_client.rs

@@ -220,7 +220,7 @@ async fn generate_upload_pack_content(have: &Vec<String>, want: &Vec<String>) ->
     let mut buf = BytesMut::new();
     let mut write_first_line = false;
 
-    let capability = ["side-band-64k", "ofs-delta"].join(" ");
+    let capability = ["side-band-64k", "ofs-delta", "multi_ack_detailed"].join(" ");
     for w in want {
         if !write_first_line {
             add_pkt_line_string(

libra/src/internal/protocol/lfs_client.rs

@@ -93,6 +93,11 @@ impl LFSClient {
             })
         }
 
+        if lfs_objs.is_empty() {
+            tracing::info!("No LFS objects to push.");
+            return Ok(());
+        }
+
         { // verify locks
             let (code, locks) = self.verify_locks(VerifiableLockRequest {
                 refs: Ref { name: command::lfs::current_refspec().await.unwrap() },

libra/src/utils/client_storage.rs

@@ -3,13 +3,14 @@ use std::collections::HashSet;
 use std::io::prelude::*;
 use std::path::{Path, PathBuf};
 use std::str::FromStr;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 
 use byteorder::{BigEndian, ReadBytesExt};
 use flate2::Compression;
 use flate2::read::ZlibDecoder;
 use flate2::write::ZlibEncoder;
-
+use lru_mem::LruCache;
+use once_cell::sync::Lazy;
 use mercury::internal::pack::cache_object::CacheObject;
 use mercury::internal::pack::Pack;
 use mercury::errors::GitError;
@@ -18,6 +19,10 @@ use mercury::internal::object::types::ObjectType;
 use mercury::utils::read_sha1;
 
 use crate::command;
+static PACK_OBJ_CACHE: Lazy<Mutex<LruCache<String, CacheObject>>> = Lazy::new(|| {
+    // `lazy_static!` may affect IDE's code completion
+    Mutex::new(LruCache::new(1024 * 1024 * 200))
+});
 
 #[derive(Default)]
 pub struct ClientStorage {
@@ -54,7 +59,7 @@ impl ClientStorage {
             let (obj_type, _, _) = Self::parse_header(&data);
             ObjectType::from_string(&obj_type)
         } else {
-            self.get_from_pack(obj_id).unwrap()
+            self.get_from_pack(obj_id)?
                 .map(|x| x.1)
                 .ok_or(GitError::ObjectNotFound(obj_id.to_plain_str()))
         }
@@ -163,7 +168,7 @@ impl ClientStorage {
             Ok(data[end_of_header + 1..].to_vec())
         } else {
             // Ok(self.get_from_pack(object_id)?.unwrap().0)
-            self.get_from_pack(object_id).unwrap()
+            self.get_from_pack(object_id)?
                 .map(|x| x.0)
                 .ok_or(GitError::ObjectNotFound(object_id.to_plain_str()))
         }
@@ -308,19 +313,26 @@ impl ClientStorage {
 
     /// Read object from pack file, with offset
     fn read_pack_obj(pack_file: &Path, offset: u64) -> Result<CacheObject, GitError> {
+        let cache_key = format!("{:?}-{}", pack_file.file_name().unwrap(), offset);
+        // read cache
+        if let Some(cached) = PACK_OBJ_CACHE.lock().unwrap().get(&cache_key) {
+            return Ok(cached.clone());
+        }
+
         let file = fs::File::open(pack_file)?;
         let mut pack_reader = io::BufReader::new(&file);
         pack_reader.seek(io::SeekFrom::Start(offset))?;
         let mut pack = Pack::new(None, None, None, false);
-        let mut offset = offset as usize;
-        let obj = pack.decode_pack_object(&mut pack_reader, &mut offset)?;
-        match obj.obj_type {
+        let obj = {
+            let mut offset = offset as usize;
+            pack.decode_pack_object(&mut pack_reader, &mut offset)? // offset will be updated!
+        };
+        let full_obj = match obj.obj_type {
             ObjectType::OffsetDelta => {
                 let base_offset = obj.base_offset;
                 let base_obj = Self::read_pack_obj(pack_file, base_offset as u64)?;
                 let base_obj = Arc::new(base_obj);
-                let new_obj = Pack::rebuild_delta(obj, base_obj);
-                Ok(new_obj)
+                Pack::rebuild_delta(obj, base_obj) // new obj
             }
             ObjectType::HashDelta => {
                 let base_hash = obj.base_ref;
@@ -329,11 +341,15 @@ impl ClientStorage {
 
                 let base_obj = Self::read_pack_obj(pack_file, base_offset)?;
                 let base_obj = Arc::new(base_obj);
-                let new_obj = Pack::rebuild_delta(obj, base_obj);
-                Ok(new_obj)
+                Pack::rebuild_delta(obj, base_obj) // new obj
             },
-            _ => Ok(obj),
+            _ => obj,
+        };
+        // write cache
+        if PACK_OBJ_CACHE.lock().unwrap().insert(cache_key, full_obj.clone()).is_err() {
+            eprintln!("Warn: EntryTooLarge");
         }
+        Ok(full_obj)
     }
 }