fix: more robustness in the face of a trampling-herd of threads loading a single index.

The motivating example is here: https://github.com/praetorian-inc/noseyparker/issues/179 Previously, it was possible for a trampling herd of threads to consolidate the disk state. Most of them would be 'needs-init' threads which could notice that the initialization already happened, and just use that. But a thread might be late for the party and somehow manages to not get any newly loaded index, and thus tries to consolidate with what's on disk again. Then it would again determine no change, and return nothing, causing the caller to abort and not find objects it should find because it wouldn't see the index that it should have seen. The reason the thread got into this mess is that the 'is-load-ongoing' flagging was racy itself, so it would not wait for ongoing loads and just conclude nothing happened. An extra delay (by yielding) now assures it either seees the loading state and waits for it, sees the newly loaded indices. Note that this issue can be reproduced with: ``` './target/release/gix -r repo-with-one-pack -t10 --trace odb stats --extra-header-lookup' ```

Sebastian Thiel committed 2 years ago

Commit addf4461 parent 7b3dc925 files · +142 −63Browse files

Unified · Split

5 files

gitoxide-core/src/repository/odb.rs +67 −9
gix-odb/src/store_impls/dynamic/load_index.rs +62 −49
gix-odb/src/store_impls/dynamic/types.rs +2 −2
src/plumbing/main.rs +6 −2
src/plumbing/options/mod.rs +5 −1

gitoxide-core/src/repository/odb.rs+67−9

 use std::io;
+use std::sync::atomic::Ordering;
 use anyhow::bail;
     pub struct Options {
         pub format: OutputFormat,
         pub thread_limit: Option<usize>,
+        /// A debug-flag that triggers looking up the headers of all objects again, but without indices preloaded
+        pub extra_header_lookup: bool,
     }
 }
     mut progress: impl gix::Progress,
     out: impl io::Write,
     mut err: impl io::Write,
