This is an archive of the discontinued Mercurial Phabricator instance.

Differential D8518

rust-status: collect traversed directories if required
ClosedPublic

Authored by Alphare on May 12 2020, 7:50 AM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHGc802ec4f7196: rust-status: collect traversed directories if required
Summary

Some commands (hg purge notably) register the traversedir callback on their
matcher to run said callback every time a directory is traversed.

This is the first of three patches, further broadening Rust support for status.

Unfortunately, there is no way around collecting a full Vec (or any other
owned datastructure, like a radix tree) and pushing it back up the Python layer
since keeping the Python callback in a closure would mean giving up
multithreading because of the GIL, which is obviously unacceptable.

Performance is still a lot better than the Python+C path.

Running hg clean/purge on Netbeans' repo (100k files):

     | No-op | 30% unknown
--------------------------
Rust | 1.0s  | 1.67s
C    | 2.0s  | 2.87s

Diff Detail

Repository
rHG Mercurial
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Event Timeline

Alphare created this revision.May 12 2020, 7:50 AM
Alphare added a commit: rHGc802ec4f7196: rust-status: collect traversed directories if required.May 14 2020, 4:30 PM
This revision was not accepted when it landed; it landed in state Needs Review.
Closed by commit rHGc802ec4f7196: rust-status: collect traversed directories if required (authored by Alphare). · Explain Why
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 21380

rust/hg-core/src/dirstate/status.rs

