( )⚙ D10551 dirstate-tree: Paralellize the status algorithm with Rayon

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D10551

dirstate-tree: Paralellize the status algorithm with Rayon
ClosedPublic

Authored by SimonSapin on May 3 2021, 6:25 AM.

Details

Reviewers
Alphare
Group Reviewers
hg-reviewers
Commits
rHG60d852ae7e7b: dirstate-tree: Paralellize the status algorithm with Rayon
rHG1762ba4a6d73: dirstate-tree: Paralellize the status algorithm with Rayon
Summary

The rayon crate exposes "parallel iterators" that work like normal iterators
but dispatch work on different items to an implicit global thread pool.

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

SimonSapin created this revision.May 3 2021, 6:25 AM
Alphare accepted this revision.May 3 2021, 10:48 AM
Alphare added a subscriber: Alphare.

I wonder which of using std::Mutex vs a crossbeam channel is faster. But this can be fine-tuned when we actually have most of the work done.

This revision is now accepted and ready to land.May 3 2021, 10:48 AM
SimonSapin added a comment.May 6 2021, 6:07 AM

Yes, this is worth benchmarking later. I think a mutex is at least not horribly slow because each thread spend a lot of time in readdir and stat without holding the mutex, and taking a mutex without contention should be fast (a few atomic operations). If the system’s / standard library’s mutex isn’t fast (does a syscall every time) we can try https://crates.io/crates/parking_lot. And channels also need synchronization: https://github.com/zesterer/flume is basically Arc<Mutex<VecDeque<T>> and has benchmarks claiming it’s competitive with crossbeam-channel. It’s possible that crossbeam-channel will still be faster (as far as I understand it has more fine-grained synchronization, with atomic metadata for each slot in the queue). Or maybe none of this is a bottleneck in the first place.

baymax updated this revision to Diff 27627.May 6 2021, 12:03 PM

✅ refresh by Heptapod after a successful CI run (🐙 💚)

Revision Contents
Changeset List

Diff 27661

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

