This is an archive of the discontinued Mercurial Phabricator instance.

Differential D9684

copies-rust: move CPU-heavy Rust processing into a child thread
ClosedPublic

Authored by SimonSapin on Jan 6 2021, 9:12 AM.

Details

Reviewers
Alphare
Group Reviewers
hg-reviewers
Commits
rHG47557ea79fc7: copies-rust: move CPU-heavy Rust processing into a child thread
rHGc6b88b93b977: copies-rust: move CPU-heavy Rust processing into a child thread
Summary

… that runs in parallel with the parent thread fetching data.
This can be disabled through a new config. CLI example:

hg --config=devel.copy-tracing.multi-thread=no

For now both threads use the GIL, later commits will reduce this.

Diff Detail

Repository
rHG Mercurial
Branch
default
Lint
No Linters Available
Unit
No Unit Test Coverage

Event Timeline

SimonSapin created this revision.Jan 6 2021, 9:12 AM
Alphare added a subscriber: Alphare.Jan 22 2021, 4:58 AM
Alphare added inline comments.
rust/hg-cpython/src/copy_tracing.rs
54

This is a better comment, I almost said something on the last commit

78

This will not honor the RAYON_NUM_THREADS environment variable that we use to contain the number of threads in tests, but also technically for users as an implementation detail. Should we use rayon threads here so as to have the same threadpool everywhere?

SimonSapin added inline comments.Jan 22 2021, 11:48 AM
rust/hg-cpython/src/copy_tracing.rs
54

This was a pre-existing comment, the previous commit only moved it around. But yeah I could have included this change in the previous commit.

78

I feel that starting a pool of multiple thread to run a single task on it and leaving it running until process exit is counter-productive compared to the one thread started and then stopped here. Unless combine_changeset_copies_wrapper itself is called with high concurrency from many Python threads (which would contend the GIL) we’re not gonna have many of these Rust threads.

Alphare accepted this revision.Jan 24 2021, 12:23 AM
baymax updated this revision to Diff 25704.Feb 22 2021, 9:25 AM

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

baymax updated this revision to Diff 25741.Feb 22 2021, 11:00 AM

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

baymax updated this revision to Diff 25786.Feb 22 2021, 3:47 PM

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

SimonSapin added a commit: rHGc6b88b93b977: copies-rust: move CPU-heavy Rust processing into a child thread.Feb 24 2021, 11:03 AM
This revision was not accepted when it landed; it landed in state Needs Review.
Closed by commit rHGc6b88b93b977: copies-rust: move CPU-heavy Rust processing into a child thread (authored by SimonSapin). · Explain Why
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 25704

mercurial/configitems.py

) )
coreconfigitem( coreconfigitem(
b'devel', b'devel',
b'debug.copies', b'debug.copies',
default=False, default=False,
) )
coreconfigitem( coreconfigitem(
b'devel', b'devel',
b'copy-tracing.multi-thread',
default=True,
)
coreconfigitem(
b'devel',
b'debug.extensions', b'debug.extensions',
default=False, default=False,
) )
coreconfigitem( coreconfigitem(
b'devel', b'devel',
b'debug.repo-filters', b'debug.repo-filters',
default=False, default=False,
) )

mercurial/copies.py

