This is an archive of the discontinued Mercurial Phabricator instance.

Differential D6519

rust-discovery: takefullsample() core implementation
AbandonedPublic

Authored by gracinet on Jun 12 2019, 2:16 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Summary

take_full_sample() browses the undecided set in both directions: from
its roots as well as from its heads.

Following what's done on the Python side, we alter update_sample()
signature to take a closure returning an iterator: either ParentsIterator
or an iterator over the children found in children_cache. These constructs
should probably be split off in a separate module.

This is a first concrete example where a more abstract graph notion (probably
a trait) would be useful, as this is nothing but an operation on the reversed
DAG.

A similar motivation in the context of the discovery
process would be to replace the call to dagops::range in
add_missing_revisions() with a simple iteration over descendents, again an
operation on the reversed graph.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Revision Contents
Changeset List

Diff 15467

rust/hg-core/src/discovery.rs

extern crate rand; extern crate rand;
extern crate rand_pcg; extern crate rand_pcg;
use self::rand::seq::SliceRandom; use self::rand::seq::SliceRandom;
use self::rand::{thread_rng, RngCore, SeedableRng}; use self::rand::{thread_rng, RngCore, SeedableRng};
use super::{Graph, GraphError, Revision, NULL_REVISION}; use super::{Graph, GraphError, Revision, NULL_REVISION};
use crate::ancestors::MissingAncestors; use crate::ancestors::MissingAncestors;
use crate::dagops; use crate::dagops;
use std::cmp::{max, min};
use std::collections::{HashMap, HashSet, VecDeque}; use std::collections::{HashMap, HashSet, VecDeque};
type Rng = self::rand_pcg::Pcg32; type Rng = self::rand_pcg::Pcg32;
pub struct PartialDiscovery<G: Graph + Clone> { pub struct PartialDiscovery<G: Graph + Clone> {
target_heads: Option<Vec<Revision>>, target_heads: Option<Vec<Revision>>,
graph: G, // plays the role of self._repo graph: G, // plays the role of self._repo
common: MissingAncestors<G>, common: MissingAncestors<G>,
undecided: Option<HashSet<Revision>>, undecided: Option<HashSet<Revision>>,
children_cache: Option<HashMap<Revision, Vec<Revision>>>,
missing: HashSet<Revision>, missing: HashSet<Revision>,
rng: Rng, rng: Rng,
respect_size: bool,
} }
pub struct DiscoveryStats { pub struct DiscoveryStats {
pub undecided: Option<usize>, pub undecided: Option<usize>,
} }
/// Update an existing sample to match the expected size /// Update an existing sample to match the expected size
/// ///
/// The sample is updated with revisions exponentially distant from each /// The sample is updated with revisions exponentially distant from each
/// element of `heads`. /// element of `heads`.
/// ///
/// If a target size is specified, the sampling will stop once this size is /// If a target size is specified, the sampling will stop once this size is
/// reached. Otherwise sampling will happen until roots of the <revs> set are /// reached. Otherwise sampling will happen until roots of the <revs> set are
/// reached. /// reached.
/// ///
/// - `revs`: set of revs we want to discover (if None, `assume` the whole dag /// - `revs`: set of revs we want to discover (if None, `assume` the whole dag
/// represented by `parentfn` /// represented by `parentfn`
/// - `heads`: set of DAG head revs /// - `heads`: set of DAG head revs
/// - `sample`: a sample to update /// - `sample`: a sample to update
/// - `parentfn`: a callable to resolve parents for a revision /// - `parentfn`: a callable to resolve parents for a revision
/// - `quicksamplesize`: optional target size of the sample /// - `quicksamplesize`: optional target size of the sample
fn update_sample( fn update_sample<I>(
revs: Option<&HashSet<Revision>>, revs: Option<&HashSet<Revision>>,
heads: impl IntoIterator<Item = Revision>, heads: impl IntoIterator<Item = Revision>,
sample: &mut HashSet<Revision>, sample: &mut HashSet<Revision>,
parentsfn: impl Fn(Revision) -> Result<[Revision; 2], GraphError>, parentsfn: impl Fn(Revision) -> Result<I, GraphError>,
quicksamplesize: Option<usize>, quicksamplesize: Option<usize>,
) -> Result<(), GraphError> { ) -> Result<(), GraphError>
where
I: Iterator<Item = Revision>,
{
let mut distances: HashMap<Revision, u32> = HashMap::new(); let mut distances: HashMap<Revision, u32> = HashMap::new();
let mut visit: VecDeque<Revision> = heads.into_iter().collect(); let mut visit: VecDeque<Revision> = heads.into_iter().collect();
let mut factor: u32 = 1; let mut factor: u32 = 1;
let mut seen: HashSet<Revision> = HashSet::new(); let mut seen: HashSet<Revision> = HashSet::new();
loop { loop {
let current = match visit.pop_front() { let current = match visit.pop_front() {
None => { None => {
break; break;
if d == factor { if d == factor {
sample.insert(current); sample.insert(current);
if let Some(sz) = quicksamplesize { if let Some(sz) = quicksamplesize {
if sample.len() >= sz { if sample.len() >= sz {
return Ok(()); return Ok(());
} }
} }
} }
for &p in &parentsfn(current)? { for p in parentsfn(current)? {
if p == NULL_REVISION {
continue;
}
if let Some(revs) = revs { if let Some(revs) = revs {
if !revs.contains(&p) { if !revs.contains(&p) {
continue; continue;
} }
} }
distances.entry(p).or_insert(d + 1); distances.entry(p).or_insert(d + 1);
visit.push_back(p); visit.push_back(p);
} }
} }
Ok(()) Ok(())
} }
struct ParentsIterator {
parents: [Revision; 2],
cur: usize,
}
impl ParentsIterator {
fn graph_parents(
graph: &impl Graph,
r: Revision,
) -> Result<ParentsIterator, GraphError> {
Ok(ParentsIterator {
parents: graph.parents(r)?,
cur: 0,
})
}
}
impl Iterator for ParentsIterator {
type Item = Revision;
fn next(&mut self) -> Option<Revision> {
if self.cur > 1 {
return None;
}
let rev = self.parents[self.cur];
self.cur += 1;
if rev == NULL_REVISION {
return self.next();
}
Some(rev)
}
}
impl<G: Graph + Clone> PartialDiscovery<G> { impl<G: Graph + Clone> PartialDiscovery<G> {
/// Create a PartialDiscovery object, with the intent /// Create a PartialDiscovery object, with the intent
/// of comparing our `::<target_heads>` revset to the contents of another /// of comparing our `::<target_heads>` revset to the contents of another
/// repo. /// repo.
/// ///
/// For now `target_heads` is passed as a vector, and will be used /// For now `target_heads` is passed as a vector, and will be used
/// at the first call to `ensure_undecided()`. /// at the first call to `ensure_undecided()`.
/// ///
/// If we want to make the signature more flexible, /// If we want to make the signature more flexible,
/// we'll have to make it a type argument of `PartialDiscovery` or a trait /// we'll have to make it a type argument of `PartialDiscovery` or a trait
/// object since we'll keep it in the meanwhile /// object since we'll keep it in the meanwhile
pub fn new(graph: G, target_heads: Vec<Revision>) -> Self { pub fn new(
graph: G,
target_heads: Vec<Revision>,
respect_size: bool,
) -> Self {
let mut seed: [u8; 16] = [0; 16]; let mut seed: [u8; 16] = [0; 16];
thread_rng().fill_bytes(&mut seed); thread_rng().fill_bytes(&mut seed);
Self::new_with_seed(graph, target_heads, seed) Self::new_with_seed(graph, target_heads, seed, respect_size)
} }
pub fn new_with_seed( pub fn new_with_seed(
graph: G, graph: G,
target_heads: Vec<Revision>, target_heads: Vec<Revision>,
seed: [u8; 16], seed: [u8; 16],
respect_size: bool,
) -> Self { ) -> Self {
PartialDiscovery { PartialDiscovery {
undecided: None, undecided: None,
children_cache: None,
target_heads: Some(target_heads), target_heads: Some(target_heads),
graph: graph.clone(), graph: graph.clone(),
common: MissingAncestors::new(graph, vec![]), common: MissingAncestors::new(graph, vec![]),
missing: HashSet::new(), missing: HashSet::new(),
rng: Rng::from_seed(seed), rng: Rng::from_seed(seed),
respect_size: respect_size,
} }
} }
/// Extract at most `size` random elements from sample and return them /// Extract at most `size` random elements from sample and return them
/// as a vector /// as a vector
fn limit_sample( fn limit_sample(
&mut self, &mut self,
mut sample: Vec<Revision>, mut sample: Vec<Revision>,
return Ok(()); return Ok(());
} }
let tgt = self.target_heads.take().unwrap(); let tgt = self.target_heads.take().unwrap();
self.undecided = self.undecided =
Some(self.common.missing_ancestors(tgt)?.into_iter().collect()); Some(self.common.missing_ancestors(tgt)?.into_iter().collect());
Ok(()) Ok(())
} }
fn ensure_children_cache(&mut self) -> Result<(), GraphError> {
if self.children_cache.is_some() {
return Ok(());
}
self.ensure_undecided()?;
let mut children: HashMap<Revision, Vec<Revision>> = HashMap::new();
for &rev in self.undecided.as_ref().unwrap() {
for p in ParentsIterator::graph_parents(&self.graph, rev)? {
children.entry(p).or_insert_with(|| Vec::new()).push(rev);
}
}
self.children_cache = Some(children);
Ok(())
}
/// Provide statistics about the current state of the discovery process /// Provide statistics about the current state of the discovery process
pub fn stats(&self) -> DiscoveryStats { pub fn stats(&self) -> DiscoveryStats {
DiscoveryStats { DiscoveryStats {
undecided: self.undecided.as_ref().map(|s| s.len()), undecided: self.undecided.as_ref().map(|s| s.len()),
} }
} }
pub fn take_quick_sample( pub fn take_quick_sample(
}; };
if sample.len() >= size { if sample.len() >= size {
return Ok(self.limit_sample(sample.into_iter().collect(), size)); return Ok(self.limit_sample(sample.into_iter().collect(), size));
} }
update_sample( update_sample(
None, None,
headrevs, headrevs,
&mut sample, &mut sample,
|r| self.graph.parents(r), |r| ParentsIterator::graph_parents(&self.graph, r),
Some(size), Some(size),
)?; )?;
Ok(sample.into_iter().collect()) Ok(sample.into_iter().collect())
} }
/// Extract a sample from `self.undecided`, going from its heads and roots.
///
/// The `size` parameter is used to avoid useless computations if
/// it turns out to be bigger than the whole set of undecided Revisions.
///
/// The sample is taken by using `update_sample` from the heads, then
/// from the roots, working on the reverse DAG,
/// expressed by `self.children_cache`.
///
/// No effort is being made to complete or limit the sample to `size`
/// but this method returns another interesting size that it derives
/// from its knowledge of the structure of the various sets, leaving
/// to the caller the decision to use it or not.
fn bidirectional_sample(
&mut self,
size: usize,
) -> Result<(HashSet<Revision>, usize), GraphError> {
self.ensure_undecided()?;
{
// we don't want to compute children_cache before this
// but doing it after extracting self.undecided takes a mutable
// ref to self while a shareable one is still active.
let undecided = self.undecided.as_ref().unwrap();
if undecided.len() <= size {
return Ok((undecided.clone(), size));
}
}
self.ensure_children_cache()?;
let revs = self.undecided.as_ref().unwrap();
let mut sample: HashSet<Revision> = revs.clone();
// it's possible that leveraging the children cache would be more
// efficient here
dagops::retain_heads(&self.graph, &mut sample)?;
let revsheads = sample.clone(); // was again heads(revs) in python
// update from heads
update_sample(
Some(revs),
revsheads.iter().cloned(),
&mut sample,
|r| ParentsIterator::graph_parents(&self.graph, r),
None,
)?;
// update from roots
let revroots: HashSet<Revision> =
dagops::roots(&self.graph, revs)?.into_iter().collect();
let prescribed_size = max(size, min(revroots.len(), revsheads.len()));
let children = self.children_cache.as_ref().unwrap();
let empty_vec: Vec<Revision> = Vec::new();
update_sample(
Some(revs),
revroots,
&mut sample,
|r| Ok(children.get(&r).unwrap_or(&empty_vec).iter().cloned()),
None,
)?;
Ok((sample, prescribed_size))
}
/// Fill up sample up to the wished size with random undecided Revisions.
///
/// This is intended to be used as a last resort completion if the
/// regular sampling algorithm returns too few elements.
fn random_complete_sample(
&mut self,
sample: &mut Vec<Revision>,
size: usize,
) {
let sample_len = sample.len();
if size <= sample_len {
return;
}
let take_from: Vec<Revision> = self
.undecided
.as_ref()
.unwrap()
.iter()
.filter(|&r| !sample.contains(r))
.cloned()
.collect();
sample.extend(self.limit_sample(take_from, size - sample_len));
}
pub fn take_full_sample(
&mut self,
size: usize,
) -> Result<Vec<Revision>, GraphError> {
let (sample_set, prescribed_size) = self.bidirectional_sample(size)?;
let size = if self.respect_size { size } else { prescribed_size };
let mut sample =
self.limit_sample(sample_set.into_iter().collect(), size);
self.random_complete_sample(&mut sample, size);
Ok(sample)
}
} }
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
use crate::testing::SampleGraph; use crate::testing::SampleGraph;
/// A PartialDiscovery as for pushing all the heads of `SampleGraph` /// A PartialDiscovery as for pushing all the heads of `SampleGraph`
/// ///
/// To avoid actual randomness in tests, we give it a fixed random seed. /// To avoid actual randomness in tests, we give it a fixed random seed.
fn full_disco() -> PartialDiscovery<SampleGraph> { fn full_disco() -> PartialDiscovery<SampleGraph> {
PartialDiscovery::new_with_seed( PartialDiscovery::new_with_seed(
SampleGraph, SampleGraph,
vec![10, 11, 12, 13], vec![10, 11, 12, 13],
[0; 16], [0; 16],
true,
) )
} }
/// A PartialDiscovery as for pushing the 12 head of `SampleGraph` /// A PartialDiscovery as for pushing the 12 head of `SampleGraph`
/// ///
/// To avoid actual randomness in tests, we give it a fixed random seed. /// To avoid actual randomness in tests, we give it a fixed random seed.
fn disco12() -> PartialDiscovery<SampleGraph> { fn disco12() -> PartialDiscovery<SampleGraph> {
PartialDiscovery::new_with_seed(SampleGraph, vec![12], [0; 16]) PartialDiscovery::new_with_seed(SampleGraph, vec![12], [0; 16], true)
} }
fn sorted_undecided( fn sorted_undecided(
disco: &PartialDiscovery<SampleGraph>, disco: &PartialDiscovery<SampleGraph>,
) -> Vec<Revision> { ) -> Vec<Revision> {
let mut as_vec: Vec<Revision> = let mut as_vec: Vec<Revision> =
disco.undecided.as_ref().unwrap().iter().cloned().collect(); disco.undecided.as_ref().unwrap().iter().cloned().collect();
as_vec.sort(); as_vec.sort();
let mut sample_vec = disco.take_quick_sample(vec![12], 4)?; let mut sample_vec = disco.take_quick_sample(vec![12], 4)?;
sample_vec.sort(); sample_vec.sort();
// r12's only parent is r9, whose unique grand-parent through the // r12's only parent is r9, whose unique grand-parent through the
// diamond shape is r4. This ends there because the distance from r4 // diamond shape is r4. This ends there because the distance from r4
// to the root is only 3. // to the root is only 3.
assert_eq!(sample_vec, vec![4, 9, 12]); assert_eq!(sample_vec, vec![4, 9, 12]);
Ok(()) Ok(())
} }
#[test]
fn test_children_cache() -> Result<(), GraphError> {
let mut disco = full_disco();
disco.ensure_children_cache()?;
let cache = disco.children_cache.unwrap();
assert_eq!(cache.get(&2).cloned(), Some(vec![4]));
assert_eq!(cache.get(&10).cloned(), None);
let mut children_4 = cache.get(&4).cloned().unwrap();
children_4.sort();
assert_eq!(children_4, vec![5, 6, 7]);
let mut children_7 = cache.get(&7).cloned().unwrap();
children_7.sort();
assert_eq!(children_7, vec![9, 11]);
Ok(())
}
#[test]
fn test_complete_sample() {
let mut disco = full_disco();
let undecided: HashSet<Revision> =
[4, 7, 9, 2, 3].iter().cloned().collect();
disco.undecided = Some(undecided);
let mut sample = vec![0];
disco.random_complete_sample(&mut sample, 3);
assert_eq!(sample.len(), 3);
let mut sample = vec![2, 4, 7];
disco.random_complete_sample(&mut sample, 1);
assert_eq!(sample.len(), 3);
}
#[test]
fn test_bidirectional_sample() -> Result<(), GraphError> {
let mut disco = full_disco();
disco.undecided = Some((0..=13).into_iter().collect());
let (sample_set, size) = disco.bidirectional_sample(7)?;
assert_eq!(size, 7);
let mut sample: Vec<Revision> = sample_set.into_iter().collect();
sample.sort();
// our DAG is a bit too small for the results to be really interesting
// at least it shows that
// - we went both ways
// - we didn't take all Revisions (6 is not in the sample)
assert_eq!(sample, vec![0, 1, 2, 3, 4, 5, 7, 8, 9, 10, 11, 12, 13]);
Ok(())
}
} }