use crate::dirstate::status::IgnoreFnType; use crate::dirstate::status::IgnoreFnType;
use crate::dirstate_tree::dirstate_map::ChildNodes; use crate::dirstate_tree::dirstate_map::ChildNodes;
use crate::dirstate_tree::dirstate_map::DirstateMap; use crate::dirstate_tree::dirstate_map::DirstateMap;
use crate::dirstate_tree::dirstate_map::Node; use crate::dirstate_tree::dirstate_map::Node;
use crate::matchers::get_ignore_function; use crate::matchers::get_ignore_function;
use crate::matchers::Matcher; use crate::matchers::Matcher;
use crate::utils::files::get_bytes_from_os_string; use crate::utils::files::get_bytes_from_os_string;
use crate::utils::hg_path::HgPath; use crate::utils::hg_path::HgPath;
use crate::BadMatch; use crate::BadMatch;
use crate::DirstateStatus; use crate::DirstateStatus;
use crate::EntryState; use crate::EntryState;
use crate::HgPathBuf; use crate::HgPathBuf;
use crate::PatternFileWarning; use crate::PatternFileWarning;
use crate::StatusError; use crate::StatusError;
use crate::StatusOptions; use crate::StatusOptions;
use rayon::prelude::*;
use std::borrow::Cow; use std::borrow::Cow;
use std::io; use std::io;
use std::path::Path; use std::path::Path;
use std::path::PathBuf; use std::path::PathBuf;
use std::sync::Mutex;
/// Returns the status of the working directory compared to its parent /// Returns the status of the working directory compared to its parent
/// changeset. /// changeset.
/// ///
/// This algorithm is based on traversing the filesystem tree (`fs` in function /// This algorithm is based on traversing the filesystem tree (`fs` in function
/// and variable names) and dirstate tree at the same time. The core of this /// and variable names) and dirstate tree at the same time. The core of this
/// traversal is the recursive `traverse_fs_directory_and_dirstate` function /// traversal is the recursive `traverse_fs_directory_and_dirstate` function
/// and its use of `itertools::merge_join_by`. When reaching a path that only /// and its use of `itertools::merge_join_by`. When reaching a path that only
/// exists in one of the two trees, depending on information requested by /// exists in one of the two trees, depending on information requested by
/// `options` we may need to traverse the remaining subtree. /// `options` we may need to traverse the remaining subtree.
pub fn status<'tree>( pub fn status<'tree>(
dmap: &'tree mut DirstateMap, dmap: &'tree mut DirstateMap,
matcher: &(dyn Matcher + Sync), matcher: &(dyn Matcher + Sync),
root_dir: PathBuf, root_dir: PathBuf,
ignore_files: Vec<PathBuf>, ignore_files: Vec<PathBuf>,
options: StatusOptions, options: StatusOptions,
) -> Result<(DirstateStatus<'tree>, Vec<PatternFileWarning>), StatusError> { ) -> Result<(DirstateStatus<'tree>, Vec<PatternFileWarning>), StatusError> {
let (ignore_fn, warnings): (IgnoreFnType, _) = let (ignore_fn, warnings): (IgnoreFnType, _) =
if options.list_ignored || options.list_unknown { if options.list_ignored || options.list_unknown {
get_ignore_function(ignore_files, &root_dir)? get_ignore_function(ignore_files, &root_dir)?
} else { } else {
(Box::new(|&_| true), vec![]) (Box::new(|&_| true), vec![])
}; };
let mut common = StatusCommon { let common = StatusCommon {
options, options,
matcher, matcher,
ignore_fn, ignore_fn,
outcome: DirstateStatus::default(), outcome: Mutex::new(DirstateStatus::default()),
}; };
let is_at_repo_root = true; let is_at_repo_root = true;
let hg_path = HgPath::new(""); let hg_path = HgPath::new("");
let has_ignored_ancestor = false; let has_ignored_ancestor = false;
common.traverse_fs_directory_and_dirstate( common.traverse_fs_directory_and_dirstate(
has_ignored_ancestor, has_ignored_ancestor,
&mut dmap.root, &mut dmap.root,
hg_path, hg_path,
&root_dir, &root_dir,
is_at_repo_root, is_at_repo_root,
); );
Ok((common.outcome, warnings)) Ok((common.outcome.into_inner().unwrap(), warnings))
} }
/// Bag of random things needed by various parts of the algorithm. Reduces the /// Bag of random things needed by various parts of the algorithm. Reduces the
/// number of parameters passed to functions. /// number of parameters passed to functions.
struct StatusCommon<'tree, 'a> { struct StatusCommon<'tree, 'a> {
options: StatusOptions, options: StatusOptions,
matcher: &'a (dyn Matcher + Sync), matcher: &'a (dyn Matcher + Sync),
ignore_fn: IgnoreFnType<'a>, ignore_fn: IgnoreFnType<'a>,
outcome: DirstateStatus<'tree>, outcome: Mutex<DirstateStatus<'tree>>,
} }
impl<'tree, 'a> StatusCommon<'tree, 'a> { impl<'tree, 'a> StatusCommon<'tree, 'a> {
fn read_dir( fn read_dir(
&mut self, &self,
hg_path: &HgPath, hg_path: &HgPath,
fs_path: &Path, fs_path: &Path,
is_at_repo_root: bool, is_at_repo_root: bool,
) -> Result<Vec<DirEntry>, ()> { ) -> Result<Vec<DirEntry>, ()> {
DirEntry::read_dir(fs_path, is_at_repo_root).map_err(|error| { DirEntry::read_dir(fs_path, is_at_repo_root).map_err(|error| {
let errno = error.raw_os_error().expect("expected real OS error"); let errno = error.raw_os_error().expect("expected real OS error");
self.outcome self.outcome
.lock()
.unwrap()
.bad .bad
.push((hg_path.to_owned().into(), BadMatch::OsError(errno))) .push((hg_path.to_owned().into(), BadMatch::OsError(errno)))
}) })
} }
fn traverse_fs_directory_and_dirstate( fn traverse_fs_directory_and_dirstate(
&mut self, &self,
has_ignored_ancestor: bool, has_ignored_ancestor: bool,
dirstate_nodes: &'tree mut ChildNodes, dirstate_nodes: &'tree mut ChildNodes,
directory_hg_path: &'tree HgPath, directory_hg_path: &'tree HgPath,
directory_fs_path: &Path, directory_fs_path: &Path,
is_at_repo_root: bool, is_at_repo_root: bool,
) { ) {
let mut fs_entries = if let Ok(entries) = self.read_dir( let mut fs_entries = if let Ok(entries) = self.read_dir(
directory_hg_path, directory_hg_path,
directory_fs_path, directory_fs_path,
is_at_repo_root, is_at_repo_root,
) { ) {
entries entries
} else { } else {
return; return;
}; };
// `merge_join_by` requires both its input iterators to be sorted: // `merge_join_by` requires both its input iterators to be sorted:
//
// * `BTreeMap` iterates according to keys’ ordering by definition // * `BTreeMap` iterates according to keys’ ordering by definition
// `sort_unstable_by_key` doesn’t allow keys borrowing from the value: // `sort_unstable_by_key` doesn’t allow keys borrowing from the value:
// https://github.com/rust-lang/rust/issues/34162 // https://github.com/rust-lang/rust/issues/34162
fs_entries.sort_unstable_by(|e1, e2| e1.base_name.cmp(&e2.base_name)); fs_entries.sort_unstable_by(|e1, e2| e1.base_name.cmp(&e2.base_name));
for pair in itertools::merge_join_by( itertools::merge_join_by(
dirstate_nodes, dirstate_nodes,
&fs_entries, &fs_entries,
|(full_path, _node), fs_entry| { |(full_path, _node), fs_entry| {
full_path.base_name().cmp(&fs_entry.base_name) full_path.base_name().cmp(&fs_entry.base_name)
}, },
) { )
.par_bridge()
.for_each(|pair| {
use itertools::EitherOrBoth::*; use itertools::EitherOrBoth::*;
match pair { match pair {
Both((hg_path, dirstate_node), fs_entry) => { Both((hg_path, dirstate_node), fs_entry) => {
self.traverse_fs_and_dirstate( self.traverse_fs_and_dirstate(
fs_entry, fs_entry,
hg_path.full_path(), hg_path.full_path(),
dirstate_node, dirstate_node,
has_ignored_ancestor, has_ignored_ancestor,
); );
} }
Left((hg_path, dirstate_node)) => self.traverse_dirstate_only( Left((hg_path, dirstate_node)) => self.traverse_dirstate_only(
hg_path.full_path(), hg_path.full_path(),
dirstate_node, dirstate_node,
), ),
Right(fs_entry) => self.traverse_fs_only( Right(fs_entry) => self.traverse_fs_only(
has_ignored_ancestor, has_ignored_ancestor,
directory_hg_path, directory_hg_path,
fs_entry, fs_entry,
), ),
} }
} })
} }
fn traverse_fs_and_dirstate( fn traverse_fs_and_dirstate(
&mut self, &self,
fs_entry: &DirEntry, fs_entry: &DirEntry,
hg_path: &'tree HgPath, hg_path: &'tree HgPath,
dirstate_node: &'tree mut Node, dirstate_node: &'tree mut Node,
has_ignored_ancestor: bool, has_ignored_ancestor: bool,
) { ) {
let file_type = fs_entry.metadata.file_type(); let file_type = fs_entry.metadata.file_type();
let file_or_symlink = file_type.is_file() || file_type.is_symlink(); let file_or_symlink = file_type.is_file() || file_type.is_symlink();
if !file_or_symlink { if !file_or_symlink {
// If we previously had a file here, it was removed (with // If we previously had a file here, it was removed (with
// `hg rm` or similar) or deleted before it could be // `hg rm` or similar) or deleted before it could be
// replaced by a directory or something else. // replaced by a directory or something else.
self.mark_removed_or_deleted_if_file( self.mark_removed_or_deleted_if_file(
hg_path, hg_path,
dirstate_node.state(), dirstate_node.state(),
); );
} }
if file_type.is_dir() { if file_type.is_dir() {
if self.options.collect_traversed_dirs { if self.options.collect_traversed_dirs {
self.outcome.traversed.push(hg_path.into()) self.outcome.lock().unwrap().traversed.push(hg_path.into())
} }
let is_ignored = has_ignored_ancestor || (self.ignore_fn)(hg_path); let is_ignored = has_ignored_ancestor || (self.ignore_fn)(hg_path);
let is_at_repo_root = false; let is_at_repo_root = false;
self.traverse_fs_directory_and_dirstate( self.traverse_fs_directory_and_dirstate(
is_ignored, is_ignored,
&mut dirstate_node.children, &mut dirstate_node.children,
hg_path, hg_path,
&fs_entry.full_path, &fs_entry.full_path,
is_at_repo_root, is_at_repo_root,
); );
} else { } else {
if file_or_symlink && self.matcher.matches(hg_path) { if file_or_symlink && self.matcher.matches(hg_path) {
let full_path = Cow::from(hg_path); let full_path = Cow::from(hg_path);
if let Some(entry) = &dirstate_node.entry { if let Some(entry) = &dirstate_node.entry {
match entry.state { match entry.state {
EntryState::Added => { EntryState::Added => {
self.outcome.added.push(full_path) self.outcome.lock().unwrap().added.push(full_path)
}
EntryState::Removed => {
self.outcome.removed.push(full_path)
}
EntryState::Merged => {
self.outcome.modified.push(full_path)
} }
EntryState::Removed => self
.outcome
.lock()
.unwrap()
.removed
.push(full_path),
EntryState::Merged => self
.outcome
.lock()
.unwrap()
.modified
.push(full_path),
EntryState::Normal => { EntryState::Normal => {
self.handle_normal_file( self.handle_normal_file(
full_path, full_path,
dirstate_node, dirstate_node,
entry, entry,
fs_entry, fs_entry,
); );
} }
) )
} }
} }
} }
/// A file with `EntryState::Normal` in the dirstate was found in the /// A file with `EntryState::Normal` in the dirstate was found in the
/// filesystem /// filesystem
fn handle_normal_file( fn handle_normal_file(
&mut self, &self,
full_path: Cow<'tree, HgPath>, full_path: Cow<'tree, HgPath>,
dirstate_node: &Node, dirstate_node: &Node,
entry: &crate::DirstateEntry, entry: &crate::DirstateEntry,
fs_entry: &DirEntry, fs_entry: &DirEntry,
) { ) {
// Keep the low 31 bits // Keep the low 31 bits
fn truncate_u64(value: u64) -> i32 { fn truncate_u64(value: u64) -> i32 {
(value & 0x7FFF_FFFF) as i32 (value & 0x7FFF_FFFF) as i32
} }
fn truncate_i64(value: i64) -> i32 { fn truncate_i64(value: i64) -> i32 {
(value & 0x7FFF_FFFF) as i32 (value & 0x7FFF_FFFF) as i32
} }
let mode_changed = || { let mode_changed = || {
self.options.check_exec && entry.mode_changed(&fs_entry.metadata) self.options.check_exec && entry.mode_changed(&fs_entry.metadata)
}; };
let size_changed = entry.size != truncate_u64(fs_entry.metadata.len()); let size_changed = entry.size != truncate_u64(fs_entry.metadata.len());
if entry.size >= 0 if entry.size >= 0
&& size_changed && size_changed
&& fs_entry.metadata.file_type().is_symlink() && fs_entry.metadata.file_type().is_symlink()
{ {
// issue6456: Size returned may be longer due to encryption // issue6456: Size returned may be longer due to encryption
// on EXT-4 fscrypt. TODO maybe only do it on EXT4? // on EXT-4 fscrypt. TODO maybe only do it on EXT4?
self.outcome.unsure.push(full_path) self.outcome.lock().unwrap().unsure.push(full_path)
} else if dirstate_node.copy_source.is_some() } else if dirstate_node.copy_source.is_some()
|| entry.is_from_other_parent() || entry.is_from_other_parent()
|| (entry.size >= 0 && (size_changed || mode_changed())) || (entry.size >= 0 && (size_changed || mode_changed()))
{ {
self.outcome.modified.push(full_path) self.outcome.lock().unwrap().modified.push(full_path)
} else { } else {
let mtime = mtime_seconds(&fs_entry.metadata); let mtime = mtime_seconds(&fs_entry.metadata);
if truncate_i64(mtime) != entry.mtime if truncate_i64(mtime) != entry.mtime
|| mtime == self.options.last_normal_time || mtime == self.options.last_normal_time
{ {
self.outcome.unsure.push(full_path) self.outcome.lock().unwrap().unsure.push(full_path)
} else if self.options.list_clean { } else if self.options.list_clean {
self.outcome.clean.push(full_path) self.outcome.lock().unwrap().clean.push(full_path)
} }
} }
} }
/// A node in the dirstate tree has no corresponding filesystem entry /// A node in the dirstate tree has no corresponding filesystem entry
fn traverse_dirstate_only( fn traverse_dirstate_only(
&mut self, &self,
hg_path: &'tree HgPath, hg_path: &'tree HgPath,
dirstate_node: &'tree mut Node, dirstate_node: &'tree mut Node,
) { ) {
self.mark_removed_or_deleted_if_file(hg_path, dirstate_node.state()); self.mark_removed_or_deleted_if_file(hg_path, dirstate_node.state());
for (child_hg_path, child_node) in &mut dirstate_node.children { dirstate_node.children.par_iter_mut().for_each(
self.traverse_dirstate_only(child_hg_path.full_path(), child_node) |(child_hg_path, child_node)| {
} self.traverse_dirstate_only(
child_hg_path.full_path(),
child_node,
)
},
)
} }
/// A node in the dirstate tree has no corresponding *file* on the /// A node in the dirstate tree has no corresponding *file* on the
/// filesystem /// filesystem
/// ///
/// Does nothing on a "directory" node /// Does nothing on a "directory" node
fn mark_removed_or_deleted_if_file( fn mark_removed_or_deleted_if_file(
&mut self, &self,
hg_path: &'tree HgPath, hg_path: &'tree HgPath,
dirstate_node_state: Option<EntryState>, dirstate_node_state: Option<EntryState>,
) { ) {
if let Some(state) = dirstate_node_state { if let Some(state) = dirstate_node_state {
if self.matcher.matches(hg_path) { if self.matcher.matches(hg_path) {
if let EntryState::Removed = state { if let EntryState::Removed = state {
self.outcome.removed.push(hg_path.into()) self.outcome.lock().unwrap().removed.push(hg_path.into())
} else { } else {
self.outcome.deleted.push(hg_path.into()) self.outcome.lock().unwrap().deleted.push(hg_path.into())
} }
} }
} }
} }
/// Something in the filesystem has no corresponding dirstate node /// Something in the filesystem has no corresponding dirstate node
fn traverse_fs_only( fn traverse_fs_only(
&mut self, &self,
has_ignored_ancestor: bool, has_ignored_ancestor: bool,
directory_hg_path: &HgPath, directory_hg_path: &HgPath,
fs_entry: &DirEntry, fs_entry: &DirEntry,
) { ) {
let hg_path = directory_hg_path.join(&fs_entry.base_name); let hg_path = directory_hg_path.join(&fs_entry.base_name);
let file_type = fs_entry.metadata.file_type(); let file_type = fs_entry.metadata.file_type();
let file_or_symlink = file_type.is_file() || file_type.is_symlink(); let file_or_symlink = file_type.is_file() || file_type.is_symlink();
if file_type.is_dir() { if file_type.is_dir() {
}; };
if traverse_children { if traverse_children {
let is_at_repo_root = false; let is_at_repo_root = false;
if let Ok(children_fs_entries) = self.read_dir( if let Ok(children_fs_entries) = self.read_dir(
&hg_path, &hg_path,
&fs_entry.full_path, &fs_entry.full_path,
is_at_repo_root, is_at_repo_root,
) { ) {
for child_fs_entry in children_fs_entries { children_fs_entries.par_iter().for_each(|child_fs_entry| {
self.traverse_fs_only( self.traverse_fs_only(
is_ignored, is_ignored,
&hg_path, &hg_path,
&child_fs_entry, child_fs_entry,
) )
} })
} }
} }
if self.options.collect_traversed_dirs { if self.options.collect_traversed_dirs {
self.outcome.traversed.push(hg_path.into()) self.outcome.lock().unwrap().traversed.push(hg_path.into())
} }
} else if file_or_symlink && self.matcher.matches(&hg_path) { } else if file_or_symlink && self.matcher.matches(&hg_path) {
self.mark_unknown_or_ignored(has_ignored_ancestor, hg_path.into()) self.mark_unknown_or_ignored(has_ignored_ancestor, hg_path.into())
} }
} }
fn mark_unknown_or_ignored( fn mark_unknown_or_ignored(
&mut self, &self,
has_ignored_ancestor: bool, has_ignored_ancestor: bool,
hg_path: Cow<'tree, HgPath>, hg_path: Cow<'tree, HgPath>,
) { ) {
let is_ignored = has_ignored_ancestor || (self.ignore_fn)(&hg_path); let is_ignored = has_ignored_ancestor || (self.ignore_fn)(&hg_path);
if is_ignored { if is_ignored {
if self.options.list_ignored { if self.options.list_ignored {
self.outcome.ignored.push(hg_path) self.outcome.lock().unwrap().ignored.push(hg_path)
} }
} else { } else {
if self.options.list_unknown { if self.options.list_unknown {
self.outcome.unknown.push(hg_path) self.outcome.lock().unwrap().unknown.push(hg_path)
} }
} }
} }
} }
#[cfg(unix)] // TODO #[cfg(unix)] // TODO
fn mtime_seconds(metadata: &std::fs::Metadata) -> i64 { fn mtime_seconds(metadata: &std::fs::Metadata) -> i64 {
// Going through `Metadata::modified()` would be portable, but would take // Going through `Metadata::modified()` would be portable, but would take