Newer
Older
//! Tools for moving the repository to a given revision
use std::{
fs::Permissions,
io::Write,
os::unix::fs::{MetadataExt, PermissionsExt},
path::Path,
sync::atomic::Ordering,
time::Duration,
};
use crate::{
dirstate::entry::{ParentFileData, TruncatedTimestamp},
dirstate::{dirstate_map::DirstateEntryReset, on_disk::write_tracked_key},
errors::{HgError, IoResultExt},
operations::{list_rev_tracked_files, ExpandedManifestEntry},
progress::Progress,
repo::Repo,
revlog::filelog::Filelog,
revlog::node::NULL_NODE,
revlog::options::{default_revlog_options, RevlogOpenOptions},
revlog::RevlogError,
sparse,
utils::{
cap_default_rayon_threads,
files::{filesystem_now, get_path_from_bytes},
hg_path::{hg_path_to_path_buf, HgPath, HgPathError},
path_auditor::PathAuditor,
},
vfs::{is_on_nfs_mount, VfsImpl},
DirstateParents, UncheckedRevision, INTERRUPT_RECEIVED,
};
use crossbeam_channel::{Receiver, Sender};
use rayon::prelude::*;
fn write_dirstate(repo: &Repo) -> Result<(), HgError> {
repo.write_dirstate()
.map_err(|e| HgError::abort(e.to_string(), exit_codes::ABORT, None))?;
write_tracked_key(repo)
}
/// Update the current working copy of `repo` to the given revision `to`, from
/// the null revision and set + write out the dirstate to reflect that.
///
/// Do not call this outside of a Python context. This does *not* handle any
/// of the checks, hooks, lock taking needed to setup and get out of this
/// update from the null revision.
pub fn update_from_null(
repo: &Repo,
to: UncheckedRevision,
progress: &dyn Progress,
workers: Option<usize>,
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
) -> Result<usize, HgError> {
// Ignore the warnings, they've been displayed by Python already
// TODO non-Python clients: display narrow warnings
let (narrow_matcher, _) = narrow::matcher(repo)?;
let files_for_rev = list_rev_tracked_files(repo, to, narrow_matcher)
.map_err(handle_revlog_error)?;
repo.manually_set_parents(DirstateParents {
p1: repo.node(to).expect("update target should exist"),
p2: NULL_NODE,
})?;
// Filter the working copy according to the sparse spec
let tracked_files: Result<Vec<_>, _> = if !repo.has_sparse() {
files_for_rev.iter().collect()
} else {
// Ignore the warnings, they've been displayed by Python already
// TODO non-Python clients: display sparse warnings
let (sparse_matcher, _) = sparse::matcher(repo)?;
files_for_rev
.iter()
.filter(|f| {
match f {
Ok(f) => sparse_matcher.matches(f.0),
Err(_) => true, // Errors stop the update, include them
}
})
.collect()
};
let tracked_files = tracked_files?;
if tracked_files.is_empty() {
// Still write the dirstate because we might not be in the null
// revision.
// This can happen in narrow repos where all paths are excluded in
// this revision.
write_dirstate(repo)?;
return Ok(0);
}
let store_vfs = &repo.store_vfs();
let options = default_revlog_options(
repo.config(),
repo.requirements(),
crate::revlog::RevlogType::Filelog,
let (errors_sender, errors_receiver) = crossbeam_channel::unbounded();
let (files_sender, files_receiver) = crossbeam_channel::unbounded();
let working_directory_path = &repo.working_directory_path();
let files_count = tracked_files.len();
let chunks = chunk_tracked_files(tracked_files);
progress.update(0, Some(files_count as u64));
// TODO find a way (with `nix` or `signal-hook`?) of resetting the
// previous signal handler directly after. Currently, this is Python's
// job, but:
// - it introduces a (small) race between catching and resetting
// - it would break signal handlers in other contexts like `rhg``
let _ = ctrlc::set_handler(|| {
INTERRUPT_RECEIVED.store(true, Ordering::Relaxed)
});
create_working_copy(
chunks,
working_directory_path,
store_vfs,
options,
files_sender,
errors_sender,
progress,
workers,
// Reset the global interrupt now that we're done
if INTERRUPT_RECEIVED.swap(false, Ordering::Relaxed) {
// The threads have all exited early, let's re-raise
return Err(HgError::InterruptReceived);
}
let errors: Vec<HgError> = errors_receiver.iter().collect();
if !errors.is_empty() {
log::debug!("{} errors during update (see trace logs)", errors.len());
for error in errors.iter() {
log::trace!("{}", error);
}
// Best we can do is raise the first error (in order of the channel)
return Err(errors.into_iter().next().expect("can never be empty"));
}
// TODO try to run this concurrently to update the dirstate while we're
// still writing out the working copy to see if that improves performance.
let total = update_dirstate(repo, files_receiver)?;
write_dirstate(repo)?;
Ok(total)
}
fn handle_revlog_error(e: RevlogError) -> HgError {
match e {
crate::revlog::RevlogError::Other(hg_error) => hg_error,
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
e => HgError::abort(
format!("revlog error: {}", e),
exit_codes::ABORT,
None,
),
}
}
/// Preallocated size of Vec holding directory contents. This aims at
/// preventing the need for re-allocating the Vec in most cases.
///
/// The value is arbitrarily picked as a little over an average number of files
/// per directory done by looking at a few larger open-source repos.
/// Most of the runtime is IO anyway, so this doesn't matter too much.
const FILES_PER_DIRECTORY: usize = 16;
/// Chunk files per directory prefix, so almost every directory is handled
/// in a separate thread, which works around the FS inode mutex.
/// Chunking less (and doing approximately `files_count`/`threads`) actually
/// ends up being less performant: my hypothesis is `rayon`'s work stealing
/// being more efficient with tasks of varying lengths.
#[logging_timer::time("trace")]
fn chunk_tracked_files(
tracked_files: Vec<ExpandedManifestEntry>,
) -> Vec<(&HgPath, Vec<ExpandedManifestEntry>)> {
let files_count = tracked_files.len();
let mut chunks = Vec::with_capacity(files_count / FILES_PER_DIRECTORY);
let mut current_chunk = Vec::with_capacity(FILES_PER_DIRECTORY);
let mut last_directory = tracked_files[0].0.parent();
for file_info in tracked_files {
let current_directory = file_info.0.parent();
let different_directory = current_directory != last_directory;
if different_directory {
chunks.push((last_directory, current_chunk));
current_chunk = Vec::with_capacity(FILES_PER_DIRECTORY);
}
current_chunk.push(file_info);
last_directory = current_directory;
}
chunks.push((last_directory, current_chunk));
chunks
}
#[logging_timer::time("trace")]
#[allow(clippy::too_many_arguments)]
fn create_working_copy<'a: 'b, 'b>(
chunks: Vec<(&HgPath, Vec<ExpandedManifestEntry<'a>>)>,
working_directory_path: &Path,
store_vfs: &VfsImpl,
options: RevlogOpenOptions,
files_sender: Sender<(&'b HgPath, u32, usize, TruncatedTimestamp)>,
error_sender: Sender<HgError>,
progress: &dyn Progress,
workers: Option<usize>,
) {
let auditor = PathAuditor::new(working_directory_path);
let work_closure = |(dir_path, chunk)| -> Result<(), HgError> {
if let Err(e) = working_copy_worker(
dir_path,
chunk,
working_directory_path,
store_vfs,
options,
&files_sender,
progress,
&auditor,
) {
error_sender
.send(e)
.expect("channel should not be disconnected")
}
};
if let Some(workers) = workers {
if workers > 1 {
// Work in parallel, potentially restricting the number of threads
match rayon::ThreadPoolBuilder::new().num_threads(workers).build()
{
Err(error) => error_sender
.send(HgError::abort(
error.to_string(),
exit_codes::ABORT,
None,
))
.expect("channel should not be disconnected"),
Ok(pool) => {
log::trace!("restricting update to {} threads", workers);
pool.install(|| {
let _ =
chunks.into_par_iter().try_for_each(work_closure);
});
}
}
} else {
// Work sequentially, don't even invoke rayon
let _ = chunks.into_iter().try_for_each(work_closure);
}
} else {
// Work in parallel by default in the global threadpool
let _ = cap_default_rayon_threads();
let _ = chunks.into_par_iter().try_for_each(work_closure);
}
}
/// Represents a work unit for a single thread, responsible for this set of
/// files and restoring them to the working copy.
#[allow(clippy::too_many_arguments)]
fn working_copy_worker<'a: 'b, 'b>(
dir_path: &HgPath,
chunk: Vec<ExpandedManifestEntry<'a>>,
working_directory_path: &Path,
store_vfs: &VfsImpl,
options: RevlogOpenOptions,
files_sender: &Sender<(&'b HgPath, u32, usize, TruncatedTimestamp)>,
progress: &dyn Progress,
auditor: &PathAuditor,
) -> Result<(), HgError> {
let dir_path =
hg_path_to_path_buf(dir_path).expect("invalid path in manifest");
let dir_path = working_directory_path.join(dir_path);
std::fs::create_dir_all(&dir_path).when_writing_file(&dir_path)?;
if INTERRUPT_RECEIVED.load(Ordering::Relaxed) {
// Stop working, the user has requested that we stop
return Err(HgError::InterruptReceived);
}
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
for (file, file_node, flags) in chunk {
auditor.audit_path(file)?;
let flags = flags.map(|f| f.into());
let path =
working_directory_path.join(get_path_from_bytes(file.as_bytes()));
// Treemanifest is not supported
assert!(flags != Some(b't'));
let filelog = Filelog::open_vfs(store_vfs, file, options)?;
let filelog_revision_data = &filelog
.data_for_node(file_node)
.map_err(handle_revlog_error)?;
let file_data = filelog_revision_data.file_data()?;
if flags == Some(b'l') {
let target = get_path_from_bytes(file_data);
if let Err(e) = std::os::unix::fs::symlink(target, &path) {
// If the path already exists either:
// - another process created this file while ignoring the
// lock => error
// - our check for the fast path is incorrect => error
// - this is a malicious repo/bundle and this is symlink that
// tries to write things where it shouldn't be able to.
match e.kind() {
std::io::ErrorKind::AlreadyExists => {
let metadata = std::fs::symlink_metadata(&path)
.when_reading_file(&path)?;
if metadata.is_dir() {
return Err(HgError::Path(
HgPathError::TraversesSymbolicLink {
// Technically it should be one of the
// children, but good enough
path: file
.join(HgPath::new(b"*"))
.to_owned(),
symlink: file.to_owned(),
},
));
}
return Err(e).when_writing_file(&path);
}
_ => return Err(e).when_writing_file(&path),
}
}
} else {
let mut f =
std::fs::File::create(&path).when_writing_file(&path)?;
f.write_all(file_data).when_writing_file(&path)?;
}
if flags == Some(b'x') {
std::fs::set_permissions(&path, Permissions::from_mode(0o755))
.when_writing_file(&path)?;
}
let metadata =
std::fs::symlink_metadata(&path).when_reading_file(&path)?;
let mode = metadata.mode();
files_sender
.send((
file,
mode,
file_data.len(),
TruncatedTimestamp::for_mtime_of(&metadata)
.when_reading_file(&path)?,
))
.expect("channel should not be closed");
progress.increment(1, None);
}
Ok(())
}
#[logging_timer::time("trace")]
fn update_dirstate(
repo: &Repo,
files_receiver: Receiver<(&HgPath, u32, usize, TruncatedTimestamp)>,
) -> Result<usize, HgError> {
let mut dirstate = repo
.dirstate_map_mut()
.map_err(|e| HgError::abort(e.to_string(), exit_codes::ABORT, None))?;
// (see the comments in `filter_ambiguous_files` in `merge.py` for more)
// It turns out that (on Linux at least) the filesystem resolution time
// for most filesystems is based on the HZ kernel config. Their internal
// clocks do return nanoseconds if the hardware clock is precise enough,
// which should be the case on most recent computers but are only updated
// every few milliseconds at best (every "jiffy").
//
// We are still not concerned with fixing the race with other
// processes that might modify the working copy right after it was created
// within the same tick, because it is impossible to catch.
// However, we might as well not race with operations that could run right
// after this one, especially other Mercurial operations that could be
// waiting for the wlock to change file contents and the dirstate.
//
// Thus: wait until the filesystem clock has ticked to filter ambiguous
// entries and write the dirstate, but only for dirstate-v2, since v1 only
// has second-level granularity and waiting for a whole second is too much
// of a penalty in the general case.
// Although we're assuming that people running dirstate-v2 on Linux
// don't have a second-granularity FS (with the exclusion of NFS), users
// can be surprising, and at some point in the future dirstate-v2 will
// become the default. To that end, we limit the wait time to 100ms and
// fall back to the filter method in case of a timeout.
//
// +------------+------+--------------+
// | version | wait | filter level |
// +------------+------+--------------+
// | V1 | No | Second |
// | V2 | Yes | Nanosecond |
// | V2-slow-fs | No | Second |
// +------------+------+--------------+
let dirstate_v2 = repo.use_dirstate_v2();
// Let's ignore NFS right off the bat
let mut fast_enough_fs = !is_on_nfs_mount(repo.working_directory_path());
let fs_time_now = if dirstate_v2 && fast_enough_fs {
match wait_until_fs_tick(repo.working_directory_path()) {
None => None,
Some(Ok(time)) => Some(time),
Some(Err(time)) => {
fast_enough_fs = false;
Some(time)
}
}
} else {
filesystem_now(repo.working_directory_path())
.ok()
.map(TruncatedTimestamp::from)
};
let mut total = 0;
for (filename, mode, size, mtime) in files_receiver.into_iter() {
total += 1;
// When using dirstate-v2 on a filesystem with reasonable performance
// this is basically always true unless you get a mtime from the
// far future.
let has_meaningful_mtime = if let Some(fs_time) = fs_time_now {
mtime.for_reliable_mtime_of_self(&fs_time).is_some_and(|t| {
// Dirstate-v1 only has second-level information
!t.second_ambiguous || dirstate_v2 && fast_enough_fs
})
} else {
// We somehow failed to write to the filesystem, so don't store
// the cache information.
false
};
let reset = DirstateEntryReset {
filename,
wc_tracked: true,
p1_tracked: true,
p2_info: false,
has_meaningful_mtime,
parent_file_data_opt: Some(ParentFileData {
mode_size: Some((
mode,
size.try_into().expect("invalid file size in manifest"),
)),
mtime: Some(mtime),
}),
from_empty: true,
};
dirstate.reset_state(reset).map_err(|e| {
HgError::abort(e.to_string(), exit_codes::ABORT, None)
})?;
}
Ok(total)
}
/// Wait until the next update from the filesystem time by writing in a loop
/// a new temporary file inside the working directory and checking if its time
/// differs from the first one observed.
///
/// Returns `None` if we are unable to get the filesystem time,
/// `Some(Err(timestamp))` if we've timed out waiting for the filesystem clock
/// to tick, and `Some(Ok(timestamp))` if we've waited successfully.
///
/// On Linux, your average tick is going to be a "jiffy", or 1/HZ.
/// HZ is your kernel's tick rate (if it has one configured) and the value
/// is the one returned by `grep 'CONFIG_HZ=' /boot/config-$(uname -r)`,
/// again assuming a normal setup.
///
/// In my case (Alphare) at the time of writing, I get `CONFIG_HZ=250`,
/// which equates to 4ms.
///
/// This might change with a series that could make it to Linux 6.12:
/// https://lore.kernel.org/all/20241002-mgtime-v10-8-d1c4717f5284@kernel.org
fn wait_until_fs_tick(
working_directory_path: &Path,
) -> Option<Result<TruncatedTimestamp, TruncatedTimestamp>> {
let start = std::time::Instant::now();
let old_fs_time = filesystem_now(working_directory_path).ok()?;
let mut fs_time = filesystem_now(working_directory_path).ok()?;
const FS_TICK_WAIT_TIMEOUT: Duration = Duration::from_millis(100);
while fs_time == old_fs_time {
if std::time::Instant::now() - start > FS_TICK_WAIT_TIMEOUT {
log::trace!(
"timed out waiting for the fs clock to tick after {:?}",
FS_TICK_WAIT_TIMEOUT
);
return Some(Err(TruncatedTimestamp::from(old_fs_time)));
}
fs_time = filesystem_now(working_directory_path).ok()?;
}
log::trace!(
"waited for {:?} before writing the dirstate",
fs_time.duration_since(old_fs_time)
);
Some(Ok(TruncatedTimestamp::from(fs_time)))
}
#[cfg(test)]
mod test {
use super::*;
use pretty_assertions::assert_eq;
#[test]
fn test_chunk_tracked_files() {
fn chunk(v: Vec<&'static str>) -> Vec<ExpandedManifestEntry> {
v.into_iter()
.map(|f| (HgPath::new(f.as_bytes()), NULL_NODE, None))
.collect()
}
let p = HgPath::new;
let files = chunk(vec!["a"]);
let expected = vec![(p(""), chunk(vec!["a"]))];
assert_eq!(chunk_tracked_files(files), expected);
let files = chunk(vec!["a", "b", "c"]);
let expected = vec![(p(""), chunk(vec!["a", "b", "c"]))];
assert_eq!(chunk_tracked_files(files), expected);
let files = chunk(vec![
"dir/a-new",
"dir/a/mut",
"dir/a/mut-mut",
"dir/albert",
"dir/b",
"dir/subdir/c",
"dir/subdir/d",
"file",
]);
let expected = vec![
(p("dir"), chunk(vec!["dir/a-new"])),
(p("dir/a"), chunk(vec!["dir/a/mut", "dir/a/mut-mut"])),
(p("dir"), chunk(vec!["dir/albert", "dir/b"])),
(p("dir/subdir"), chunk(vec!["dir/subdir/c", "dir/subdir/d"])),
(p(""), chunk(vec!["file"])),
];
assert_eq!(chunk_tracked_files(files), expected);
// Doesn't get split
let large_dir = vec![
"1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12",
"13", "14", "15", "16", "17", "18", "19", "20", "21", "22", "23",
];
let files = chunk(large_dir.clone());
let expected = vec![(p(""), chunk(large_dir))];
assert_eq!(chunk_tracked_files(files), expected);
}
}