parents = cl._uncheckedparentrevs parents = cl._uncheckedparentrevs
graph_roots = (nullrev, nullrev) graph_roots = (nullrev, nullrev)
ancestors = cl.ancestors([a.rev()], inclusive=True) ancestors = cl.ancestors([a.rev()], inclusive=True)
revs = cl.findmissingrevs(common=[a.rev()], heads=[b.rev()]) revs = cl.findmissingrevs(common=[a.rev()], heads=[b.rev()])
roots = set() roots = set()
has_graph_roots = False has_graph_roots = False
multi_thread = repo.ui.configbool(b'devel', b'copy-tracing.multi-thread')
# iterate over `only(B, A)` # iterate over `only(B, A)`
for r in revs: for r in revs:
ps = parents(r) ps = parents(r)
if ps == graph_roots: if ps == graph_roots:
has_graph_roots = True has_graph_roots = True
else: else:
p1, p2 = ps p1, p2 = ps
for p in cl.parentrevs(r): for p in cl.parentrevs(r):
if p == nullrev: if p == nullrev:
continue continue
children_count[r] = 0 children_count[r] = 0
if p in children_count: if p in children_count:
children_count[p] += 1 children_count[p] += 1
revinfo = _revinfo_getter(repo, match) revinfo = _revinfo_getter(repo, match)
return _combine_changeset_copies( return _combine_changeset_copies(
revs, children_count, b.rev(), revinfo, match, isancestor revs,
children_count,
b.rev(),
revinfo,
match,
isancestor,
multi_thread,
) )
else: else:
# When not using side-data, we will process the edges "from" the parent. # When not using side-data, we will process the edges "from" the parent.
# so we need a full mapping of the parent -> children relation. # so we need a full mapping of the parent -> children relation.
children = dict((r, []) for r in roots) children = dict((r, []) for r in roots)
for r in revs: for r in revs:
for p in cl.parentrevs(r): for p in cl.parentrevs(r):
if p == nullrev: if p == nullrev:
continue continue
children[r] = [] children[r] = []
if p in children: if p in children:
children[p].append(r) children[p].append(r)
x = revs.pop() x = revs.pop()
assert x == b.rev() assert x == b.rev()
revs.extend(roots) revs.extend(roots)
revs.sort() revs.sort()
revinfo = _revinfo_getter_extra(repo) revinfo = _revinfo_getter_extra(repo)
return _combine_changeset_copies_extra( return _combine_changeset_copies_extra(
revs, children, b.rev(), revinfo, match, isancestor revs, children, b.rev(), revinfo, match, isancestor
) )
def _combine_changeset_copies( def _combine_changeset_copies(
revs, children_count, targetrev, revinfo, match, isancestor revs, children_count, targetrev, revinfo, match, isancestor, multi_thread
): ):
"""combine the copies information for each item of iterrevs """combine the copies information for each item of iterrevs
revs: sorted iterable of revision to visit revs: sorted iterable of revision to visit
children_count: a {parent: <number-of-relevant-children>} mapping. children_count: a {parent: <number-of-relevant-children>} mapping.
targetrev: the final copies destination revision (not in iterrevs) targetrev: the final copies destination revision (not in iterrevs)
revinfo(rev): a function that return (p1, p2, p1copies, p2copies, removed) revinfo(rev): a function that return (p1, p2, p1copies, p2copies, removed)
match: a matcher match: a matcher
It returns the aggregated copies information for `targetrev`. It returns the aggregated copies information for `targetrev`.
""" """
alwaysmatch = match.always() alwaysmatch = match.always()
if rustmod is not None: if rustmod is not None:
final_copies = rustmod.combine_changeset_copies( final_copies = rustmod.combine_changeset_copies(
list(revs), children_count, targetrev, revinfo list(revs), children_count, targetrev, revinfo, multi_thread
) )
else: else:
isancestor = cached_is_ancestor(isancestor) isancestor = cached_is_ancestor(isancestor)
all_copies = {} all_copies = {}
# iterate over all the "children" side of copy tracing "edge" # iterate over all the "children" side of copy tracing "edge"
for current_rev in revs: for current_rev in revs:
p1, p2, changes = revinfo(current_rev) p1, p2, changes = revinfo(current_rev)

rust/Cargo.lock

"zstd 0.5.3+zstd.1.4.5 (registry+https://github.com/rust-lang/crates.io-index)", "zstd 0.5.3+zstd.1.4.5 (registry+https://github.com/rust-lang/crates.io-index)",
] ]
[[package]] [[package]]
name = "hg-cpython" name = "hg-cpython"
version = "0.1.0" version = "0.1.0"
dependencies = [ dependencies = [
"cpython 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)", "cpython 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)",
"crossbeam-channel 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)",
"env_logger 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)", "env_logger 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)",
"hg-core 0.1.0", "hg-core 0.1.0",
"libc 0.2.81 (registry+https://github.com/rust-lang/crates.io-index)", "libc 0.2.81 (registry+https://github.com/rust-lang/crates.io-index)",
"log 0.4.11 (registry+https://github.com/rust-lang/crates.io-index)", "log 0.4.11 (registry+https://github.com/rust-lang/crates.io-index)",
] ]
[[package]] [[package]]
name = "home" name = "home"

rust/hg-cpython/Cargo.toml