rust/hg-cpython/src/discovery.rs

py_class!(pub class PartialDiscovery |py| { py_class!(pub class PartialDiscovery |py| {
data inner: RefCell<Box<CorePartialDiscovery<Index>>>; data inner: RefCell<Box<CorePartialDiscovery<Index>>>;
def __new__( def __new__(
_cls, _cls,
repo: PyObject, repo: PyObject,
targetheads: PyObject, targetheads: PyObject,
_respectsize: bool respectsize: bool
) -> PyResult<PartialDiscovery> { ) -> PyResult<PartialDiscovery> {
let index = repo.getattr(py, "changelog")?.getattr(py, "index")?; let index = repo.getattr(py, "changelog")?.getattr(py, "index")?;
Self::create_instance( Self::create_instance(
py, py,
RefCell::new(Box::new(CorePartialDiscovery::new( RefCell::new(Box::new(CorePartialDiscovery::new(
Index::new(py, index)?, Index::new(py, index)?,
rev_pyiter_collect(py, &targetheads)?, rev_pyiter_collect(py, &targetheads)?,
respectsize
))) )))
) )
} }
def addcommons(&self, commons: PyObject) -> PyResult<PyObject> { def addcommons(&self, commons: PyObject) -> PyResult<PyObject> {
let mut inner = self.inner(py).borrow_mut(); let mut inner = self.inner(py).borrow_mut();
let commons_vec: Vec<Revision> = rev_pyiter_collect(py, &commons)?; let commons_vec: Vec<Revision> = rev_pyiter_collect(py, &commons)?;
inner.add_common_revisions(commons_vec) inner.add_common_revisions(commons_vec)