1use anyhow::Context;
105use std::io::IsTerminal;
106use tracing::instrument;
107
108mod auto_meta;
109pub mod chmod;
110pub mod cli;
111pub mod cmp;
112pub mod config;
113pub mod copy;
114pub mod copy_data;
115pub mod delete;
116pub mod dry_run;
117pub mod error;
118pub mod error_collector;
119pub mod filegen;
120pub mod filter;
121pub mod histogram_logger;
122pub mod histogram_panel;
123pub mod link;
124pub mod observability;
125pub mod preserve;
126pub mod remote_tracing;
127pub mod rm;
128mod runtime_setup;
129pub mod safedir;
130mod settings_parse;
131pub mod version;
132
133pub mod filecmp;
134pub mod progress;
135mod testutils;
136pub mod toctou_check;
137pub mod walk;
138pub mod walk_driver;
139
140pub use config::{
141 AutoMetaThrottleConfig, DryRunMode, DryRunWarnings, OutputConfig, RuntimeConfig,
142 ThrottleConfig, TracingConfig,
143};
144pub use congestion::{MetadataOp, Side};
149pub use progress::{RcpdProgressPrinter, SerializableProgress};
150pub use runtime_setup::{
154 collect_runtime_stats, generate_debug_log_filename, generate_trace_filename,
155};
156pub use settings_parse::{
157 parse_compare_settings, parse_metadata_cmp_settings, parse_preserve_settings,
158 validate_update_compare_vs_preserve,
159};
160
161#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, enum_map::Enum)]
163pub enum RcpdType {
164 Source,
165 Destination,
166}
167
168impl std::fmt::Display for RcpdType {
169 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
170 match self {
171 RcpdType::Source => write!(f, "source"),
172 RcpdType::Destination => write!(f, "destination"),
173 }
174 }
175}
176
177pub type ProgressSnapshot<T> = enum_map::EnumMap<RcpdType, T>;
179
180#[derive(Clone, Debug, Default, serde::Deserialize, serde::Serialize, PartialEq, Eq)]
182pub struct RuntimeStats {
183 pub cpu_time_user_ms: u64,
185 pub cpu_time_kernel_ms: u64,
187 pub peak_rss_bytes: u64,
189}
190
191#[derive(Debug, Default)]
193pub struct RemoteRuntimeStats {
194 pub source_host: String,
195 pub source_stats: RuntimeStats,
196 pub dest_host: String,
197 pub dest_stats: RuntimeStats,
198}
199
200#[must_use]
203pub fn is_localhost(host: &str) -> bool {
204 if host == "localhost" || host == "127.0.0.1" || host == "::1" || host == "[::1]" {
205 return true;
206 }
207 let mut buf = [0u8; 256];
209 let result = unsafe { libc::gethostname(buf.as_mut_ptr() as *mut libc::c_char, buf.len()) };
211 if result == 0
212 && let Ok(hostname_cstr) = std::ffi::CStr::from_bytes_until_nul(&buf)
213 && let Ok(hostname) = hostname_cstr.to_str()
214 && host == hostname
215 {
216 return true;
217 }
218 false
219}
220
221pub(crate) static PROGRESS: std::sync::LazyLock<progress::Progress> =
222 std::sync::LazyLock::new(progress::Progress::new);
223pub(crate) static PBAR: std::sync::LazyLock<indicatif::ProgressBar> =
224 std::sync::LazyLock::new(indicatif::ProgressBar::new_spinner);
225pub(crate) static REMOTE_RUNTIME_STATS: std::sync::LazyLock<
226 std::sync::Mutex<Option<RemoteRuntimeStats>>,
227> = std::sync::LazyLock::new(|| std::sync::Mutex::new(None));
228static HISTOGRAM_LOGGER_CANCEL: std::sync::Mutex<Option<tokio::sync::watch::Sender<bool>>> =
229 std::sync::Mutex::new(None);
230static HISTOGRAM_LOGGER_HANDLE: std::sync::Mutex<Option<tokio::task::JoinHandle<()>>> =
231 std::sync::Mutex::new(None);
232
233pub(crate) fn store_logger_cancel(tx: tokio::sync::watch::Sender<bool>) {
234 *HISTOGRAM_LOGGER_CANCEL
235 .lock()
236 .expect("histogram logger cancel mutex poisoned") = Some(tx);
237}
238
239pub(crate) fn store_logger_handle(handle: tokio::task::JoinHandle<()>) {
240 *HISTOGRAM_LOGGER_HANDLE
241 .lock()
242 .expect("histogram logger handle mutex poisoned") = Some(handle);
243}
244
245fn take_logger_handle() -> Option<tokio::task::JoinHandle<()>> {
246 HISTOGRAM_LOGGER_HANDLE
247 .lock()
248 .expect("histogram logger handle mutex poisoned")
249 .take()
250}
251
252fn signal_logger_cancel() {
253 if let Some(tx) = HISTOGRAM_LOGGER_CANCEL
254 .lock()
255 .expect("histogram logger cancel mutex poisoned")
256 .take()
257 && let Err(err) = tx.send(true)
258 {
259 tracing::debug!("histogram-logger cancel send failed (already gone): {err:#}");
260 }
261}
262
263#[must_use]
264pub fn get_progress() -> &'static progress::Progress {
265 &PROGRESS
266}
267
268pub fn set_remote_runtime_stats(stats: RemoteRuntimeStats) {
270 *REMOTE_RUNTIME_STATS.lock().unwrap() = Some(stats);
271}
272
273struct ProgressTracker {
274 lock_cvar: std::sync::Arc<(std::sync::Mutex<bool>, std::sync::Condvar)>,
275 pbar_thread: Option<std::thread::JoinHandle<()>>,
276}
277
278#[derive(Copy, Clone, Debug, Default, clap::ValueEnum)]
279pub enum ProgressType {
280 #[default]
281 #[value(name = "auto", alias = "Auto")]
282 Auto,
283 #[value(name = "ProgressBar", alias = "progress-bar")]
284 ProgressBar,
285 #[value(name = "TextUpdates", alias = "text-updates")]
286 TextUpdates,
287}
288
289pub enum GeneralProgressType {
290 User {
291 progress_type: ProgressType,
292 kind: progress::LocalProgressKind,
293 },
294 Remote(tokio::sync::mpsc::UnboundedSender<remote_tracing::TracingMessage>),
295 RemoteMaster {
296 progress_type: ProgressType,
297 get_progress_snapshot:
298 Box<dyn Fn() -> ProgressSnapshot<SerializableProgress> + Send + 'static>,
299 },
300}
301
302impl std::fmt::Debug for GeneralProgressType {
303 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
304 match self {
305 GeneralProgressType::User {
306 progress_type,
307 kind,
308 } => write!(f, "User(progress_type: {progress_type:?}, kind: {kind:?})"),
309 GeneralProgressType::Remote(_) => write!(f, "Remote(<sender>)"),
310 GeneralProgressType::RemoteMaster { progress_type, .. } => {
311 write!(
312 f,
313 "RemoteMaster(progress_type: {progress_type:?}, <function>)"
314 )
315 }
316 }
317 }
318}
319
320#[derive(Debug)]
321pub struct ProgressSettings {
322 pub progress_type: GeneralProgressType,
323 pub progress_delay: Option<String>,
324}
325
326fn progress_bar(
327 lock: &std::sync::Mutex<bool>,
328 cvar: &std::sync::Condvar,
329 delay_opt: &Option<std::time::Duration>,
330 kind: progress::LocalProgressKind,
331) {
332 let delay = delay_opt.unwrap_or(std::time::Duration::from_millis(200));
333 PBAR.set_style(
334 indicatif::ProgressStyle::with_template("{spinner:.cyan} {msg}")
335 .unwrap()
336 .tick_strings(&["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]),
337 );
338 let mut prog_printer = progress::make_local_printer(kind, &PROGRESS);
339 let mut is_done = lock.lock().unwrap();
340 loop {
341 PBAR.set_position(PBAR.position() + 1); let mut msg = prog_printer.print().unwrap();
343 msg.push_str(&observability::render_lines());
344 msg.push_str(&render_panel_from_registry());
345 PBAR.set_message(msg);
346 let result = cvar.wait_timeout(is_done, delay).unwrap();
347 is_done = result.0;
348 if *is_done {
349 break;
350 }
351 }
352 PBAR.finish_and_clear();
353}
354
355fn get_datetime_prefix() -> String {
356 chrono::Local::now()
357 .format("%Y-%m-%dT%H:%M:%S%.3f%:z")
358 .to_string()
359}
360
361fn text_updates(
362 lock: &std::sync::Mutex<bool>,
363 cvar: &std::sync::Condvar,
364 delay_opt: &Option<std::time::Duration>,
365 kind: progress::LocalProgressKind,
366) {
367 let delay = delay_opt.unwrap_or(std::time::Duration::from_secs(10));
368 let mut prog_printer = progress::make_local_printer(kind, &PROGRESS);
369 let mut is_done = lock.lock().unwrap();
370 loop {
371 eprintln!("=======================");
372 eprintln!(
373 "{}\n--{}{}{}",
374 get_datetime_prefix(),
375 prog_printer.print().unwrap(),
376 observability::render_lines(),
377 render_panel_from_registry(),
378 );
379 let result = cvar.wait_timeout(is_done, delay).unwrap();
380 is_done = result.0;
381 if *is_done {
382 break;
383 }
384 }
385}
386
387fn rcpd_updates(
388 lock: &std::sync::Mutex<bool>,
389 cvar: &std::sync::Condvar,
390 delay_opt: &Option<std::time::Duration>,
391 sender: tokio::sync::mpsc::UnboundedSender<remote_tracing::TracingMessage>,
392) {
393 tracing::debug!("Starting rcpd progress updates");
394 let delay = delay_opt.unwrap_or(std::time::Duration::from_millis(200));
395 let mut is_done = lock.lock().unwrap();
396 loop {
397 if remote_tracing::send_progress_update(&sender, &PROGRESS).is_err() {
398 tracing::debug!("Progress update channel closed, stopping progress updates");
400 break;
401 }
402 let result = cvar.wait_timeout(is_done, delay).unwrap();
403 is_done = result.0;
404 if *is_done {
405 break;
406 }
407 }
408}
409
410fn remote_master_updates<F>(
411 lock: &std::sync::Mutex<bool>,
412 cvar: &std::sync::Condvar,
413 delay_opt: &Option<std::time::Duration>,
414 get_progress_snapshot: F,
415 progress_type: ProgressType,
416) where
417 F: Fn() -> ProgressSnapshot<SerializableProgress> + Send + 'static,
418{
419 let interactive = match progress_type {
420 ProgressType::Auto => std::io::stderr().is_terminal(),
421 ProgressType::ProgressBar => true,
422 ProgressType::TextUpdates => false,
423 };
424 if interactive {
425 PBAR.set_style(
426 indicatif::ProgressStyle::with_template("{spinner:.cyan} {msg}")
427 .unwrap()
428 .tick_strings(&["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]),
429 );
430 let delay = delay_opt.unwrap_or(std::time::Duration::from_millis(200));
431 let mut printer = RcpdProgressPrinter::new();
432 let mut is_done = lock.lock().unwrap();
433 loop {
434 let progress_map = get_progress_snapshot();
435 let source_progress = &progress_map[RcpdType::Source];
436 let destination_progress = &progress_map[RcpdType::Destination];
437 PBAR.set_position(PBAR.position() + 1); let mut msg = printer
439 .print(source_progress, destination_progress)
440 .unwrap();
441 msg.push_str(&render_panel_from_registry());
442 PBAR.set_message(msg);
443 let result = cvar.wait_timeout(is_done, delay).unwrap();
444 is_done = result.0;
445 if *is_done {
446 break;
447 }
448 }
449 PBAR.finish_and_clear();
450 } else {
451 let delay = delay_opt.unwrap_or(std::time::Duration::from_secs(10));
452 let mut printer = RcpdProgressPrinter::new();
453 let mut is_done = lock.lock().unwrap();
454 loop {
455 let progress_map = get_progress_snapshot();
456 let source_progress = &progress_map[RcpdType::Source];
457 let destination_progress = &progress_map[RcpdType::Destination];
458 eprintln!("=======================");
459 eprintln!(
460 "{}\n--{}{}",
461 get_datetime_prefix(),
462 printer
463 .print(source_progress, destination_progress)
464 .unwrap(),
465 render_panel_from_registry(),
466 );
467 let result = cvar.wait_timeout(is_done, delay).unwrap();
468 is_done = result.0;
469 if *is_done {
470 break;
471 }
472 }
473 }
474}
475
476impl ProgressTracker {
477 pub fn new(progress_type: GeneralProgressType, delay_opt: Option<std::time::Duration>) -> Self {
478 let lock_cvar =
479 std::sync::Arc::new((std::sync::Mutex::new(false), std::sync::Condvar::new()));
480 let lock_cvar_clone = lock_cvar.clone();
481 let pbar_thread = std::thread::spawn(move || {
482 let (lock, cvar) = &*lock_cvar_clone;
483 match progress_type {
484 GeneralProgressType::Remote(sender) => {
485 rcpd_updates(lock, cvar, &delay_opt, sender);
486 }
487 GeneralProgressType::RemoteMaster {
488 progress_type,
489 get_progress_snapshot,
490 } => {
491 remote_master_updates(
492 lock,
493 cvar,
494 &delay_opt,
495 get_progress_snapshot,
496 progress_type,
497 );
498 }
499 GeneralProgressType::User {
500 progress_type,
501 kind,
502 } => {
503 let interactive = match progress_type {
504 ProgressType::Auto => std::io::stderr().is_terminal(),
505 ProgressType::ProgressBar => true,
506 ProgressType::TextUpdates => false,
507 };
508 if interactive {
509 progress_bar(lock, cvar, &delay_opt, kind);
510 } else {
511 text_updates(lock, cvar, &delay_opt, kind);
512 }
513 }
514 }
515 });
516 Self {
517 lock_cvar,
518 pbar_thread: Some(pbar_thread),
519 }
520 }
521}
522
523impl Drop for ProgressTracker {
524 fn drop(&mut self) {
525 let (lock, cvar) = &*self.lock_cvar;
526 let mut is_done = lock.lock().unwrap();
527 *is_done = true;
528 cvar.notify_one();
529 drop(is_done);
530 if let Some(pbar_thread) = self.pbar_thread.take() {
531 pbar_thread.join().unwrap();
532 }
533 }
534}
535
536pub async fn cmp(
537 src: &std::path::Path,
538 dst: &std::path::Path,
539 log: &cmp::LogWriter,
540 settings: &cmp::Settings,
541) -> Result<cmp::Summary, anyhow::Error> {
542 cmp::cmp(&PROGRESS, src, dst, log, settings).await
543}
544
545pub async fn copy(
546 src: &std::path::Path,
547 dst: &std::path::Path,
548 settings: ©::Settings,
549 preserve: &preserve::Settings,
550) -> Result<copy::Summary, copy::Error> {
551 copy::copy(&PROGRESS, src, dst, settings, preserve, false).await
552}
553
554pub async fn rm(path: &std::path::Path, settings: &rm::Settings) -> Result<rm::Summary, rm::Error> {
555 rm::rm(&PROGRESS, path, settings).await
556}
557
558pub async fn chmod(
559 path: &std::path::Path,
560 settings: &chmod::Settings,
561) -> Result<chmod::Summary, chmod::Error> {
562 chmod::chmod(&PROGRESS, path, settings).await
563}
564
565pub async fn link(
566 src: &std::path::Path,
567 dst: &std::path::Path,
568 update: &Option<std::path::PathBuf>,
569 settings: &link::Settings,
570) -> Result<link::Summary, link::Error> {
571 let cwd = std::env::current_dir()
572 .with_context(|| "failed to get current working directory")
573 .map_err(|err| link::Error::new(err, link::Summary::default()))?;
574 link::link(&PROGRESS, &cwd, src, dst, update, settings, false).await
575}
576
577fn render_panel_from_registry() -> String {
578 let entries = observability::registered_histograms();
579 if entries.is_empty() {
580 return String::new();
581 }
582 let snapshots: Vec<hdrhistogram::Histogram<u64>> = entries
583 .iter()
584 .map(|e| (*e.snapshot_rx.borrow()).clone())
585 .collect();
586 let units: Vec<histogram_panel::PanelUnit> = entries
587 .iter()
588 .zip(snapshots.iter())
589 .map(|(e, snap)| histogram_panel::PanelUnit {
590 label: e.label,
591 histogram: snap,
592 interval: e.interval,
593 })
594 .collect();
595 histogram_panel::render_histogram_panel(&units)
596}
597
598#[instrument(skip(func))] pub fn run<Fut, Summary, Error>(
600 progress: Option<ProgressSettings>,
601 output: OutputConfig,
602 runtime_config: RuntimeConfig,
603 throttle_config: ThrottleConfig,
604 tracing_config: TracingConfig,
605 func: impl FnOnce() -> Fut,
606) -> Option<Summary>
607where
609 Summary: std::fmt::Display,
610 Error: std::fmt::Display + std::fmt::Debug,
611 Fut: std::future::Future<Output = Result<Summary, Error>>,
612{
613 let _ = get_progress();
617 if let Err(e) = throttle_config.validate() {
618 eprintln!("Configuration error: {e}");
619 return None;
620 }
621 let OutputConfig {
622 quiet,
623 verbose,
624 print_summary,
625 suppress_runtime_stats,
626 } = output;
627 let trace_identifier = tracing_config.trace_identifier.clone();
630 if let Err(e) =
631 runtime_setup::validate_histogram_log_target(&throttle_config, &trace_identifier)
632 {
633 eprintln!("Configuration error: {e}");
634 return None;
635 }
636 let _tracing_guards = runtime_setup::install_tracing_subscriber(quiet, verbose, tracing_config);
637 let res = {
638 let runtime = runtime_setup::build_tokio_runtime(&runtime_config, &throttle_config);
639 runtime_setup::spawn_throttle_replenishers(&runtime, &throttle_config, &trace_identifier);
640 let res = {
641 let _progress_tracker = progress.map(|settings| {
642 tracing::debug!("Requesting progress updates {settings:?}");
643 let delay = settings.progress_delay.map(|delay_str| {
644 humantime::parse_duration(&delay_str)
645 .expect("Couldn't parse duration out of --progress-delay")
646 });
647 ProgressTracker::new(settings.progress_type, delay)
648 });
649 runtime.block_on(func())
650 };
651 match &res {
652 Ok(summary) => {
653 if print_summary || verbose > 0 {
654 println!("{summary}");
655 }
656 }
657 Err(err) => {
658 if !quiet {
659 println!("{err:?}");
660 }
661 }
662 }
663 if (print_summary || verbose > 0)
664 && !suppress_runtime_stats
665 && let Err(err) = runtime_setup::print_runtime_stats()
666 {
667 println!("Failed to print runtime stats: {err:?}");
668 }
669 signal_logger_cancel();
673 if let Some(handle) = take_logger_handle() {
674 let _ = runtime.block_on(async {
677 tokio::time::timeout(std::time::Duration::from_secs(1), handle).await
678 });
679 }
680 res
681 };
684 reset_process_throttle_state();
689 res.ok()
690}
691
692fn reset_process_throttle_state() {
698 congestion::clear_sample_sink();
699 observability::clear();
700 for &side in &throttle::Side::ALL {
701 for &op in &throttle::MetadataOp::ALL {
702 throttle::set_max_ops_in_flight(throttle::Resource::meta(side, op), 0);
703 }
704 }
705 throttle::disable_ops_throttle();
706 throttle::set_max_open_files(0);
713 throttle::init_iops_tokens(0);
714}