/// Get stat data about the files explicitly specified by match. /// Get stat data about the files explicitly specified by match.
/// TODO subrepos /// TODO subrepos
#[timed] #[timed]
fn walk_explicit<'a>( fn walk_explicit<'a>(
files: Option<&'a HashSet<&HgPath>>, files: Option<&'a HashSet<&HgPath>>,
dmap: &'a DirstateMap, dmap: &'a DirstateMap,
root_dir: impl AsRef<Path> + Sync + Send + 'a, root_dir: impl AsRef<Path> + Sync + Send + 'a,
options: StatusOptions, options: StatusOptions,
traversed_sender: crossbeam::Sender<HgPathBuf>,
) -> impl ParallelIterator<Item = IoResult<(&'a HgPath, Dispatch)>> { ) -> impl ParallelIterator<Item = IoResult<(&'a HgPath, Dispatch)>> {
files files
.unwrap_or(&DEFAULT_WORK) .unwrap_or(&DEFAULT_WORK)
.par_iter() .par_iter()
.map(move |filename| { .map(move |filename| {
// TODO normalization // TODO normalization
let normalized = filename.as_ref(); let normalized = filename.as_ref();
&dmap.copy_map, &dmap.copy_map,
options, options,
), ),
))); )));
} }
Some(Ok((normalized, Dispatch::Unknown))) Some(Ok((normalized, Dispatch::Unknown)))
} else { } else {
if file_type.is_dir() { if file_type.is_dir() {
if options.collect_traversed_dirs {
// The receiver always outlives the sender,
// so unwrap.
traversed_sender
.send(normalized.to_owned())
.unwrap()
}
Some(Ok(( Some(Ok((
normalized, normalized,
Dispatch::Directory { Dispatch::Directory {
was_file: in_dmap.is_some(), was_file: in_dmap.is_some(),
}, },
))) )))
} else { } else {
Some(Ok(( Some(Ok((
/// sure we won't miss future size-preserving file content modifications /// sure we won't miss future size-preserving file content modifications
/// that happen within the same timeslot. /// that happen within the same timeslot.
pub last_normal_time: i64, pub last_normal_time: i64,
/// Whether we are on a filesystem with UNIX-like exec flags /// Whether we are on a filesystem with UNIX-like exec flags
pub check_exec: bool, pub check_exec: bool,
pub list_clean: bool, pub list_clean: bool,
pub list_unknown: bool, pub list_unknown: bool,
pub list_ignored: bool, pub list_ignored: bool,
/// Whether to collect traversed dirs for applying a callback later.
/// Used by `hg purge` for example.
pub collect_traversed_dirs: bool,
} }
/// Dispatch a single entry (file, folder, symlink...) found during `traverse`. /// Dispatch a single entry (file, folder, symlink...) found during `traverse`.
/// If the entry is a folder that needs to be traversed, it will be handled /// If the entry is a folder that needs to be traversed, it will be handled
/// in a separate thread. /// in a separate thread.
fn handle_traversed_entry<'a>( fn handle_traversed_entry<'a>(
scope: &rayon::Scope<'a>, scope: &rayon::Scope<'a>,
files_sender: &'a crossbeam::Sender<IoResult<(HgPathBuf, Dispatch)>>, files_sender: &'a crossbeam::Sender<IoResult<(HgPathBuf, Dispatch)>>,
matcher: &'a (impl Matcher + Sync), matcher: &'a (impl Matcher + Sync),
root_dir: impl AsRef<Path> + Sync + Send + Copy + 'a, root_dir: impl AsRef<Path> + Sync + Send + Copy + 'a,
dmap: &'a DirstateMap, dmap: &'a DirstateMap,
old_results: &'a FastHashMap<Cow<HgPath>, Dispatch>, old_results: &'a FastHashMap<Cow<HgPath>, Dispatch>,
ignore_fn: &'a IgnoreFnType, ignore_fn: &'a IgnoreFnType,
dir_ignore_fn: &'a IgnoreFnType, dir_ignore_fn: &'a IgnoreFnType,
options: StatusOptions, options: StatusOptions,
filename: HgPathBuf, filename: HgPathBuf,
dir_entry: DirEntry, dir_entry: DirEntry,
traversed_sender: crossbeam::Sender<HgPathBuf>,
) -> IoResult<()> { ) -> IoResult<()> {
let file_type = dir_entry.file_type()?; let file_type = dir_entry.file_type()?;
let entry_option = dmap.get(&filename); let entry_option = dmap.get(&filename);
if filename.as_bytes() == b".hg" { if filename.as_bytes() == b".hg" {
// Could be a directory or a symlink // Could be a directory or a symlink
return Ok(()); return Ok(());
} }
if file_type.is_dir() { if file_type.is_dir() {
handle_traversed_dir( handle_traversed_dir(
scope, scope,
files_sender, files_sender,
matcher, matcher,
root_dir, root_dir,
dmap, dmap,
old_results, old_results,
ignore_fn, ignore_fn,
dir_ignore_fn, dir_ignore_fn,
options, options,
entry_option, entry_option,
filename, filename,
traversed_sender,
); );
} else if file_type.is_file() || file_type.is_symlink() { } else if file_type.is_file() || file_type.is_symlink() {
if let Some(entry) = entry_option { if let Some(entry) = entry_option {
if matcher.matches_everything() || matcher.matches(&filename) { if matcher.matches_everything() || matcher.matches(&filename) {
let metadata = dir_entry.metadata()?; let metadata = dir_entry.metadata()?;
files_sender files_sender
.send(Ok(( .send(Ok((
filename.to_owned(), filename.to_owned(),
root_dir: impl AsRef<Path> + Sync + Send + Copy + 'a, root_dir: impl AsRef<Path> + Sync + Send + Copy + 'a,
dmap: &'a DirstateMap, dmap: &'a DirstateMap,
old_results: &'a FastHashMap<Cow<HgPath>, Dispatch>, old_results: &'a FastHashMap<Cow<HgPath>, Dispatch>,
ignore_fn: &'a IgnoreFnType, ignore_fn: &'a IgnoreFnType,
dir_ignore_fn: &'a IgnoreFnType, dir_ignore_fn: &'a IgnoreFnType,
options: StatusOptions, options: StatusOptions,
entry_option: Option<&'a DirstateEntry>, entry_option: Option<&'a DirstateEntry>,
directory: HgPathBuf, directory: HgPathBuf,
traversed_sender: crossbeam::Sender<HgPathBuf>,
) { ) {
scope.spawn(move |_| { scope.spawn(move |_| {
// Nested `if` until `rust-lang/rust#53668` is stable // Nested `if` until `rust-lang/rust#53668` is stable
if let Some(entry) = entry_option { if let Some(entry) = entry_option {
// Used to be a file, is now a folder // Used to be a file, is now a folder
if matcher.matches_everything() || matcher.matches(&directory) { if matcher.matches_everything() || matcher.matches(&directory) {
files_sender files_sender
.send(Ok(( .send(Ok((
matcher, matcher,
root_dir, root_dir,
dmap, dmap,
directory, directory,
&old_results, &old_results,
ignore_fn, ignore_fn,
dir_ignore_fn, dir_ignore_fn,
options, options,
traversed_sender,
) )
.unwrap_or_else(|e| files_sender.send(Err(e)).unwrap()) .unwrap_or_else(|e| files_sender.send(Err(e)).unwrap())
} }
}); });
} }
/// Decides whether the directory needs to be listed, and if so handles the /// Decides whether the directory needs to be listed, and if so handles the
/// entries in a separate thread. /// entries in a separate thread.
fn traverse_dir<'a>( fn traverse_dir<'a>(
files_sender: &crossbeam::Sender<IoResult<(HgPathBuf, Dispatch)>>, files_sender: &crossbeam::Sender<IoResult<(HgPathBuf, Dispatch)>>,
matcher: &'a (impl Matcher + Sync), matcher: &'a (impl Matcher + Sync),
root_dir: impl AsRef<Path> + Sync + Send + Copy, root_dir: impl AsRef<Path> + Sync + Send + Copy,
dmap: &'a DirstateMap, dmap: &'a DirstateMap,
directory: impl AsRef<HgPath>, directory: impl AsRef<HgPath>,
old_results: &FastHashMap<Cow<'a, HgPath>, Dispatch>, old_results: &FastHashMap<Cow<'a, HgPath>, Dispatch>,
ignore_fn: &IgnoreFnType, ignore_fn: &IgnoreFnType,
dir_ignore_fn: &IgnoreFnType, dir_ignore_fn: &IgnoreFnType,
options: StatusOptions, options: StatusOptions,
traversed_sender: crossbeam::Sender<HgPathBuf>,
) -> IoResult<()> { ) -> IoResult<()> {
let directory = directory.as_ref(); let directory = directory.as_ref();
if options.collect_traversed_dirs {
// The receiver always outlives the sender, so unwrap.
traversed_sender.send(directory.to_owned()).unwrap()
}
let visit_entries = match matcher.visit_children_set(directory) { let visit_entries = match matcher.visit_children_set(directory) {
VisitChildrenSet::Empty => return Ok(()), VisitChildrenSet::Empty => return Ok(()),
VisitChildrenSet::This | VisitChildrenSet::Recursive => None, VisitChildrenSet::This | VisitChildrenSet::Recursive => None,
VisitChildrenSet::Set(set) => Some(set), VisitChildrenSet::Set(set) => Some(set),
}; };
let buf = hg_path_to_path_buf(directory)?; let buf = hg_path_to_path_buf(directory)?;
let dir_path = root_dir.as_ref().join(buf); let dir_path = root_dir.as_ref().join(buf);
root_dir, root_dir,
dmap, dmap,
old_results, old_results,
ignore_fn, ignore_fn,
dir_ignore_fn, dir_ignore_fn,
options, options,
filename, filename,
dir_entry, dir_entry,
traversed_sender.clone(),
)?; )?;
} }
} }
Ok(()) Ok(())
}) })
} }
/// Walk the working directory recursively to look for changes compared to the /// Walk the working directory recursively to look for changes compared to the
/// current `DirstateMap`. /// current `DirstateMap`.
/// ///
/// This takes a mutable reference to the results to account for the `extend` /// This takes a mutable reference to the results to account for the `extend`
/// in timings /// in timings
#[timed] #[timed]
fn traverse<'a>( fn traverse<'a>(
matcher: &'a (impl Matcher + Sync), matcher: &'a (impl Matcher + Sync),
root_dir: impl AsRef<Path> + Sync + Send + Copy, root_dir: impl AsRef<Path> + Sync + Send + Copy,
dmap: &'a DirstateMap, dmap: &'a DirstateMap,
path: impl AsRef<HgPath>, path: impl AsRef<HgPath>,
old_results: &FastHashMap<Cow<'a, HgPath>, Dispatch>, old_results: &FastHashMap<Cow<'a, HgPath>, Dispatch>,
ignore_fn: &IgnoreFnType, ignore_fn: &IgnoreFnType,
dir_ignore_fn: &IgnoreFnType, dir_ignore_fn: &IgnoreFnType,
options: StatusOptions, options: StatusOptions,
results: &mut Vec<(Cow<'a, HgPath>, Dispatch)>, results: &mut Vec<(Cow<'a, HgPath>, Dispatch)>,
traversed_sender: crossbeam::Sender<HgPathBuf>,
) -> IoResult<()> { ) -> IoResult<()> {
let root_dir = root_dir.as_ref(); let root_dir = root_dir.as_ref();
// The traversal is done in parallel, so use a channel to gather entries. // The traversal is done in parallel, so use a channel to gather entries.
// `crossbeam::Sender` is `Send`, while `mpsc::Sender` is not. // `crossbeam::Sender` is `Send`, while `mpsc::Sender` is not.
let (files_transmitter, files_receiver) = crossbeam::channel::unbounded(); let (files_transmitter, files_receiver) = crossbeam::channel::unbounded();
traverse_dir( traverse_dir(
&files_transmitter, &files_transmitter,
matcher, matcher,
root_dir, root_dir,
&dmap, &dmap,
path, path,
&old_results, &old_results,
&ignore_fn, &ignore_fn,
&dir_ignore_fn, &dir_ignore_fn,
options, options,
traversed_sender,
)?; )?;
// Disconnect the channel so the receiver stops waiting // Disconnect the channel so the receiver stops waiting
drop(files_transmitter); drop(files_transmitter);
// TODO don't collect. Find a way of replicating the behavior of // TODO don't collect. Find a way of replicating the behavior of
// `itertools::process_results`, but for `rayon::ParallelIterator` // `itertools::process_results`, but for `rayon::ParallelIterator`
let new_results: IoResult<Vec<(Cow<'a, HgPath>, Dispatch)>> = let new_results: IoResult<Vec<(Cow<'a, HgPath>, Dispatch)>> =
pub modified: Vec<Cow<'a, HgPath>>, pub modified: Vec<Cow<'a, HgPath>>,
pub added: Vec<Cow<'a, HgPath>>, pub added: Vec<Cow<'a, HgPath>>,
pub removed: Vec<Cow<'a, HgPath>>, pub removed: Vec<Cow<'a, HgPath>>,
pub deleted: Vec<Cow<'a, HgPath>>, pub deleted: Vec<Cow<'a, HgPath>>,
pub clean: Vec<Cow<'a, HgPath>>, pub clean: Vec<Cow<'a, HgPath>>,
pub ignored: Vec<Cow<'a, HgPath>>, pub ignored: Vec<Cow<'a, HgPath>>,
pub unknown: Vec<Cow<'a, HgPath>>, pub unknown: Vec<Cow<'a, HgPath>>,
pub bad: Vec<(Cow<'a, HgPath>, BadMatch)>, pub bad: Vec<(Cow<'a, HgPath>, BadMatch)>,
/// Only filled if `collect_traversed_dirs` is `true`
pub traversed: Vec<HgPathBuf>,
} }
#[timed] #[timed]
fn build_response<'a>( fn build_response<'a>(
results: impl IntoIterator<Item = (Cow<'a, HgPath>, Dispatch)>, results: impl IntoIterator<Item = (Cow<'a, HgPath>, Dispatch)>,
traversed: Vec<HgPathBuf>,
) -> (Vec<Cow<'a, HgPath>>, DirstateStatus<'a>) { ) -> (Vec<Cow<'a, HgPath>>, DirstateStatus<'a>) {
let mut lookup = vec![]; let mut lookup = vec![];
let mut modified = vec![]; let mut modified = vec![];
let mut added = vec![]; let mut added = vec![];
let mut removed = vec![]; let mut removed = vec![];
let mut deleted = vec![]; let mut deleted = vec![];
let mut clean = vec![]; let mut clean = vec![];
let mut ignored = vec![]; let mut ignored = vec![];
modified, modified,
added, added,
removed, removed,
deleted, deleted,
clean, clean,
ignored, ignored,
unknown, unknown,
bad, bad,
traversed,
}, },
) )
} }
#[derive(Debug)] #[derive(Debug)]
pub enum StatusError { pub enum StatusError {
IO(std::io::Error), IO(std::io::Error),
Path(HgPathError), Path(HgPathError),
(dir_ignore_fn, warnings) (dir_ignore_fn, warnings)
} else { } else {
ignore_fn = Box::new(|&_| true); ignore_fn = Box::new(|&_| true);
(Box::new(|&_| true), vec![]) (Box::new(|&_| true), vec![])
}; };
let files = matcher.file_set(); let files = matcher.file_set();
// `crossbeam::Sender` is `Send`, while `mpsc::Sender` is not.
let (traversed_sender, traversed_recv) = crossbeam::channel::unbounded();
// Step 1: check the files explicitly mentioned by the user // Step 1: check the files explicitly mentioned by the user
let explicit = walk_explicit(files, &dmap, root_dir, options); let explicit = walk_explicit(
files,
&dmap,
root_dir,
options,
traversed_sender.clone(),
);
// Collect results into a `Vec` because we do very few lookups in most // Collect results into a `Vec` because we do very few lookups in most
// cases. // cases.
let (work, mut results): (Vec<_>, Vec<_>) = explicit let (work, mut results): (Vec<_>, Vec<_>) = explicit
.filter_map(Result::ok) .filter_map(Result::ok)
.map(|(filename, dispatch)| (Cow::Borrowed(filename), dispatch)) .map(|(filename, dispatch)| (Cow::Borrowed(filename), dispatch))
.partition(|(_, dispatch)| match dispatch { .partition(|(_, dispatch)| match dispatch {
Dispatch::Directory { .. } => true, Dispatch::Directory { .. } => true,
root_dir, root_dir,
&dmap, &dmap,
&dir, &dir,
&old_results, &old_results,
&ignore_fn, &ignore_fn,
&dir_ignore_fn, &dir_ignore_fn,
options, options,
&mut results, &mut results,
traversed_sender.clone(),
)?; )?;
} }
} }
_ => unreachable!("There can only be directories in `work`"), _ => unreachable!("There can only be directories in `work`"),
} }
} }
} }
if !matcher.is_exact() { if !matcher.is_exact() {
// Step 3: Check the remaining files from the dmap. // Step 3: Check the remaining files from the dmap.
// If a dmap file is not in results yet, it was either // If a dmap file is not in results yet, it was either
// a) not matched b) ignored, c) missing, or d) under a // a) not matched b) ignored, c) missing, or d) under a
// symlink directory. // symlink directory.
if options.list_unknown { if options.list_unknown {
handle_unknowns(dmap, matcher, root_dir, options, &mut results)?; handle_unknowns(dmap, matcher, root_dir, options, &mut results)?;
} else { } else {
// We may not have walked the full directory tree above, so stat // We may not have walked the full directory tree above, so stat
// and check everything we missed. // and check everything we missed.
extend_from_dmap(&dmap, root_dir, options, &mut results); extend_from_dmap(&dmap, root_dir, options, &mut results);
} }
} }
Ok((build_response(results), warnings)) // Close the channel
drop(traversed_sender);
let traversed_dirs = traversed_recv.into_iter().collect();
Ok((build_response(results, traversed_dirs), warnings))
} }