Move slice_repository to its own library

Summary:
The `slice_repository` function is used in derived data backfiller, to remove the need from loading the whole commit graph to start deriving. This is very useful and could be used in other places, so this diff extracts that behaviour and makes it more generic.

Done:
- Move to a different lib
- Deblobrepoify it
- Make it more generic, so it's not derived data specific, and can be used for anything you can say "still needs to be done on this commit", and depends on its parents being done.

I will use this library in another place on the next commit.

Reviewed By: markbt

Differential Revision: D42640268

fbshipit-source-id: 66bf297c6790dce6412c7f0a283bc0379fc125c3

Author: yancouto

eden/mononoke/Cargo.toml

@@ -256,6 +256,7 @@ members = [
   "commit_rewriting/mononoke_x_repo_sync_job",
   "commit_rewriting/movers",
   "commit_rewriting/synced_commit_mapping",
+  "commit_traversal/slice_repository",
   "common/allocation_tracing",
   "common/async_limiter",
   "common/async_limiter/examples/tokio_v2",

eden/mononoke/backfill_derived_data/Cargo.toml

@@ -52,6 +52,7 @@ repo_identity = { version = "0.1.0", path = "../repo_attributes/repo_identity" }
 scuba_ext = { version = "0.1.0", path = "../common/scuba_ext" }
 skeleton_manifest = { version = "0.1.0", path = "../derived_data/skeleton_manifest" }
 skiplist = { version = "0.1.0", path = "../reachabilityindex/skiplist" }
+slice_repository = { version = "0.1.0", path = "../commit_traversal/slice_repository" }
 slog = { version = "2.7", features = ["max_level_trace", "nested-values"] }
 stats = { version = "0.1.0", git = "https://github.com/facebookexperimental/rust-shed.git", branch = "main" }
 time_ext = { version = "0.1.0", git = "https://github.com/facebookexperimental/rust-shed.git", branch = "main" }

eden/mononoke/cmds/backfill_derived_data/main.rs

@@ -1028,15 +1028,16 @@ async fn backfill_heads(
     wait_for_replication: WaitForReplication,
 ) -> Result<()> {
     if let (Some(skiplist_index), Some(slice_size)) = (skiplist_index, slice_size) {
-        let (count, slices) =
+        let slices =
             slice::slice_repository(ctx, repo, skiplist_index, derivers, heads, slice_size).await?;
-        for (index, (id, slice_heads)) in slices.enumerate() {
+        let slice_count = slices.len();
+        for (index, (id, slice_heads)) in slices.into_iter().enumerate() {
             info!(
                 ctx.logger(),
                 "Deriving slice {} ({}/{}) with {} heads",
                 id,
                 index + 1,
-                count,
+                slice_count,
                 slice_heads.len()
             );
             tail_batch_iteration(

eden/mononoke/cmds/backfill_derived_data/slice.rs

@@ -5,27 +5,20 @@
  * GNU General Public License version 2.
  */
 
-use std::collections::BTreeMap;
-use std::collections::HashMap;
 use std::collections::HashSet;
 use std::sync::Arc;
 
 use anyhow::Error;
 use anyhow::Result;
 use blobrepo::BlobRepo;
-use changeset_fetcher::ChangesetFetcherArc;
-use changeset_fetcher::ChangesetFetcherRef;
 use context::CoreContext;
 use derived_data_utils::DerivedUtils;
 use futures::stream;
 use futures::stream::FuturesUnordered;
-use futures::stream::StreamExt;
 use futures::stream::TryStreamExt;
 use mononoke_types::ChangesetId;
-use mononoke_types::Generation;
 use repo_derived_data::RepoDerivedDataArc;
 use skiplist::SkiplistIndex;
-use slog::info;
 
 /// Determine which heads are underived in any of the derivers.
 async fn underived_heads(
@@ -51,29 +44,6 @@ async fn underived_heads(
         .await
 }
 
-/// If skiplist parents are not available, fetch the parents and their
-/// generation from the repo.
-async fn parents_with_generations(
-    ctx: &CoreContext,
-    repo: &BlobRepo,
-    csid: ChangesetId,
-) -> Result<Vec<(ChangesetId, Generation)>> {
-    let parents = repo.changeset_fetcher().get_parents(ctx, csid).await?;
-    let parents_with_generations =
-        stream::iter(parents.into_iter().map(|parent_csid| async move {
-            let gen = repo
-                .changeset_fetcher()
-                .get_generation_number(ctx, parent_csid)
-                .await?;
-            Ok(Some((parent_csid, gen)))
-        }))
-        .buffered(100)
-        .try_filter_map(|maybe_csid_gen| async move { Ok::<_, Error>(maybe_csid_gen) })
-        .try_collect::<Vec<_>>()
-        .await?;
-    Ok(parents_with_generations)
-}
-
 /// Slice a respository into a sequence of slices for derivation.
 ///
 /// For large repositories with a long history, computing the full set of
@@ -124,114 +94,14 @@ pub(crate) async fn slice_repository(
     derivers: &[Arc<dyn DerivedUtils>],
     heads: Vec<ChangesetId>,
     slice_size: u64,
-) -> Result<(usize, impl Iterator<Item = (u64, Vec<ChangesetId>)>)> {
-    let heads = underived_heads(ctx, repo, derivers, heads.as_slice()).await?;
-
-    if skiplist_index.indexed_node_count() == 0 {
-        // This skiplist index is not populated.  Generate a single
-        // slice with all heads.
-        info!(
-            ctx.logger(),
-            "Repository not sliced as skiplist index is not populated",
-        );
-        let heads = heads.into_iter().collect();
-        return Ok((1, vec![(0, heads)].into_iter().rev()));
-    }
-
-    // Add any unindexed heads to the skiplist index.
-    let changeset_fetcher = repo.changeset_fetcher_arc();
-    for head in heads.iter() {
-        skiplist_index
-            .add_node(ctx, &changeset_fetcher, *head, std::u64::MAX)
-            .await?;
-    }
-
-    let mut head_generation_groups: BTreeMap<u64, Vec<ChangesetId>> = BTreeMap::new();
-    stream::iter(heads.into_iter().map(|csid| async move {
-        let gen = repo
-            .changeset_fetcher()
-            .get_generation_number(ctx, csid)
-            .await?;
-        Ok(Some((csid, gen)))
-    }))
-    .buffered(100)
-    .try_for_each(|maybe_csid_gen| {
-        if let Some((csid, gen)) = maybe_csid_gen {
-            let gen_group = (gen.value() / slice_size) * slice_size;
-            head_generation_groups
-                .entry(gen_group)
-                .or_default()
-                .push(csid);
-        }
-        async { Ok::<_, Error>(()) }
-    })
-    .await?;
-
-    let mut slices = Vec::new();
-    while let Some((cur_gen, mut heads)) = head_generation_groups.pop_last() {
-        info!(
-            ctx.logger(),
-            "Adding slice starting at generation {} with {} heads ({} slices queued)",
-            cur_gen,
-            heads.len(),
-            head_generation_groups.len()
-        );
-        let mut new_heads_groups = HashMap::new();
-        let mut seen: HashSet<_> = heads.iter().cloned().collect();
-        while let Some(csid) = heads.pop() {
-            let skip_parents = match skiplist_index.get_furthest_edges(csid) {
-                Some(skip_parents) => skip_parents,
-                None => {
-                    // Ordinarily this shouldn't happen, as the skiplist ought
-                    // to refer to commits that are also in the skiplist.
-                    // However, if the commit is missing from the skiplist, we
-                    // can look up the parents and their generations directly.
-                    parents_with_generations(ctx, repo, csid).await?
-                }
-            };
-
-            for (parent, gen) in skip_parents {
-                if gen.value() >= cur_gen {
-                    // This commit is in the same generation group.
-                    if seen.insert(parent) {
-                        heads.push(parent);
-                    }
-                } else {
-                    // This commit is in a new generation group.
-                    let gen_group = (gen.value() / slice_size) * slice_size;
-                    new_heads_groups.insert(parent, gen_group);
-                }
-            }
-        }
-
-        // Add all commits we've seen to the slice.  The heads from the start
-        // of this iteration would be sufficient, however providing additional
-        // changesets will allow traversal of the graph to find all commits to
-        // run faster as it can fetch the parents of multiple commits at once.
-        slices.push((cur_gen, seen.into_iter().collect()));
-
-        // For each new head, check if it needs derivation, and if so, add it
-        // to its generation group.
-        let new_heads: Vec<_> = new_heads_groups.keys().cloned().collect();
-        let underived_new_heads =
-            underived_heads(ctx, repo, derivers, new_heads.as_slice()).await?;
-        for head in underived_new_heads {
-            if let Some(gen_group) = new_heads_groups.get(&head) {
-                head_generation_groups
-                    .entry(*gen_group)
-                    .or_default()
-                    .push(head);
-            }
-        }
-    }
-
-    if !slices.is_empty() {
-        info!(
-            ctx.logger(),
-            "Repository sliced into {} slices requiring derivation",
-            slices.len()
-        );
-    }
-
-    Ok((slices.len(), slices.into_iter().rev()))
+) -> Result<Vec<(u64, Vec<ChangesetId>)>> {
+    slice_repository::slice_repository(
+        ctx,
+        repo,
+        skiplist_index,
+        heads,
+        |heads| async move { underived_heads(ctx, repo, derivers, heads.as_slice()).await },
+        slice_size,
+    )
+    .await
 }

eden/mononoke/commit_traversal/slice_repository/Cargo.toml

@@ -0,0 +1,17 @@
+# @generated by autocargo
+
+[package]
+name = "slice_repository"
+version = "0.1.0"
+authors = ["Facebook"]
+edition = "2021"
+license = "GPLv2+"
+
+[dependencies]
+anyhow = "1.0.65"
+changeset_fetcher = { version = "0.1.0", path = "../../blobrepo/changeset_fetcher" }
+context = { version = "0.1.0", path = "../../server/context" }
+futures = { version = "0.3.22", features = ["async-await", "compat"] }
+mononoke_types = { version = "0.1.0", path = "../../mononoke_types" }
+skiplist = { version = "0.1.0", path = "../../reachabilityindex/skiplist" }
+slog = { version = "2.7", features = ["max_level_trace", "nested-values"] }