|
|
use cpython::ObjectProtocol;
|
|
|
use cpython::PyBytes;
|
|
|
use cpython::PyDict;
|
|
|
use cpython::PyDrop;
|
|
|
use cpython::PyList;
|
|
|
use cpython::PyModule;
|
|
|
use cpython::PyObject;
|
|
|
use cpython::PyResult;
|
|
|
use cpython::PyTuple;
|
|
|
use cpython::Python;
|
|
|
|
|
|
use hg::copy_tracing::ChangedFiles;
|
|
|
use hg::copy_tracing::CombineChangesetCopies;
|
|
|
use hg::Revision;
|
|
|
|
|
|
use self::pybytes_with_data::PyBytesWithData;
|
|
|
|
|
|
// Module to encapsulate private fields
|
|
|
mod pybytes_with_data {
|
|
|
use cpython::{PyBytes, Python};
|
|
|
|
|
|
/// Safe abstraction over a `PyBytes` together with the `&[u8]` slice
|
|
|
/// that borrows it.
|
|
|
///
|
|
|
/// Calling `PyBytes::data` requires a GIL marker but we want to access the
|
|
|
/// data in a thread that (ideally) does not need to acquire the GIL.
|
|
|
/// This type allows separating the call an the use.
|
|
|
pub(super) struct PyBytesWithData {
|
|
|
#[allow(unused)]
|
|
|
keep_alive: PyBytes,
|
|
|
|
|
|
/// Borrows the buffer inside `self.keep_alive`,
|
|
|
/// but the borrow-checker cannot express self-referential structs.
|
|
|
data: *const [u8],
|
|
|
}
|
|
|
|
|
|
fn require_send<T: Send>() {}
|
|
|
|
|
|
#[allow(unused)]
|
|
|
fn static_assert_pybytes_is_send() {
|
|
|
require_send::<PyBytes>;
|
|
|
}
|
|
|
|
|
|
// Safety: PyBytes is Send. Raw pointers are not by default,
|
|
|
// but here sending one to another thread is fine since we ensure it stays
|
|
|
// valid.
|
|
|
unsafe impl Send for PyBytesWithData {}
|
|
|
|
|
|
impl PyBytesWithData {
|
|
|
pub fn new(py: Python, bytes: PyBytes) -> Self {
|
|
|
Self {
|
|
|
data: bytes.data(py),
|
|
|
keep_alive: bytes,
|
|
|
}
|
|
|
}
|
|
|
|
|
|
pub fn data(&self) -> &[u8] {
|
|
|
// Safety: the raw pointer is valid as long as the PyBytes is still
|
|
|
// alive, and the returned slice borrows `self`.
|
|
|
unsafe { &*self.data }
|
|
|
}
|
|
|
|
|
|
pub fn unwrap(self) -> PyBytes {
|
|
|
self.keep_alive
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
|
|
|
/// Combines copies information contained into revision `revs` to build a copy
|
|
|
/// map.
|
|
|
///
|
|
|
/// See mercurial/copies.py for details
|
|
|
pub fn combine_changeset_copies_wrapper(
|
|
|
py: Python,
|
|
|
revs: PyList,
|
|
|
children_count: PyDict,
|
|
|
target_rev: Revision,
|
|
|
rev_info: PyObject,
|
|
|
multi_thread: bool,
|
|
|
) -> PyResult<PyDict> {
|
|
|
let children_count = children_count
|
|
|
.items(py)
|
|
|
.iter()
|
|
|
.map(|(k, v)| Ok((k.extract(py)?, v.extract(py)?)))
|
|
|
.collect::<PyResult<_>>()?;
|
|
|
|
|
|
/// (Revision number, parent 1, parent 2, copy data for this revision)
|
|
|
type RevInfo<Bytes> = (Revision, Revision, Revision, Option<Bytes>);
|
|
|
|
|
|
let revs_info =
|
|
|
revs.iter(py).map(|rev_py| -> PyResult<RevInfo<PyBytes>> {
|
|
|
let rev = rev_py.extract(py)?;
|
|
|
let tuple: PyTuple =
|
|
|
rev_info.call(py, (rev_py,), None)?.cast_into(py)?;
|
|
|
let p1 = tuple.get_item(py, 0).extract(py)?;
|
|
|
let p2 = tuple.get_item(py, 1).extract(py)?;
|
|
|
let opt_bytes = tuple.get_item(py, 2).extract(py)?;
|
|
|
Ok((rev, p1, p2, opt_bytes))
|
|
|
});
|
|
|
|
|
|
let path_copies;
|
|
|
if !multi_thread {
|
|
|
let mut combine_changeset_copies =
|
|
|
CombineChangesetCopies::new(children_count);
|
|
|
|
|
|
for rev_info in revs_info {
|
|
|
let (rev, p1, p2, opt_bytes) = rev_info?;
|
|
|
let files = match &opt_bytes {
|
|
|
Some(bytes) => ChangedFiles::new(bytes.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)
|
|
|
}
|
|
|
path_copies = 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<PyBytesWithData>>(1000);
|
|
|
|
|
|
// This channel (going the other way around) however is unbounded.
|
|
|
// If they were both bounded, there might potentially be deadlocks
|
|
|
// where both channels are full and both threads are waiting on each
|
|
|
// other.
|
|
|
let (pybytes_sender, pybytes_receiver) =
|
|
|
crossbeam_channel::unbounded();
|
|
|
|
|
|
// 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 || {
|
|
|
let mut combine_changeset_copies =
|
|
|
CombineChangesetCopies::new(children_count);
|
|
|
for (rev, p1, p2, opt_bytes) in rev_info_receiver {
|
|
|
let files = match &opt_bytes {
|
|
|
Some(raw) => ChangedFiles::new(raw.data()),
|
|
|
// 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);
|
|
|
|
|
|
// Send `PyBytes` back to the parent thread so the parent
|
|
|
// thread can drop it. Otherwise the GIL would be implicitly
|
|
|
// acquired here through `impl Drop for PyBytes`.
|
|
|
if let Some(bytes) = opt_bytes {
|
|
|
if let Err(_) = pybytes_sender.send(bytes.unwrap()) {
|
|
|
// The channel is disconnected, meaning the parent
|
|
|
// thread panicked or returned
|
|
|
// early through
|
|
|
// `?` to propagate a Python exception.
|
|
|
break;
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
|
|
|
combine_changeset_copies.finish(target_rev)
|
|
|
});
|
|
|
|
|
|
for rev_info in revs_info {
|
|
|
let (rev, p1, p2, opt_bytes) = rev_info?;
|
|
|
let opt_bytes = opt_bytes.map(|b| PyBytesWithData::new(py, b));
|
|
|
|
|
|
// 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",
|
|
|
);
|
|
|
});
|
|
|
|
|
|
// Drop anything in the channel, without blocking
|
|
|
for pybytes in pybytes_receiver.try_iter() {
|
|
|
pybytes.release_ref(py)
|
|
|
}
|
|
|
}
|
|
|
// We’d prefer to avoid the child thread calling into Python code,
|
|
|
// but this avoids a potential deadlock on the GIL if it does:
|
|
|
path_copies = 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)
|
|
|
})
|
|
|
});
|
|
|
|
|
|
// Drop anything left in the channel
|
|
|
for pybytes in pybytes_receiver.iter() {
|
|
|
pybytes.release_ref(py)
|
|
|
}
|
|
|
};
|
|
|
|
|
|
let out = PyDict::new(py);
|
|
|
for (dest, source) in path_copies.into_iter() {
|
|
|
out.set_item(
|
|
|
py,
|
|
|
PyBytes::new(py, &dest.into_vec()),
|
|
|
PyBytes::new(py, &source.into_vec()),
|
|
|
)?;
|
|
|
}
|
|
|
Ok(out)
|
|
|
}
|
|
|
|
|
|
/// Create the module, with `__package__` given from parent
|
|
|
pub fn init_module(py: Python, package: &str) -> PyResult<PyModule> {
|
|
|
let dotted_name = &format!("{}.copy_tracing", package);
|
|
|
let m = PyModule::new(py, dotted_name)?;
|
|
|
|
|
|
m.add(py, "__package__", package)?;
|
|
|
m.add(py, "__doc__", "Copy tracing - Rust implementation")?;
|
|
|
|
|
|
m.add(
|
|
|
py,
|
|
|
"combine_changeset_copies",
|
|
|
py_fn!(
|
|
|
py,
|
|
|
combine_changeset_copies_wrapper(
|
|
|
revs: PyList,
|
|
|
children: PyDict,
|
|
|
target_rev: Revision,
|
|
|
rev_info: PyObject,
|
|
|
multi_thread: bool
|
|
|
)
|
|
|
),
|
|
|
)?;
|
|
|
|
|
|
let sys = PyModule::import(py, "sys")?;
|
|
|
let sys_modules: PyDict = sys.get(py, "modules")?.extract(py)?;
|
|
|
sys_modules.set_item(py, dotted_name, &m)?;
|
|
|
|
|
|
Ok(m)
|
|
|
}
|
|
|
|
