cli: Add timing information to `rad-sync`
This commit is contained in:
parent
ddd4bde302
commit
f15afa84be
|
|
@ -64,6 +64,7 @@ Options
|
||||||
--seed <nid> Sync with the given node (may be specified multiple times)
|
--seed <nid> Sync with the given node (may be specified multiple times)
|
||||||
-r, --replicas <count> Sync with a specific number of seeds
|
-r, --replicas <count> Sync with a specific number of seeds
|
||||||
-v, --verbose Verbose output
|
-v, --verbose Verbose output
|
||||||
|
--debug Print debug information afer sync
|
||||||
--help Print help
|
--help Print help
|
||||||
"#,
|
"#,
|
||||||
};
|
};
|
||||||
|
|
@ -171,6 +172,7 @@ pub enum SyncDirection {
|
||||||
#[derive(Default, Debug)]
|
#[derive(Default, Debug)]
|
||||||
pub struct Options {
|
pub struct Options {
|
||||||
pub rid: Option<RepoId>,
|
pub rid: Option<RepoId>,
|
||||||
|
pub debug: bool,
|
||||||
pub verbose: bool,
|
pub verbose: bool,
|
||||||
pub timeout: time::Duration,
|
pub timeout: time::Duration,
|
||||||
pub sort_by: SortBy,
|
pub sort_by: SortBy,
|
||||||
|
|
@ -188,6 +190,7 @@ impl Args for Options {
|
||||||
let mut fetch = false;
|
let mut fetch = false;
|
||||||
let mut announce = false;
|
let mut announce = false;
|
||||||
let mut inventory = false;
|
let mut inventory = false;
|
||||||
|
let mut debug = false;
|
||||||
let mut replicas = None;
|
let mut replicas = None;
|
||||||
let mut seeds = BTreeSet::new();
|
let mut seeds = BTreeSet::new();
|
||||||
let mut sort_by = SortBy::default();
|
let mut sort_by = SortBy::default();
|
||||||
|
|
@ -195,6 +198,9 @@ impl Args for Options {
|
||||||
|
|
||||||
while let Some(arg) = parser.next()? {
|
while let Some(arg) = parser.next()? {
|
||||||
match arg {
|
match arg {
|
||||||
|
Long("debug") => {
|
||||||
|
debug = true;
|
||||||
|
}
|
||||||
Long("verbose") | Short('v') => {
|
Long("verbose") | Short('v') => {
|
||||||
verbose = true;
|
verbose = true;
|
||||||
}
|
}
|
||||||
|
|
@ -279,6 +285,7 @@ impl Args for Options {
|
||||||
Ok((
|
Ok((
|
||||||
Options {
|
Options {
|
||||||
rid,
|
rid,
|
||||||
|
debug,
|
||||||
verbose,
|
verbose,
|
||||||
timeout,
|
timeout,
|
||||||
sort_by,
|
sort_by,
|
||||||
|
|
@ -334,7 +341,14 @@ pub fn run(options: Options, ctx: impl term::Context) -> anyhow::Result<()> {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if [SyncDirection::Announce, SyncDirection::Both].contains(&direction) {
|
if [SyncDirection::Announce, SyncDirection::Both].contains(&direction) {
|
||||||
announce_refs(rid, settings, options.timeout, &mut node, &profile)?;
|
announce_refs(
|
||||||
|
rid,
|
||||||
|
settings,
|
||||||
|
options.timeout,
|
||||||
|
options.debug,
|
||||||
|
&mut node,
|
||||||
|
&profile,
|
||||||
|
)?;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Operation::Synchronize(SyncMode::Inventory) => {
|
Operation::Synchronize(SyncMode::Inventory) => {
|
||||||
|
|
@ -417,6 +431,7 @@ fn announce_refs(
|
||||||
rid: RepoId,
|
rid: RepoId,
|
||||||
settings: RepoSync,
|
settings: RepoSync,
|
||||||
timeout: time::Duration,
|
timeout: time::Duration,
|
||||||
|
debug: bool,
|
||||||
node: &mut Node,
|
node: &mut Node,
|
||||||
profile: &Profile,
|
profile: &Profile,
|
||||||
) -> anyhow::Result<()> {
|
) -> anyhow::Result<()> {
|
||||||
|
|
@ -469,18 +484,18 @@ fn announce_refs(
|
||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
|
|
||||||
let mut spinner = term::spinner(format!("Syncing with {} node(s)..", unsynced.len()));
|
let mut spinner = term::spinner(format!("Found {} seed(s)..", unsynced.len()));
|
||||||
let result = node.announce(rid, unsynced, timeout, |event, replicas| match event {
|
let result = node.announce(rid, unsynced, timeout, |event, replicas| match event {
|
||||||
node::AnnounceEvent::Announced => ControlFlow::Continue(()),
|
node::AnnounceEvent::Announced => ControlFlow::Continue(()),
|
||||||
node::AnnounceEvent::RefsSynced { remote } => {
|
node::AnnounceEvent::RefsSynced { remote, time } => {
|
||||||
spinner.message(format!("Synced with {remote}.."));
|
spinner.message(format!("Synced with {remote} in {time:?}.."));
|
||||||
|
|
||||||
// We're done syncing when both of these conditions are met:
|
// We're done syncing when both of these conditions are met:
|
||||||
//
|
//
|
||||||
// 1. We've matched or exceeded our target replica count.
|
// 1. We've matched or exceeded our target replica count.
|
||||||
// 2. We've synced with the seeds specified manually.
|
// 2. We've synced with the seeds specified manually.
|
||||||
if replicas.len() >= settings.replicas
|
if replicas.len() >= settings.replicas
|
||||||
&& settings.seeds.iter().all(|s| replicas.contains(s))
|
&& settings.seeds.iter().all(|s| replicas.contains_key(s))
|
||||||
{
|
{
|
||||||
ControlFlow::Break(())
|
ControlFlow::Break(())
|
||||||
} else {
|
} else {
|
||||||
|
|
@ -494,6 +509,14 @@ fn announce_refs(
|
||||||
} else {
|
} else {
|
||||||
spinner.message(format!("Synced with {} node(s)", result.synced.len()));
|
spinner.message(format!("Synced with {} node(s)", result.synced.len()));
|
||||||
spinner.finish();
|
spinner.finish();
|
||||||
|
if debug {
|
||||||
|
for (seed, time) in &result.synced {
|
||||||
|
term::println(
|
||||||
|
" ",
|
||||||
|
term::format::dim(format!("Synced with {seed} in {time:?}")),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
for seed in result.timeout {
|
for seed in result.timeout {
|
||||||
term::notice!("Seed {seed} timed out..");
|
term::notice!("Seed {seed} timed out..");
|
||||||
|
|
|
||||||
|
|
@ -36,8 +36,8 @@ fn announce_(rid: RepoId, node: &mut Node) -> Result<(), radicle::node::Error> {
|
||||||
Duration::from_secs(9),
|
Duration::from_secs(9),
|
||||||
|event, _| match event {
|
|event, _| match event {
|
||||||
node::AnnounceEvent::Announced => ControlFlow::Continue(()),
|
node::AnnounceEvent::Announced => ControlFlow::Continue(()),
|
||||||
node::AnnounceEvent::RefsSynced { remote } => {
|
node::AnnounceEvent::RefsSynced { remote, time } => {
|
||||||
spinner.message(format!("Synced with {remote}.."));
|
spinner.message(format!("Synced with {remote} in {time:?}.."));
|
||||||
ControlFlow::Continue(())
|
ControlFlow::Continue(())
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
|
|
|
||||||
|
|
@ -706,9 +706,12 @@ fn sync(
|
||||||
DEFAULT_SYNC_TIMEOUT,
|
DEFAULT_SYNC_TIMEOUT,
|
||||||
|event, _| match event {
|
|event, _| match event {
|
||||||
node::AnnounceEvent::Announced => ControlFlow::Continue(()),
|
node::AnnounceEvent::Announced => ControlFlow::Continue(()),
|
||||||
node::AnnounceEvent::RefsSynced { remote } => {
|
node::AnnounceEvent::RefsSynced { remote, time } => {
|
||||||
replicated.insert(remote);
|
replicated.insert(remote);
|
||||||
spinner.message(format!("Synced with {}..", cli::format::dim(remote)));
|
spinner.message(format!(
|
||||||
|
"Synced with {} in {time:?}..",
|
||||||
|
cli::format::dim(remote)
|
||||||
|
));
|
||||||
ControlFlow::Continue(())
|
ControlFlow::Continue(())
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
|
|
|
||||||
|
|
@ -633,14 +633,17 @@ pub struct AnnounceResult {
|
||||||
/// Nodes that timed out.
|
/// Nodes that timed out.
|
||||||
pub timeout: Vec<NodeId>,
|
pub timeout: Vec<NodeId>,
|
||||||
/// Nodes that synced.
|
/// Nodes that synced.
|
||||||
pub synced: Vec<NodeId>,
|
pub synced: Vec<(NodeId, time::Duration)>,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// A sync event, emitted by [`Node::announce`].
|
/// A sync event, emitted by [`Node::announce`].
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub enum AnnounceEvent {
|
pub enum AnnounceEvent {
|
||||||
/// Refs were synced with the given node.
|
/// Refs were synced with the given node.
|
||||||
RefsSynced { remote: NodeId },
|
RefsSynced {
|
||||||
|
remote: NodeId,
|
||||||
|
time: time::Duration,
|
||||||
|
},
|
||||||
/// Refs were announced to all given nodes.
|
/// Refs were announced to all given nodes.
|
||||||
Announced,
|
Announced,
|
||||||
}
|
}
|
||||||
|
|
@ -914,14 +917,15 @@ impl Node {
|
||||||
rid: RepoId,
|
rid: RepoId,
|
||||||
seeds: impl IntoIterator<Item = NodeId>,
|
seeds: impl IntoIterator<Item = NodeId>,
|
||||||
timeout: time::Duration,
|
timeout: time::Duration,
|
||||||
mut callback: impl FnMut(AnnounceEvent, &HashSet<PublicKey>) -> ControlFlow<()>,
|
mut callback: impl FnMut(AnnounceEvent, &HashMap<PublicKey, time::Duration>) -> ControlFlow<()>,
|
||||||
) -> Result<AnnounceResult, Error> {
|
) -> Result<AnnounceResult, Error> {
|
||||||
let events = self.subscribe(timeout)?;
|
let events = self.subscribe(timeout)?;
|
||||||
let refs = self.announce_refs(rid)?;
|
let refs = self.announce_refs(rid)?;
|
||||||
|
|
||||||
let mut unsynced = seeds.into_iter().collect::<BTreeSet<_>>();
|
let mut unsynced = seeds.into_iter().collect::<BTreeSet<_>>();
|
||||||
let mut synced = HashSet::new();
|
let mut synced = HashMap::new();
|
||||||
let mut timeout: Vec<NodeId> = Vec::new();
|
let mut timeout: Vec<NodeId> = Vec::new();
|
||||||
|
let started = time::Instant::now();
|
||||||
|
|
||||||
callback(AnnounceEvent::Announced, &synced);
|
callback(AnnounceEvent::Announced, &synced);
|
||||||
|
|
||||||
|
|
@ -932,13 +936,22 @@ impl Node {
|
||||||
rid: rid_,
|
rid: rid_,
|
||||||
at,
|
at,
|
||||||
}) if rid == rid_ && refs.at == at => {
|
}) if rid == rid_ && refs.at == at => {
|
||||||
|
let elapsed = started.elapsed();
|
||||||
log::debug!(target: "radicle", "Received {e:?}");
|
log::debug!(target: "radicle", "Received {e:?}");
|
||||||
|
|
||||||
unsynced.remove(&remote);
|
unsynced.remove(&remote);
|
||||||
// We can receive synced events from nodes we didn't directly announce to,
|
// We can receive synced events from nodes we didn't directly announce to,
|
||||||
// and it's possible to receive duplicates as well.
|
// and it's possible to receive duplicates as well.
|
||||||
if synced.insert(remote) {
|
if synced.insert(remote, elapsed).is_none() {
|
||||||
if callback(AnnounceEvent::RefsSynced { remote }, &synced).is_break() {
|
if callback(
|
||||||
|
AnnounceEvent::RefsSynced {
|
||||||
|
remote,
|
||||||
|
time: elapsed,
|
||||||
|
},
|
||||||
|
&synced,
|
||||||
|
)
|
||||||
|
.is_break()
|
||||||
|
{
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue