Try to use counters directly in delta-tree resolution

However, it seems slightly slower and doesn't necessarily improve the progress printing. Also, we have ot content over two counters, not only one, which might be noticable and is probably not worth it.

Sebastian Thiel committed 4 years ago

Commit c6ded261 parent ff1db662 files · +50 −51Browse files

Unified · Split

2 files

git-pack/src/cache/delta/traverse/mod.rs +37 −41
git-pack/src/cache/delta/traverse/resolve.rs +13 −10

git-pack/src/cache/delta/traverse/mod.rs+37−41

         Options {
             thread_limit,
             object_progress,
-            size_progress,
+            mut size_progress,
             should_interrupt,
             object_hash,
         }: Options<'_, P1, P2>,
         //       allowing threads to be more busy overall. This, however, needs some refactorings to allow operation
         //       on a single item efficiently while providing real-time feedback.
         let num_objects = self.num_items();
+        let object_counter = {
+            let mut progress = lock(&object_progress);
+            progress.init(Some(num_objects), progress::count("objects"));
+            progress.counter()
+        };
+        size_progress.init(None, progress::bytes());
+        let size_counter = size_progress.counter();
+        let start = std::time::Instant::now();
         in_parallel_if(
             should_run_in_parallel,
             self.iter_root_chunks(chunk_size),
                     )
                 }
             },
-            move |root_nodes, state| resolve::deltas(root_nodes, state, object_hash.len_in_bytes()),
+            {
-            Reducer::new(num_objects, object_progress, size_progress, should_interrupt),
+                let object_counter = object_counter.clone();
+                let size_counter = size_counter.clone();
+                move |root_nodes, state| {
+                    resolve::deltas(
+                        object_counter.clone(),
+                        size_counter.clone(),
+                        root_nodes,
+                        state,
+                        object_hash.len_in_bytes(),
+                    )
+                }
+            },
+            Reducer::new(should_interrupt),
         )?;
+        lock(&object_progress).show_throughput(start);
+        size_progress.show_throughput(start);
         Ok(Outcome {
             roots: self.root_items,
             children: self.child_items,
     }
 }
-struct Reducer<'a, P1, P2> {
+struct Reducer<'a> {
-    item_count: usize,
-    progress: OwnShared<Mutable<P1>>,
-    start: std::time::Instant,
-    size_progress: P2,
     should_interrupt: &'a AtomicBool,
 }
-impl<'a, P1, P2> Reducer<'a, P1, P2>
+impl<'a> Reducer<'a> {
-where
+    pub fn new(should_interrupt: &'a AtomicBool) -> Self {
-    P1: Progress,
+        Reducer { should_interrupt }
-    P2: Progress,
-{
-    pub fn new(
-        num_objects: usize,
-        progress: OwnShared<Mutable<P1>>,
-        mut size_progress: P2,
-        should_interrupt: &'a AtomicBool,
-    ) -> Self {
-        lock(&progress).init(Some(num_objects), progress::count("objects"));
-        size_progress.init(None, progress::bytes());
-        Reducer {
-            item_count: 0,
-            progress,
-            start: std::time::Instant::now(),
-            size_progress,
-            should_interrupt,
-        }
     }
 }
-impl<'a, P1, P2> parallel::Reduce for Reducer<'a, P1, P2>
+impl<'a> parallel::Reduce for Reducer<'a> {
-where
+    type Input = Result<(), Error>;
-    P1: Progress,
-    P2: Progress,
-{
-    type Input = Result<(usize, u64), Error>;
     type FeedProduce = ();
     type Output = ();
     type Error = Error;
     fn feed(&mut self, input: Self::Input) -> Result<(), Self::Error> {
-        let (num_objects, decompressed_size) = input?;
+        input?;
-        self.item_count += num_objects;
-        self.size_progress.inc_by(decompressed_size as usize);
-        lock(&self.progress).set(self.item_count);
         if self.should_interrupt.load(Ordering::SeqCst) {
             return Err(Error::Interrupted);
         }
         Ok(())
     }
-    fn finalize(mut self) -> Result<Self::Output, Self::Error> {
+    fn finalize(self) -> Result<Self::Output, Self::Error> {
-        lock(&self.progress).show_throughput(self.start);
-        self.size_progress.show_throughput(self.start);
         Ok(())
     }
 }

git-pack/src/cache/delta/traverse/resolve.rs+13−10

+use std::sync::atomic::Ordering;
 use std::{cell::RefCell, collections::BTreeMap};
 use git_features::{
 };
 pub(crate) fn deltas<T, F, P, MBFN, S, E>(
+    object_counter: Option<git_features::progress::StepShared>,
+    size_counter: Option<git_features::progress::StepShared>,
     nodes: crate::cache::delta::Chunk<'_, T>,
     (bytes_buf, ref mut progress, state, resolve, modify_base): &mut (Vec<u8>, P, S, F, MBFN),
     hash_len: usize,
-) -> Result<(usize, u64), Error>
+) -> Result<(), Error>
 where
     F: for<'r> Fn(EntryRange, &'r mut Vec<u8>) -> Option<()>,
     P: Progress,
 {
     let mut decompressed_bytes_by_pack_offset = BTreeMap::new();
     let bytes_buf = RefCell::new(bytes_buf);
-    let mut num_objects = 0;
-    let mut decompressed_bytes: u64 = 0;
     let decompress_from_resolver = |slice: EntryRange| -> Result<(crate::data::Entry, u64, Vec<u8>), Error> {
         let mut bytes_buf = bytes_buf.borrow_mut();
         bytes_buf.resize((slice.end - slice.start) as usize, 0);
                 },
             )
             .map_err(|err| Box::new(err) as Box<dyn std::error::Error + Send + Sync>)?;
-            num_objects += 1;
+            object_counter.as_ref().map(|c| c.fetch_add(1, Ordering::SeqCst));
-            decompressed_bytes += base_bytes.len() as u64;
+            size_counter
-            progress.inc();
+                .as_ref()
+                .map(|c| c.fetch_add(base_bytes.len(), Ordering::SeqCst));
         }
         for mut child in base.into_child_iter() {
                     },
                 )
                 .map_err(|err| Box::new(err) as Box<dyn std::error::Error + Send + Sync>)?;
-                num_objects += 1;
+                object_counter.as_ref().map(|c| c.fetch_add(1, Ordering::SeqCst));
-                decompressed_bytes += fully_resolved_delta_bytes.len() as u64;
+                size_counter
-                progress.inc();
+                    .as_ref()
+                    .map(|c| c.fetch_add(base_bytes.len(), Ordering::SeqCst));
             }
         }
     }
-    Ok((num_objects, decompressed_bytes))
+    Ok(())
 }
 fn decompress_all_at_once(b: &[u8], decompressed_len: usize) -> Result<Vec<u8>, Error> {