Skip to content

Commit 3a8c2c3

Browse files
twittnertwittner
committed
Read committee.toml when "until" feature is present.
Instead of assuming a port for the HTTP API read the address from the configuration. Also consolidate the until functionality in one section and do not require yapper to run in the timeboost demo, i.e. make the number of decrypted rounds configurable.
1 parent 7d54a20 commit 3a8c2c3

File tree

3 files changed

+181
-138
lines changed

3 files changed

+181
-138
lines changed

‎scripts/run-timeboost-demo‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -116,6 +116,7 @@ i=0
116116
for f in "$config_dir"/node_*.toml; do
117117
cmd=(target/release/timeboost
118118
--committee-id 0
119+
--committee "$config_dir/committee.toml"
119120
--config "$f"
120121
--until $rounds
121122
--watchdog-timeout 120)
@@ -124,8 +125,12 @@ for f in "$config_dir"/node_*.toml; do
124125
cmd+=(--ignore-stamp)
125126
fi
126127

127-
if $late_start; then
128-
cmd+=(--late-start --late-start-node-id 0)
128+
if $late_start && (( $i == 0 )); then
129+
cmd+=(--start-delay 5)
130+
fi
131+
132+
if $yapper; then
133+
cmd+=(--require-decrypt-rounds 3)
129134
fi
130135

131136
echo "${cmd[@]}"

‎timeboost-utils/src/until.rs‎

Lines changed: 112 additions & 72 deletions
Original file line numberDiff line numberDiff line change
@@ -1,92 +1,132 @@
1-
use anyhow::Result;
1+
use anyhow::{Result, bail};
2+
use futures::FutureExt;
3+
use reqwest::Url;
24
use std::time::{Duration, Instant};
3-
use tokio::signal;
5+
use timeboost_types::sailfish::RoundNumber;
6+
use tokio::time::sleep;
7+
use tokio::{select, signal};
8+
use tracing::{error, info};
49

510
const ROUND_TIMEOUT_SECS: u64 = 30;
611
const MAX_ROUND_TIMEOUTS: u64 = 15;
712

