From 1cab4425b76db1343f35c2a5d3295b6ef43697af Mon Sep 17 00:00:00 2001 From: HouXiaoxuan Date: Sun, 29 Sep 2024 00:48:23 +0800 Subject: [PATCH 1/2] [mercury] Adjust function parameters and logic in preparation for parallel encode 1. Migrate send data from function `encode_one_object` to after return 2. `try_as_offset_delta` use &mut instead of clone and return. 3. Migrate invoke `try_as_offset_delta` to `encode` instand of `encode_one_object` Signed-off-by: HouXiaoxuan --- mercury/src/internal/pack/encode.rs | 119 ++++++++++++++-------------- 1 file changed, 61 insertions(+), 58 deletions(-) diff --git a/mercury/src/internal/pack/encode.rs b/mercury/src/internal/pack/encode.rs index bdc7b0bbe..1d00f3bbb 100644 --- a/mercury/src/internal/pack/encode.rs +++ b/mercury/src/internal/pack/encode.rs @@ -58,6 +58,54 @@ fn encode_offset(mut value: usize) -> Vec { bytes } +/// Encode one object, and update the hash +/// @offset: offset of this object if it's a delta object. For other object, it's None +fn encode_one_object(entry: &Entry, offset: Option) -> Result, GitError> { + // try encode as delta + let obj_data = &entry.data; + let obj_data_len = obj_data.len(); + let obj_type_number = entry.obj_type.to_u8(); + + let mut encoded_data = Vec::new(); + + // **header** encoding + let mut header_data = vec![(0x80 | (obj_type_number << 4)) + (obj_data_len & 0x0f) as u8]; + let mut size = obj_data_len >> 4; // 4 bit has been used in first byte + if size > 0 { + while size > 0 { + if size >> 7 > 0 { + header_data.push((0x80 | size) as u8); + size >>= 7; + } else { + header_data.push(size as u8); + break; + } + } + } else { + header_data.push(0); + } + encoded_data.extend(header_data); + + // **offset** encoding + if entry.obj_type == ObjectType::OffsetDelta { + let offset_data = encode_offset(offset.unwrap()); + encoded_data.extend(offset_data); + } else if entry.obj_type == ObjectType::HashDelta { + unreachable!("unsupported type") + } + + // **data** encoding, need zlib compress + let mut inflate = ZlibEncoder::new(Vec::new(), flate2::Compression::default()); + inflate + .write_all(obj_data) + .expect("zlib compress should never failed"); + inflate.flush().expect("zlib flush should never failed"); + let compressed_data = inflate.finish().expect("zlib compress should never failed"); + // self.write_all_and_update(&compressed_data).await; + encoded_data.extend(compressed_data); + Ok(encoded_data) +} + impl PackEncoder { pub fn new(object_number: usize, window_size: usize, sender: mpsc::Sender>) -> Self { PackEncoder { @@ -112,7 +160,11 @@ impl PackEncoder { self.process_index += 1; // push window after encode to void diff by self let offset = self.inner_offset; - self.encode_one_object(&entry).await?; + let mut try_delta_entry = entry.clone(); + let try_delfa_offset = self.try_as_offset_delta(&mut try_delta_entry); + let obj_data = encode_one_object(&try_delta_entry, try_delfa_offset)?; + + self.write_all_and_update(&obj_data).await; self.window.push_back((entry, offset)); if self.window.len() > self.window_size { self.window.pop_front(); @@ -140,9 +192,9 @@ impl PackEncoder { /// Try to encode as delta using objects in window /// # Returns - /// - Return (delta entry, offset) if success make delta - /// - Return (origin Entry,None) if didn't delta, - fn try_as_offset_delta(&mut self, entry: &Entry) -> (Entry, Option) { + /// - Return (offset) if success make delta + /// - Return (None) if didn't delta, + fn try_as_offset_delta(&mut self, entry: &mut Entry) -> Option { let mut best_base: Option<&(Entry, usize)> = None; let mut best_rate: f64 = 0.0; for try_base in self.window.iter() { @@ -159,16 +211,12 @@ impl PackEncoder { let best_base = best_base.unwrap(); // must some if best rate > 0 let delta = delta::encode(&best_base.0.data, &entry.data); let offset = self.inner_offset - best_base.1; - ( - Entry { - data: delta, - obj_type: ObjectType::OffsetDelta, - ..entry.clone() - }, - Some(offset), - ) + entry.obj_type = ObjectType::OffsetDelta; + entry.data = delta; + + Some(offset) } else { - (entry.clone(), None) + None } } @@ -179,51 +227,6 @@ impl PackEncoder { self.send_data(data.to_vec()).await; } - /// Encode one object, and update the hash - async fn encode_one_object(&mut self, entry: &Entry) -> Result<(), GitError> { - // try encode as delta - let (entry, offset) = self.try_as_offset_delta(entry); - let obj_data = entry.data; - let obj_data_len = obj_data.len(); - let obj_type_number = entry.obj_type.to_u8(); - - // **header** encoding - let mut header_data = vec![(0x80 | (obj_type_number << 4)) + (obj_data_len & 0x0f) as u8]; - let mut size = obj_data_len >> 4; // 4 bit has been used in first byte - if size > 0 { - while size > 0 { - if size >> 7 > 0 { - header_data.push((0x80 | size) as u8); - size >>= 7; - } else { - header_data.push(size as u8); - break; - } - } - } else { - header_data.push(0); - } - self.write_all_and_update(&header_data).await; - - // **offset** encoding - if entry.obj_type == ObjectType::OffsetDelta { - let offset_data = encode_offset(offset.unwrap()); - self.write_all_and_update(&offset_data).await; - } else if entry.obj_type == ObjectType::HashDelta { - unreachable!("unsupported type") - } - - // **data** encoding, need zlib compress - let mut inflate = ZlibEncoder::new(Vec::new(), flate2::Compression::default()); - inflate - .write_all(&obj_data) - .expect("zlib compress should never failed"); - inflate.flush().expect("zlib flush should never failed"); - let compressed_data = inflate.finish().expect("zlib compress should never failed"); - self.write_all_and_update(&compressed_data).await; - Ok(()) - } - /// async version of encode, result data will be returned by JoinHandle. /// It will consume PackEncoder, so you can't use it after calling this function. pub async fn encode_async( From d60f39cefbbc397a369b560a2c7f794e926f4009 Mon Sep 17 00:00:00 2001 From: HouXiaoxuan Date: Sun, 29 Sep 2024 15:48:41 +0800 Subject: [PATCH 2/2] [mercury] feat: use a parallel version of encode when delta window size is zero. Signed-off-by: HouXiaoxuan --- mercury/Cargo.toml | 1 + mercury/src/internal/pack/decode.rs | 18 +-- mercury/src/internal/pack/encode.rs | 217 ++++++++++++++++++++++++++-- mercury/src/internal/pack/mod.rs | 35 ++++- 4 files changed, 231 insertions(+), 40 deletions(-) diff --git a/mercury/Cargo.toml b/mercury/Cargo.toml index 0ce83935a..94c9d4de8 100644 --- a/mercury/Cargo.toml +++ b/mercury/Cargo.toml @@ -35,6 +35,7 @@ futures-util = { workspace = true } bytes = { workspace = true } axum = { workspace = true } memchr = { workspace = true } +rayon = "1.10.0" [target.'cfg(windows)'.dependencies] # only on Windows mimalloc = "0.1.39" # avoid sticking on dropping on Windows diff --git a/mercury/src/internal/pack/decode.rs b/mercury/src/internal/pack/decode.rs index bd7e3c712..ac723c388 100644 --- a/mercury/src/internal/pack/decode.rs +++ b/mercury/src/internal/pack/decode.rs @@ -633,27 +633,11 @@ mod tests { use flate2::write::ZlibEncoder; use flate2::Compression; use tokio_util::io::ReaderStream; - use tracing_subscriber::util::SubscriberInitExt; + use crate::internal::pack::tests::init_logger; use crate::internal::pack::Pack; use futures_util::TryStreamExt; - fn init_logger() { - let _ = tracing_subscriber::fmt::Subscriber::builder() - .with_target(false) - .without_time() - .finish() - .try_init();// avoid multi-init - - // CAUTION: This two is same - // 1. - // tracing_subscriber::fmt().init(); - // - // 2. - // env::set_var("RUST_LOG", "debug"); // must be set if use `fmt::init()`, or no output - // tracing_subscriber::fmt::init(); - } - #[test] fn test_pack_check_header() { let mut source = PathBuf::from(env::current_dir().unwrap().parent().unwrap()); diff --git a/mercury/src/internal/pack/encode.rs b/mercury/src/internal/pack/encode.rs index 1d00f3bbb..232f6764d 100644 --- a/mercury/src/internal/pack/encode.rs +++ b/mercury/src/internal/pack/encode.rs @@ -2,11 +2,13 @@ use std::collections::VecDeque; use std::io::Write; use flate2::write::ZlibEncoder; +use rayon::prelude::*; use sha1::{Digest, Sha1}; use tokio::sync::mpsc; use tokio::task::JoinHandle; use crate::internal::object::types::ObjectType; +use crate::time_it; use crate::{errors::GitError, hash::SHA1, internal::pack::entry::Entry}; const MIN_DELTA_RATE: f64 = 0.5; // minimum delta rate can accept @@ -190,6 +192,78 @@ impl PackEncoder { Ok(()) } + /// Parallel encode with rayon, only works when window_size == 0 (no delta) + pub async fn parallel_encode( + &mut self, + mut entry_rx: mpsc::Receiver, + ) -> Result<(), GitError> { + if self.window_size != 0 { + return Err(GitError::PackEncodeError( + "parallel encode only works when window_size == 0".to_string(), + )); + } + + let head = encode_header(self.object_number); + self.send_data(head.clone()).await; + self.inner_hash.update(&head); + + // ensure only one decode can only invoke once + if self.start_encoding { + return Err(GitError::PackEncodeError( + "encoding operation is already in progress".to_string(), + )); + } + + let batch_size = usize::max(1000, entry_rx.max_capacity() / 10); // A temporary value, not optimized + tracing::info!("encode with batch size: {}", batch_size); + loop { + let mut batch_entries = Vec::with_capacity(batch_size); + time_it!("parallel encode: receive batch", { + for _ in 0..batch_size { + match entry_rx.recv().await { + Some(entry) => { + batch_entries.push(entry); + self.process_index += 1; + } + None => break, + } + } + }); + + if batch_entries.is_empty() { + break; + } + + // use `collect` will return result in order, refs: https://github.com/rayon-rs/rayon/issues/551#issuecomment-371657900 + let batch_result: Vec> = time_it!("parallel encode: encode batch", { + batch_entries + .par_iter() + .map(|entry| encode_one_object(entry, None).unwrap()) + .collect() + }); + + time_it!("parallel encode: write batch", { + for obj_data in batch_result { + self.write_all_and_update(&obj_data).await; + } + }); + } + + if self.process_index != self.object_number { + panic!( + "not all objects are encoded, process:{}, total:{}", + self.process_index, self.object_number + ); + } + + // hash signature + let hash_result = self.inner_hash.clone().finalize(); + self.final_hash = Some(SHA1::from_bytes(&hash_result)); + self.send_data((hash_result).to_vec()).await; + self.drop_sender(); + Ok(()) + } + /// Try to encode as delta using objects in window /// # Returns /// - Return (offset) if success make delta @@ -233,19 +307,44 @@ impl PackEncoder { mut self, rx: mpsc::Receiver, ) -> Result, GitError> { - Ok(tokio::spawn(async move { self.encode(rx).await.unwrap() })) + Ok(tokio::spawn(async move { + if self.window_size == 0 { + self.parallel_encode(rx).await.unwrap() + } else { + self.encode(rx).await.unwrap() + } + })) } } #[cfg(test)] mod tests { + use std::env; + use std::sync::Arc; + // use tokio::{Arc, Mutex}; use std::{io::Cursor, path::PathBuf}; + use tokio::sync::Mutex; + use crate::internal::object::blob::Blob; - use crate::internal::pack::Pack; + use crate::internal::pack::{tests::init_logger, Pack}; + use crate::time_it; use super::*; + fn check_format(data: &Vec) { + let mut p = Pack::new( + None, + Some(1024 * 1024 * 1024 * 6), // 6GB + Some(PathBuf::from("/tmp/.cache_temp")), + true, + ); + let mut reader = Cursor::new(data); + tracing::debug!("start check format"); + p.decode(&mut reader, |_, _| {}) + .expect("pack file format error"); + } + #[tokio::test] async fn test_pack_encoder() { async fn encode_once(window_size: usize) -> Vec { @@ -271,26 +370,114 @@ mod tests { result } - fn check_format(data: Vec) { - let mut p = Pack::new( - None, - Some(1024 * 20), - Some(PathBuf::from("/tmp/.cache_temp")), - true, - ); - let mut reader = Cursor::new(data); - p.decode(&mut reader, |_, _| {}) - .expect("pack file format error"); - } // without delta let pack_without_delta = encode_once(0).await; let pack_without_delta_size = pack_without_delta.len(); - check_format(pack_without_delta); + check_format(&pack_without_delta); // with delta let pack_with_delta = encode_once(3).await; assert_ne!(pack_with_delta.len(), pack_without_delta_size); - check_format(pack_with_delta); + check_format(&pack_with_delta); + } + + async fn get_entries_for_test() -> Arc>> { + let mut source = PathBuf::from(env::current_dir().unwrap().parent().unwrap()); + source.push("tests/data/packs/git-2d187177923cd618a75da6c6db45bb89d92bd504.pack"); + + // decode pack file to get entries + let mut p = Pack::new( + None, + Some(1024 * 1024 * 1024 * 6), + Some(PathBuf::from("/tmp/.cache_temp")), + true, + ); + + let f = std::fs::File::open(source).unwrap(); + tracing::info!("pack file size: {}", f.metadata().unwrap().len()); + let mut reader = std::io::BufReader::new(f); + let entries = Arc::new(Mutex::new(Vec::new())); + let entries_clone = entries.clone(); + p.decode(&mut reader, move |entry, _| { + let mut entries = entries_clone.blocking_lock(); + entries.push(entry); + }) + .unwrap(); + assert_eq!(p.number, entries.lock().await.len()); + tracing::info!("total entries: {}", p.number); + drop(p); + + entries + } + + #[tokio::test] + async fn test_pack_encoder_parallel_large_file() { + init_logger(); + let entries = get_entries_for_test().await; + let entries_number = entries.lock().await.len(); + + // encode entries with parallel + let (tx, mut rx) = mpsc::channel(1_000_000); + let (entry_tx, entry_rx) = mpsc::channel::(1_000_000); + + let mut encoder = PackEncoder::new(entries_number, 0, tx); + tokio::spawn(async move { + time_it!("test parallel encode", { + encoder.parallel_encode(entry_rx).await.unwrap(); + }); + }); + + // spawn a task to send entries + tokio::spawn(async move { + let entries = entries.lock().await; + for entry in entries.iter() { + entry_tx.send(entry.clone()).await.unwrap(); + } + drop(entry_tx); + tracing::info!("all entries sent"); + }); + + let mut result = Vec::new(); + while let Some(chunk) = rx.recv().await { + result.extend(chunk); + } + tracing::info!("new pack file size: {}", result.len()); + + // check format + check_format(&result); + } + + #[tokio::test] + async fn test_pack_encoder_large_file() { + init_logger(); + let entries = get_entries_for_test().await; + let entries_number = entries.lock().await.len(); + + // encode entries + let (tx, mut rx) = mpsc::channel(100_000); + let (entry_tx, entry_rx) = mpsc::channel::(100_000); + + let mut encoder = PackEncoder::new(entries_number, 0, tx); + tokio::spawn(async move { + time_it!("test encode no parallel", { + encoder.encode(entry_rx).await.unwrap(); + }); + }); + + // spawn a task to send entries + tokio::spawn(async move { + let entries = entries.lock().await; + for entry in entries.iter() { + entry_tx.send(entry.clone()).await.unwrap(); + } + drop(entry_tx); + tracing::info!("all entries sent"); + }); + + // only receive data + while (rx.recv().await).is_some() { + // do nothing + } } #[test] diff --git a/mercury/src/internal/pack/mod.rs b/mercury/src/internal/pack/mod.rs index aeab750a7..3f942ca2a 100644 --- a/mercury/src/internal/pack/mod.rs +++ b/mercury/src/internal/pack/mod.rs @@ -2,24 +2,24 @@ //! ## Reference //! 1. Git Pack-Format [Introduce](https://git-scm.com/docs/pack-format) //! +pub mod cache; +pub mod cache_object; +pub mod channel_reader; pub mod decode; pub mod encode; -pub mod wrapper; +pub mod entry; pub mod utils; -pub mod cache; pub mod waitlist; -pub mod cache_object; -pub mod entry; -pub mod channel_reader; +pub mod wrapper; -use threadpool::ThreadPool; -use std::sync::Arc; use std::sync::atomic::AtomicUsize; +use std::sync::Arc; +use threadpool::ThreadPool; use crate::hash::SHA1; use crate::internal::object::ObjectTrait; -use crate::internal::pack::waitlist::Waitlist; use crate::internal::pack::cache::Caches; +use crate::internal::pack::waitlist::Waitlist; const DEFAULT_TMP_DIR: &str = "./.cache_temp"; pub struct Pack { @@ -36,4 +36,23 @@ pub struct Pack { #[cfg(test)] mod tests { + use tracing_subscriber::util::SubscriberInitExt; + + pub(crate) fn init_logger() { + let _ = tracing_subscriber::fmt::Subscriber::builder() + .with_target(false) + .without_time() + .with_level(true) + .with_max_level(tracing::Level::DEBUG) + .finish() + .try_init(); // avoid multi-init + + // CAUTION: This two is same + // 1. + // tracing_subscriber::fmt().init(); + // + // 2. + // env::set_var("RUST_LOG", "debug"); // must be set if use `fmt::init()`, or no output + // tracing_subscriber::fmt::init(); + } }