refactor: simplify control flow and error handling in metrics and netlink

This commit is contained in:
Nils Andreas Svee committed 2026-07-05 12:44:44 +07:00
1 parent 694cad1fac
commit 8b0c37bc83
3 files changed
+18 -20

No files matched your search

+6 -8
View File
@@ -78,8 +78,10 @@ impl Transport {
let full_data = format!("{}\n", data); let full_data = format!("{}\n", data);
// take the stream out, write, put it back or set None on failure // take the stream out, write, put it back or set None on failure
if let Some(s) = stream { let write_error = stream
if let Err(e) = std::io::Write::write_all(s, full_data.as_bytes()) { .as_mut()
.and_then(|s| std::io::Write::write_all(s, full_data.as_bytes()).err());
if let Some(e) = write_error {
warn!("TCP send failed: {}", e); warn!("TCP send failed: {}", e);
*stream = None; *stream = None;
*reconnect_backoff = (*reconnect_backoff * 2).min(MAX_RECONNECT_BACKOFF); *reconnect_backoff = (*reconnect_backoff * 2).min(MAX_RECONNECT_BACKOFF);
@@ -88,7 +90,6 @@ impl Transport {
} }
} }
} }
}
pub enum Metric { pub enum Metric {
Ping { Ping {
@@ -209,11 +210,8 @@ impl Metrics {
} }
// greedily drain without blocking // greedily drain without blocking
loop { while let Ok(metric) = self.metrics_rx.try_recv() {
match self.metrics_rx.try_recv() { batch.push(metric);
Ok(metric) => batch.push(metric),
Err(_) => break,
}
if batch.len() >= batch_size { if batch.len() >= batch_size {
break; break;
} }
+3 -5
View File
@@ -85,9 +85,9 @@ impl Netlink {
while let Some(reply) = iter.recv() { while let Some(reply) = iter.recv() {
let (header, attrs) = reply?; let (header, attrs) = reply?;
if header.ifindex == ifindex { if header.ifindex == ifindex
if let Ok(kind) = attrs.get_kind() { && let Ok(kind) = attrs.get_kind()
if kind && kind
.to_str() .to_str()
.map_err(|_| NetlinkError::NlQdiscError("Invalid UTF-8".to_string()))? .map_err(|_| NetlinkError::NlQdiscError("Invalid UTF-8".to_string()))?
== "cake" == "cake"
@@ -98,8 +98,6 @@ impl Netlink {
}); });
} }
} }
}
}
Err(NetlinkError::NoQdiscFound(ifname.to_string())) Err(NetlinkError::NoQdiscFound(ifname.to_string()))
} }
+9 -7
View File
@@ -446,8 +446,8 @@ impl Ratecontroller {
delta_delay_up: self.state_ul.delta_stat, delta_delay_up: self.state_ul.delta_stat,
}); });
if let Some(ref mut fd) = stats_fd { let stats_write_error = stats_fd.as_mut().and_then(|fd| {
if let Err(e) = fd.write_all( fd.write_all(
format!( format!(
"{},{},{},{},{},{},{},{}\n", "{},{},{},{},{},{},{},{}\n",
stats_time.secs(), stats_time.secs(),
@@ -460,16 +460,19 @@ impl Ratecontroller {
self.state_ul.current_rate as u64 self.state_ul.current_rate as u64
) )
.as_bytes(), .as_bytes(),
) { )
.err()
});
if let Some(e) = stats_write_error {
warn!("Failed to write statistics: {}", e); warn!("Failed to write statistics: {}", e);
} }
}
lastchg_t = now_t; lastchg_t = now_t;
} }
if let Some(ref mut fd) = speed_hist_fd { if let Some(ref mut fd) = speed_hist_fd
if now_t.duration_since(lastdump_t).as_secs_f64() > 300.0 { && now_t.duration_since(lastdump_t).as_secs_f64() > 300.0
{
for i in 0..self.config.speed_hist_size as usize { for i in 0..self.config.speed_hist_size as usize {
let hist_time = Time::new(ClockId::Realtime); let hist_time = Time::new(ClockId::Realtime);
if let Err(e) = fd.write_all( if let Err(e) = fd.write_all(
@@ -491,4 +494,3 @@ impl Ratecontroller {
} }
} }
} }
}