8-
pub async fn run_until(until: u64, timeout: u64, host: reqwest::Url) -> Result<()> {
9-
use futures::FutureExt;
10-
use tokio::time::sleep;
13+
#[derive(Debug)]
14+
pub struct Until {
15+
rounds: RoundNumber,
16+
duration: Duration,
17+
host: Url,
18+
require_decrypted: Option<RoundNumber>,
19+
}
1120

12-
sleep(Duration::from_secs(1)).await;
21+
impl Until {
22+
pub fn new<N>(r: N, d: Duration, host: Url) -> Self
23+
where
24+
N: Into<RoundNumber>,
25+
{
26+
Self {
27+
rounds: r.into(),
28+
duration: d,
29+
host,
30+
require_decrypted: None,
31+
}
32+
}
1333

14-
let mut timer = sleep(Duration::from_secs(timeout)).fuse().boxed();
34+
pub fn require_decrypted<N>(&mut self, rounds: Option<N>) -> &mut Self
35+
where
36+
N: Into<RoundNumber>,
37+
{
38+
self.require_decrypted = rounds.map(|r| r.into());
39+
self
40+
}
1541

16-
let mut req_timer = sleep(Duration::from_secs(1)).fuse().boxed();
42+
pub async fn run(self) -> Result<()> {
43+
let mut timer = sleep(self.duration).fuse().boxed();
1744

18-
let mut last_committed = 0;
19-
let mut last_committed_time = Instant::now();
20-
// Deliberately run this on a timeout to avoid a runaway testing scenario.
21-
loop {
22-
tokio::select! {
23-
_ = &mut timer => {
24-
tracing::error!("watchdog timed out, shutting down");
25-
anyhow::bail!("Watchdog timeout")
26-
}
27-
_ = &mut req_timer => {
28-
req_timer = sleep(Duration::from_secs(1)).fuse().boxed();
29-
if let Ok(resp) = reqwest::get(host.join("i/metrics").expect("valid url")).await {
30-
if let Ok(text) = resp.text().await {
31-
let committed_round = text
32-
.lines()
33-
.find(|line| line.starts_with("committed_round"))
34-
.and_then(|line| line.split(' ').nth(1))
35-
.and_then(|num| num.parse::<u64>().ok())
36-
.unwrap_or(0);
37-
38-
if committed_round > 0 && committed_round % 10 == 0 {
39-
tracing::info!("committed_round: {}", committed_round);
40-
}
45+
let mut req_timer = sleep(Duration::from_secs(1)).fuse().boxed();
4146

42-
let now = Instant::now();
43-
if committed_round == last_committed
44-
&& now.saturating_duration_since(last_committed_time) > Duration::from_secs(ROUND_TIMEOUT_SECS)
45-
{
46-
anyhow::bail!("Node stuck on round for more than {} seconds", ROUND_TIMEOUT_SECS)
47-
} else if committed_round > last_committed {
48-
last_committed = committed_round;
49-
last_committed_time = now;
50-
}
47+
let mut last_committed = 0;
48+
let mut last_committed_time = Instant::now();
5149

52-
let timeouts = text
53-
.lines()
54-
.find(|line| line.starts_with("rounds_timed_out"))
55-
.and_then(|line| line.split(' ').nth(1))
56-
.and_then(|num| num.parse::<u64>().ok())
57-
.unwrap_or(0);
50+
loop {
51+
select! {
52+
_ = &mut timer => {
53+
error!("watchdog timed out, shutting down");
54+
bail!("watchdog timeout")
55+
}
56+
_ = &mut req_timer => {
57+
req_timer = sleep(Duration::from_secs(1)).fuse().boxed();
58+
if let Ok(resp) = reqwest::get(self.host.join("i/metrics").expect("valid url")).await {
59+
if let Ok(text) = resp.text().await {
60+
let committed_round = text
61+
.lines()
62+
.find(|line| line.starts_with("committed_round"))
63+
.and_then(|line| line.split(' ').nth(1))
64+
.and_then(|num| num.parse::<u64>().ok())
65+
.unwrap_or(0);
5866

59-
if timeouts >= MAX_ROUND_TIMEOUTS {
60-
anyhow::bail!("Node timed out too many rounds")
61-
}
67+
if committed_round > 0 && committed_round % 10 == 0 {
68+
info!(%committed_round);
69+
}
6270

63-
let queued_encrypted = text
64-
.lines()
65-
.find(|line| line.starts_with("queued_encrypted_ilist"))
66-
.and_then(|line| line.split(' ').nth(1))
67-
.and_then(|num| num.parse::<u64>().ok())
68-
.unwrap_or(0);
69-
let output_decrypted = text
70-
.lines()
71-
.find(|line| line.starts_with("output_decrypted_ilist"))
72-
.and_then(|line| line.split(' ').nth(1))
73-
.and_then(|num| num.parse::<u64>().ok())
74-
.unwrap_or(0);
75-
76-
if committed_round >= until && output_decrypted > 0 {
77-
tracing::info!("committed_round: {}", committed_round);
78-
tracing::info!("enqueued encrypted: {}, output decrypted: {}", queued_encrypted, output_decrypted);
79-
tracing::info!("watchdog completed successfully");
80-
break;
71+
let now = Instant::now();
72+
73+
if committed_round == last_committed
74+
&& now.saturating_duration_since(last_committed_time) > Duration::from_secs(ROUND_TIMEOUT_SECS)
75+
{
76+
bail!("node stuck on round for more than {} seconds", ROUND_TIMEOUT_SECS)
77+
} else if committed_round > last_committed {
78+
last_committed = committed_round;
79+
last_committed_time = now;
80+
}
81+
82+
let timeouts = text
83+
.lines()
84+
.find(|line| line.starts_with("rounds_timed_out"))
85+
.and_then(|line| line.split(' ').nth(1))
86+
.and_then(|num| num.parse::<u64>().ok())
87+
.unwrap_or(0);
88+
89+
if timeouts >= MAX_ROUND_TIMEOUTS {
90+
bail!("node timed out too many rounds")
91+
}
92+
93+
let queued_encrypted = text
94+
.lines()
95+
.find(|line| line.starts_with("queued_encrypted_ilist"))
96+
.and_then(|line| line.split(' ').nth(1))
97+
.and_then(|num| num.parse::<u64>().ok())
98+
.unwrap_or(0);
99+
100+
let output_decrypted = text
101+
.lines()
102+
.find(|line| line.starts_with("output_decrypted_ilist"))
103+
.and_then(|line| line.split(' ').nth(1))
104+
.and_then(|num| num.parse::<u64>().ok())
105+
.unwrap_or(0);
106+
107+
if committed_round >= *self.rounds {
108+
if let Some(r) = self.require_decrypted {
109+
if output_decrypted > *r {
110+
info!(%committed_round);
111+
info!(%queued_encrypted, %output_decrypted);
112+
info!("watchdog completed successfully");
113+
break;
114+
}
115+
} else {
116+
info!(%committed_round);
117+
info!("watchdog completed successfully");
118+
break;
119+
}
120+
}
81121
}
82122
}
83123
}
84-
}
85-
_ = signal::ctrl_c() => {
86-
tracing::info!("ctrl-c received, shutting down");
87-
break;
124+
_ = signal::ctrl_c() => {
125+
tracing::info!("ctrl-c received, shutting down");
126+
break;
127+
}
88128
}
89129
}
130+
Ok(())
90131
}
91-
Ok(())
92132
}

0 commit comments

Comments
 (0)