-    statistics::Options { format, thread_limit }: statistics::Options,
+    statistics::Options {
+        format,
+        thread_limit,
+        extra_header_lookup,
+    }: statistics::Options,
 ) -> anyhow::Result<()> {
     use bytesize::ByteSize;
     use gix::odb::{find, HeaderExt};
     #[cfg_attr(feature = "serde", derive(serde::Serialize))]
     #[derive(Default)]
     struct Statistics {
+        /// All objects that were used to produce these statistics.
+        /// Only `Some` if we are doing an extra round of header queries on a repository without loaded indices.
+        #[cfg_attr(feature = "serde", serde(skip_serializing))]
+        ids: Option<Vec<gix::ObjectId>>,
         total_objects: usize,
         loose_objects: usize,
         packed_objects: usize,
     }
     impl gix::parallel::Reduce for Reduce {
-        type Input = Result<Vec<gix::odb::find::Header>, anyhow::Error>;
+        type Input = Result<Vec<(gix::ObjectId, gix::odb::find::Header)>, anyhow::Error>;
         type FeedProduce = ();
         type Output = Statistics;
         type Error = anyhow::Error;
         fn feed(&mut self, items: Self::Input) -> Result<Self::FeedProduce, Self::Error> {
-            for item in items? {
+            for (id, item) in items? {
                 self.stats.consume(item);
+                if let Some(ids) = self.stats.ids.as_mut() {
+                    ids.push(id);
+                }
             }
             Ok(())
         }
     }
     let cancelled = || anyhow::anyhow!("Cancelled by user");
-    let object_ids = repo.objects.store_ref().iter()?.filter_map(Result::ok);
+    let object_ids = repo.objects.iter()?.filter_map(Result::ok);
     let chunk_size = 1_000;
-    let stats = if gix::parallel::num_threads(thread_limit) > 1 {
+    let mut stats = if gix::parallel::num_threads(thread_limit) > 1 {
         gix::parallel::in_parallel(
             gix::interrupt::Iter::new(
                 gix::features::iter::Chunks {
                 cancelled,
             ),
             thread_limit,
-            move |_| (repo.objects.clone().into_inner(), counter),
+            {
+                let objects = repo.objects.clone();
+                move |_| (objects.clone().into_inner(), counter)
+            },
             |ids, (handle, counter)| {
                 let ids = ids?;
-                counter.fetch_add(ids.len(), std::sync::atomic::Ordering::Relaxed);
+                counter.fetch_add(ids.len(), Ordering::Relaxed);
                 let out = ids
                     .into_iter()
-                    .map(|id| handle.header(id))
+                    .map(|id| handle.header(id).map(|hdr| (id, hdr)))
                     .collect::<Result<Vec<_>, _>>()?;
                 Ok(out)
             },
-            Reduce::default(),
+            Reduce {
+                stats: Statistics {
+                    ids: extra_header_lookup.then(Vec::new),
+                    ..Default::default()
+                },
+            },
         )?
     } else {
+        if extra_header_lookup {
+            bail!("extra-header-lookup is only meaningful in threaded mode");
+        }
         let mut stats = Statistics::default();
         for (count, id) in object_ids.enumerate() {
     progress.show_throughput(start);
+    if let Some(mut 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,
+            thread_limit,
+            {
+                let objects = repo.objects.clone();
+                move |_| (objects.clone().into_inner(), counter, false)
+            },
+            |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");
+                }
+                Ok(())
+            },
+            || Some(std::time::Duration::from_millis(100)),
+            |(_, _, has_error)| has_error,
+        )?;
+        progress.show_throughput(start);
+        if errors.contains(&true) {
+            bail!("At least one object couldn't be looked up even though it must exist");
+        }
+    }
     #[cfg(feature = "serde")]
     {
         serde_json::to_writer_pretty(out, &stats)?;

gix-odb/src/store_impls/dynamic/load_index.rs+62−49

     ops::Deref,
     path::{Path, PathBuf},
     sync::{
-        atomic::{AtomicU16, AtomicUsize, Ordering},
+        atomic::{AtomicU16, Ordering},
         Arc,
     },
     time::SystemTime,
             Ok(Some(self.collect_snapshot()))
         } else {
             // always compare to the latest state
-            // Nothing changed in the mean time, try to load another index…
+            // Nothing changed in the meantime, try to load another index…
             if self.load_next_index(index) {
                 Ok(Some(self.collect_snapshot()))
             } else {
                         let slot = &self.files[index.slot_indices[slot_map_index]];
                         let _lock = slot.write.lock();
                         if slot.generation.load(Ordering::SeqCst) > index.generation {
-                            // There is a disk consolidation in progress which just overwrote a slot that cold be disposed with some other
+                            // There is a disk consolidation in progress which just overwrote a slot that could be disposed with some other
                             // index, one we didn't intend to load.
                             // Continue with the next slot index in the hope there is something else we can do…
                             continue 'retry_with_next_slot_index;
                         let bundle_mut = Arc::make_mut(&mut bundle);
                         if let Some(files) = bundle_mut.as_mut() {
                             // these are always expected to be set, unless somebody raced us. We handle this later by retrying.
-                            let _loaded_count = IncOnDrop(&index.loaded_indices);
+                            let res = {
-                            match files.load_index(self.object_hash) {
+                                let res = files.load_index(self.object_hash);
+                                slot.files.store(bundle);
+                                index.loaded_indices.fetch_add(1, Ordering::SeqCst);
+                                res
+                            };
+                            match res {
                                 Ok(_) => {
-                                    slot.files.store(bundle);
                                     break 'retry_with_next_slot_index;
                                 }
-                                Err(_) => {
+                                Err(_err) => {
-                                    slot.files.store(bundle);
+                                    gix_features::trace::error!(err=?_err, "Failed to load index file - some objects may seem to not exist");
                                     continue 'retry_with_next_slot_index;
                                 }
                             }
                         // There can be contention as many threads start working at the same time and take all the
                         // slots to load indices for. Some threads might just be left-over and have to wait for something
                         // to change.
-                        let num_load_operations = index.num_indices_currently_being_loaded.deref();
                         // TODO: potentially hot loop - could this be a condition variable?
-                        while num_load_operations.load(Ordering::Relaxed) != 0 {
+                        // This is a timing-based fix for the case that the `num_indices_being_loaded` isn't yet incremented,
+                        // and we might break out here without actually waiting for the loading operation. Then we'd fail to
+                        // observe a change and the underlying handler would not have all the indices it needs at its disposal.
+                        // Yielding means we will definitely loose enough time to observe the ongoing operation,
+                        // or its effects.
+                        std::thread::yield_now();
+                        while index.num_indices_currently_being_loaded.load(Ordering::SeqCst) != 0 {
                             std::thread::yield_now()
                         }
                         break 'retry_with_next_slot_index;
         // We might not be able to detect by pointer if the state changed, as this itself is racy. So we keep track of double-initialization
         // using a flag, which means that if `needs_init` was true we saw the index uninitialized once, but now that we are here it's
-        // initialized meaning that somebody was faster and we couldn't detect it by comparisons to the index.
+        // initialized meaning that somebody was faster, and we couldn't detect it by comparisons to the index.
         // If so, make sure we collect the snapshot instead of returning None in case nothing actually changed, which is likely with a
         // race like this.
         if !was_uninitialized && needs_init {
                     // generation stays the same, as it's the same value still but scheduled for eventual removal.
                 }
             } else {
+                // set the generation before we actually change the value, otherwise readers of old generations could observe the new one.
+                // We rather want them to turn around here and update their index, which, by that time, might actually already be available.
+                // If not, they would fail unable to load a pack or index they need, but that's preferred over returning wrong objects.
+                // Safety: can't race as we hold the lock, have to set the generation beforehand to help avoid others to observe the value.
+                slot.generation.store(generation, Ordering::SeqCst);
                 *files_mut = None;
             };
             slot.files.store(files);
-            if !needs_stable_indices {
-                // Not racy due to lock, generation must be set after unsetting the slot value AND storing it.
-                slot.generation.store(generation, Ordering::SeqCst);
-            }
         }
         let new_index = self.index.load();
         Ok(if index.state_id() == new_index.state_id() {
-            // there was no change, and nothing was loaded in the meantime, reflect that in the return value to not get into loops
+            // there was no change, and nothing was loaded in the meantime, reflect that in the return value to not get into loops.
             None
         } else {
             if load_new_index {
     }
     pub(crate) fn collect_snapshot(&self) -> Snapshot {
+        // We don't observe changes-on-disk in our 'wait-for-load' loop.
+        // That loop is meant to help assure the marker (which includes the amount of loaded indices) matches
+        // the actual amount of indices we collect.
         let index = self.index.load();
-        let indices = if index.is_initialized() {
+        loop {
-            index
+            if index.num_indices_currently_being_loaded.deref().load(Ordering::SeqCst) != 0 {
-                .slot_indices
+                std::thread::yield_now();
-                .iter()
+                continue;
-                .map(|idx| (*idx, &self.files[*idx]))
+            }
-                .filter_map(|(id, file)| {
+            let marker = index.marker();
-                    let lookup = match (**file.files.load()).as_ref()? {
+            let indices = if index.is_initialized() {
-                        types::IndexAndPacks::Index(bundle) => handle::SingleOrMultiIndex::Single {
+                index
-                            index: bundle.index.loaded()?.clone(),
+                    .slot_indices
-                            data: bundle.data.loaded().cloned(),
+                    .iter()
-                        },
+                    .map(|idx| (*idx, &self.files[*idx]))
-                        types::IndexAndPacks::MultiIndex(multi) => handle::SingleOrMultiIndex::Multi {
+                    .filter_map(|(id, file)| {
-                            index: multi.multi_index.loaded()?.clone(),
+                        let lookup = match (**file.files.load()).as_ref()? {
-                            data: multi.data.iter().map(|f| f.loaded().cloned()).collect(),
+                            types::IndexAndPacks::Index(bundle) => handle::SingleOrMultiIndex::Single {
-                        },
+                                index: bundle.index.loaded()?.clone(),
-                    };
+                                data: bundle.data.loaded().cloned(),
-                    handle::IndexLookup { file: lookup, id }.into()
+                            },
-                })
+                            types::IndexAndPacks::MultiIndex(multi) => handle::SingleOrMultiIndex::Multi {
-                .collect()
+                                index: multi.multi_index.loaded()?.clone(),
-        } else {
+                                data: multi.data.iter().map(|f| f.loaded().cloned()).collect(),
-            Vec::new()
+                            },
-        };
+                        };
+                        handle::IndexLookup { file: lookup, id }.into()
+                    })
+                    .collect()
+            } else {
+                Vec::new()
+            };
-        Snapshot {
+            return Snapshot {
-            indices,
+                indices,
-            loose_dbs: Arc::clone(&index.loose_dbs),
+                loose_dbs: Arc::clone(&index.loose_dbs),
-            marker: index.marker(),
+                marker,
+            };
         }
     }
 }
     }
 }
-struct IncOnDrop<'a>(&'a AtomicUsize);
-impl<'a> Drop for IncOnDrop<'a> {
-    fn drop(&mut self) {
-        self.0.fetch_add(1, Ordering::SeqCst);
-    }
-}
 pub(crate) enum Either {
     IndexPath(PathBuf),
     MultiIndexFile(Arc<gix_pack::multi_index::File>),

gix-odb/src/store_impls/dynamic/types.rs+2−2

@@ -18,7 +18,7 @@ pub(crate) type AtomicGeneration = AtomicU32; · 17 unmodified lines
18		18
19	/// A way to indicate which pack indices we have seen already and which of them are loaded, along with an idea	19	/// A way to indicate which pack indices we have seen already and which of them are loaded, along with an idea
20	/// of whether stored `PackId`s are still usable.	20	/// of whether stored `PackId`s are still usable.
21	#[derive(Default, Copy, Clone)]	21	#[derive(Default, Copy, Clone, Debug)]
22	pub struct SlotIndexMarker {	22	pub struct SlotIndexMarker {
23	/// The generation the `loaded_until_index` belongs to. Indices of different generations are completely incompatible.	23	/// The generation the `loaded_until_index` belongs to. Indices of different generations are completely incompatible.
24	/// This value changes once the internal representation is compacted, something that may happen only if there is no handle	24	/// This value changes once the internal representation is compacted, something that may happen only if there is no handle
@@ -262,7 +262,7 @@ impl IndexAndPacks { · 237 unmodified lines
262	}	262	}
263	}	263	}
264		264
265	/// If we are garbage, put ourselves into the loaded state. Otherwise put ourselves back to unloaded.	265	/// If we are garbage, put ourselves into the loaded state. Otherwise, put ourselves back to unloaded.
266	pub(crate) fn put_back(&mut self) {	266	pub(crate) fn put_back(&mut self) {
267	match self {	267	match self {
268	IndexAndPacks::Index(bundle) => {	268	IndexAndPacks::Index(bundle) => {

src/plumbing/main.rs+6−2

             ),
         },
         Subcommands::Odb(cmd) => match cmd {
-            odb::Subcommands::Stats => prepare_and_run(
+            odb::Subcommands::Stats { extra_header_lookup } => prepare_and_run(
                 "odb-stats",
                 trace,
                 auto_verbose,
                         progress,
                         out,
                         err,
-                        core::repository::odb::statistics::Options { format, thread_limit },
+                        core::repository::odb::statistics::Options {
+                            format,
+                            thread_limit,
+                            extra_header_lookup,
+                        },
                     )
                 },
             ),

src/plumbing/options/mod.rs+5−1

         Info,
         /// Count and obtain information on all, possibly duplicate, objects in the database.
         #[clap(visible_alias = "statistics")]
-        Stats,
+        Stats {
+            /// Lookup headers again, but without preloading indices.
+            #[clap(long)]
+            extra_header_lookup: bool,
+        },
     }
 }