python3 = ["cpython/python3-sys", "cpython/extension-module"] python3 = ["cpython/python3-sys", "cpython/extension-module"]
# Enable one of these features to build a test executable linked to libpython: # Enable one of these features to build a test executable linked to libpython:
# e.g. cargo test --no-default-features --features python27-bin # e.g. cargo test --no-default-features --features python27-bin
python27-bin = ["cpython/python27-sys"] python27-bin = ["cpython/python27-sys"]
python3-bin = ["cpython/python3-sys"] python3-bin = ["cpython/python3-sys"]
[dependencies] [dependencies]
crossbeam-channel = "0.4"
hg-core = { path = "../hg-core"} hg-core = { path = "../hg-core"}
libc = '*' libc = '*'
log = "0.4.8" log = "0.4.8"
env_logger = "0.7.1" env_logger = "0.7.1"
[dependencies.cpython] [dependencies.cpython]
version = "0.4.1" version = "0.4.1"
default-features = false default-features = false

rust/hg-cpython/src/copy_tracing.rs

/// ///
/// See mercurial/copies.py for details /// See mercurial/copies.py for details
pub fn combine_changeset_copies_wrapper( pub fn combine_changeset_copies_wrapper(
py: Python, py: Python,
revs: PyList, revs: PyList,
children_count: PyDict, children_count: PyDict,
target_rev: Revision, target_rev: Revision,
rev_info: PyObject, rev_info: PyObject,
multi_thread: bool,
) -> PyResult<PyDict> { ) -> PyResult<PyDict> {
let children_count = children_count let children_count = children_count
.items(py) .items(py)
.iter() .iter()
.map(|(k, v)| Ok((k.extract(py)?, v.extract(py)?))) .map(|(k, v)| Ok((k.extract(py)?, v.extract(py)?)))
.collect::<PyResult<_>>()?; .collect::<PyResult<_>>()?;
/// (Revision number, parent 1, parent 2, copy data for this revision) /// (Revision number, parent 1, parent 2, copy data for this revision)
type RevInfo = (Revision, Revision, Revision, Option<PyBytes>); type RevInfo = (Revision, Revision, Revision, Option<PyBytes>);
let revs_info = revs.iter(py).map(|rev_py| -> PyResult<RevInfo> { let revs_info = revs.iter(py).map(|rev_py| -> PyResult<RevInfo> {
let rev = rev_py.extract(py)?; let rev = rev_py.extract(py)?;
let tuple: PyTuple = let tuple: PyTuple =
rev_info.call(py, (rev_py,), None)?.cast_into(py)?; rev_info.call(py, (rev_py,), None)?.cast_into(py)?;
let p1 = tuple.get_item(py, 0).extract(py)?; let p1 = tuple.get_item(py, 0).extract(py)?;
let p2 = tuple.get_item(py, 1).extract(py)?; let p2 = tuple.get_item(py, 1).extract(py)?;
let opt_bytes = tuple.get_item(py, 2).extract(py)?; let opt_bytes = tuple.get_item(py, 2).extract(py)?;
Ok((rev, p1, p2, opt_bytes)) Ok((rev, p1, p2, opt_bytes))
}); });
let path_copies = if !multi_thread {
let mut combine_changeset_copies = let mut combine_changeset_copies =
CombineChangesetCopies::new(children_count); CombineChangesetCopies::new(children_count);
for rev_info in revs_info { for rev_info in revs_info {
let (rev, p1, p2, opt_bytes) = rev_info?; let (rev, p1, p2, opt_bytes) = rev_info?;
let files = match &opt_bytes { let files = match &opt_bytes {
Some(bytes) => ChangedFiles::new(bytes.data(py)), Some(bytes) => ChangedFiles::new(bytes.data(py)),
// value was presumably None, meaning they was no copy data. // Python None was extracted to Option::None,
AlphareUnsubmitted
Not Done

This is a better comment, I almost said something on the last commit

Alphare: This is a better comment, I almost said something on the last commit
SimonSapinAuthorUnsubmitted
Done

This was a pre-existing comment, the previous commit only moved it around. But yeah I could have included this change in the previous commit.

SimonSapin: This was a pre-existing comment, the previous commit only moved it around. But yeah I could…
// meaning there was no copy data.
None => ChangedFiles::new_empty(), None => ChangedFiles::new_empty(),
}; };
combine_changeset_copies.add_revision(rev, p1, p2, files) combine_changeset_copies.add_revision(rev, p1, p2, files)
} }
let path_copies = combine_changeset_copies.finish(target_rev); combine_changeset_copies.finish(target_rev)
} else {
// Use a bounded channel to provide back-pressure:
// if the child thread is slower to process revisions than this thread
// is to gather data for them, an unbounded channel would keep
// growing and eat memory.
//
// TODO: tweak the bound?
let (rev_info_sender, rev_info_receiver) =
crossbeam_channel::bounded::<RevInfo>(1000);
// Start a thread that does CPU-heavy processing in parallel with the
// loop below.
//
// If the parent thread panics, `rev_info_sender` will be dropped and
// “disconnected”. `rev_info_receiver` will be notified of this and
// exit its own loop.
let thread = std::thread::spawn(move || {
AlphareUnsubmitted
Not Done

This will not honor the RAYON_NUM_THREADS environment variable that we use to contain the number of threads in tests, but also technically for users as an implementation detail. Should we use rayon threads here so as to have the same threadpool everywhere?

Alphare: This will not honor the `RAYON_NUM_THREADS` environment variable that we use to contain the…
SimonSapinAuthorUnsubmitted
Done

I feel that starting a pool of multiple thread to run a single task on it and leaving it running until process exit is counter-productive compared to the one thread started and then stopped here. Unless combine_changeset_copies_wrapper itself is called with high concurrency from many Python threads (which would contend the GIL) we’re not gonna have many of these Rust threads.

SimonSapin: I feel that starting a pool of multiple thread to run a single task on it and leaving it…
let mut combine_changeset_copies =
CombineChangesetCopies::new(children_count);
for (rev, p1, p2, opt_bytes) in rev_info_receiver {
let gil = Python::acquire_gil();
let py = gil.python();
let files = match &opt_bytes {
Some(raw) => ChangedFiles::new(raw.data(py)),
// Python None was extracted to Option::None,
// meaning there was no copy data.
None => ChangedFiles::new_empty(),
};
combine_changeset_copies.add_revision(rev, p1, p2, files)
}
combine_changeset_copies.finish(target_rev)
});
for rev_info in revs_info {
let (rev, p1, p2, opt_bytes) = rev_info?;
// We’d prefer to avoid the child thread calling into Python code,
// but this avoids a potential deadlock on the GIL if it does:
py.allow_threads(|| {
rev_info_sender.send((rev, p1, p2, opt_bytes)).expect(
"combine_changeset_copies: channel is disconnected",
);
});
}
// We’d prefer to avoid the child thread calling into Python code,
// but this avoids a potential deadlock on the GIL if it does:
py.allow_threads(|| {
// Disconnect the channel to signal the child thread to stop:
// the `for … in rev_info_receiver` loop will end.
drop(rev_info_sender);
// Wait for the child thread to stop, and propagate any panic.
thread.join().unwrap_or_else(|panic_payload| {
std::panic::resume_unwind(panic_payload)
})
})
};
let out = PyDict::new(py); let out = PyDict::new(py);
for (dest, source) in path_copies.into_iter() { for (dest, source) in path_copies.into_iter() {
out.set_item( out.set_item(
py, py,
PyBytes::new(py, &dest.into_vec()), PyBytes::new(py, &dest.into_vec()),
PyBytes::new(py, &source.into_vec()), PyBytes::new(py, &source.into_vec()),
)?; )?;
} }
py, py,
"combine_changeset_copies", "combine_changeset_copies",
py_fn!( py_fn!(
py, py,
combine_changeset_copies_wrapper( combine_changeset_copies_wrapper(
revs: PyList, revs: PyList,
children: PyDict, children: PyDict,
target_rev: Revision, target_rev: Revision,
rev_info: PyObject rev_info: PyObject,
multi_thread: bool
) )
), ),
)?; )?;
let sys = PyModule::import(py, "sys")?; let sys = PyModule::import(py, "sys")?;
let sys_modules: PyDict = sys.get(py, "modules")?.extract(py)?; let sys_modules: PyDict = sys.get(py, "modules")?.extract(py)?;
sys_modules.set_item(py, dotted_name, &m)?; sys_modules.set_item(py, dotted_name, &m)?;
Ok(m) Ok(m)
} }
mercurial/copies.py