odb upgrades

.github/workflows/ci.yml

@@ -218,6 +218,9 @@ jobs:
         env:
           GIX_TEST_IGNORE_ARCHIVES: "1"
         run: just ci-test
+      - name: Test dynamic ODB scenarios with contending handles
+        timeout-minutes: 5
+        run: just test-odb-threaded-scenarios
 
   test-doc:
     runs-on: ubuntu-latest

gitoxide-core/src/repository/odb.rs

@@ -171,6 +171,25 @@ pub fn statistics(
         }
     }
 
+    #[derive(Default)]
+    struct MissingObjects(bool);
+
+    impl gix::parallel::Reduce for MissingObjects {
+        type Input = Result<bool, anyhow::Error>;
+        type FeedProduce = ();
+        type Output = bool;
+        type Error = anyhow::Error;
+
+        fn feed(&mut self, missing: Self::Input) -> Result<Self::FeedProduce, Self::Error> {
+            self.0 |= missing?;
+            Ok(())
+        }
+
+        fn finalize(self) -> Result<Self::Output, Self::Error> {
+            Ok(self.0)
+        }
+    }
+
     let cancelled = || anyhow::anyhow!("Cancelled by user");
     let object_ids = repo.objects.iter()?.filter_map(Result::ok);
     let chunk_size = 1_000;
@@ -205,52 +224,60 @@ pub fn statistics(
             },
         )?
     } else {
-        if extra_header_lookup {
-            bail!("extra-header-lookup is only meaningful in threaded mode");
-        }
-        let mut stats = Statistics::default();
+        let mut stats = Statistics {
+            ids: extra_header_lookup.then(Vec::new),
+            ..Default::default()
+        };
 
         for (count, id) in object_ids.enumerate() {
             if count % chunk_size == 0 && gix::interrupt::is_triggered() {
                 return Err(cancelled());
             }
             stats.consume(repo.objects.header(id)?);
+            if let Some(ids) = stats.ids.as_mut() {
+                ids.push(id);
+            }
             progress.inc();
         }
         stats
     };
 
     progress.show_throughput(start);
 
-    if let Some(mut ids) = stats.ids.take() {
+    if let Some(ids) = stats.ids.take() {
         // Critical to re-open the repo to assure we don't have any ODB state and start fresh.
         let start = std::time::Instant::now();
         let repo = gix::open_opts(repo.git_dir(), repo.open_options().to_owned())?;
         progress.set_name("re-counting".into());
         progress.init(Some(ids.len()), gix::progress::count("objects"));
         let counter = progress.counter();
         counter.store(0, Ordering::Relaxed);
-        let errors = gix::parallel::in_parallel_with_slice(
-            &mut ids,
+        let has_errors = gix::parallel::in_parallel(
+            gix::features::iter::Chunks {
+                inner: ids.into_iter(),
+                size: chunk_size,
+            },
             thread_limit,
             {
                 let objects = repo.objects.clone();
-                move |_| (objects.clone().into_inner(), counter, false)
+                move |_| (objects.clone().into_inner(), counter)
             },
-            |id, (odb, counter, has_error), _threads_left, _stop_everything| -> anyhow::Result<()> {
-                counter.fetch_add(1, Ordering::Relaxed);
-                if let Err(_err) = odb.header(id) {
-                    *has_error = true;
-                    gix::trace::error!(err = ?_err, "Object that is known to be present wasn't found");
+            |ids, (odb, counter)| -> anyhow::Result<bool> {
+                counter.fetch_add(ids.len(), Ordering::Relaxed);
+                let mut has_error = false;
+                for id in ids {
+                    if let Err(_err) = odb.header(id) {
+                        has_error = true;
+                        gix::trace::error!(err = ?_err, "Object that is known to be present wasn't found");
+                    }
                 }
-                Ok(())
+                Ok(has_error)
             },
-            || Some(std::time::Duration::from_millis(100)),
-            |(_, _, has_error)| has_error,
+            MissingObjects::default(),
         )?;
 
         progress.show_throughput(start);
-        if errors.contains(&true) {
+        if has_errors {
             bail!("At least one object couldn't be looked up even though it must exist");
         }
     }
@@ -275,3 +302,30 @@ pub fn entries(repo: gix::Repository, format: OutputFormat, mut out: impl io::Wr
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn extra_header_lookup_works_with_one_or_multiple_threads() -> anyhow::Result<()> {
+        let dir = tempfile::tempdir()?;
+        let repo = gix::init_bare(dir.path())?;
+        repo.write_blob(b"an object to look up twice")?;
+
+        for threads in [1, 2] {
+            statistics(
+                repo.clone(),
+                gix::progress::Discard,
+                Vec::new(),
+                Vec::new(),
+                statistics::Options {
+                    format: OutputFormat::Human,
+                    thread_limit: Some(threads),
+                    extra_header_lookup: true,
+                },
+            )?;
+        }
+        Ok(())
+    }
+}

gix-odb/Cargo.toml

@@ -15,6 +15,11 @@ autotests = false
 [lib]
 doctest = true
 
+[[bench]]
+name = "dynamic"
+harness = false
+required-features = ["sha1", "parallel"]
+
 [features]
 ## Enable support for the SHA-1 hash by enabling the respective feature in the `gix-hash` crate.
 sha1 = ["gix-hash/sha1", "gix-pack/sha1"]
@@ -24,6 +29,8 @@ sha256 = ["gix-hash/sha256", "gix-pack/sha256"]
 serde = ["dep:serde", "gix-hash/serde", "gix-object/serde", "gix-pack/serde"]
 ## Enable support for multi-threaded usage of the object database.
 parallel = ["gix-features/parallel"]
+## Expose deterministic synchronization hooks for tests of concurrent object-database behavior.
+test-support = []
 
 [dependencies]
 gix-features = { version = "^0.49.0", path = "../gix-features", features = ["walkdir", "crc32"] }
@@ -46,7 +53,7 @@ memmap2 = "0.9.11"
 document-features = { version = "0.2.0", optional = true }
 
 [dev-dependencies]
-gix-odb = { path = ".", features = ["sha1", "sha256"] }
+gix-odb = { path = ".", features = ["sha1", "sha256", "test-support"] }
 gix-hash = { path = "../gix-hash" }
 gix-testtools = { path = "../tests/tools", default-features = false, features = ["sha1", "sha256"] }
 gix-date = { path = "../gix-date" }
@@ -56,6 +63,8 @@ pretty_assertions = "1.0.0"
 filetime = "0.2.29"
 maplit = "1.0.2"
 crossbeam-channel = "0.5.15"
+tar = { version = "0.4.46", default-features = false }
+criterion = "0.8.2"
 
 [[test]]
 name = "odb"