diff options
author | Alex Auvolat <alex@adnab.me> | 2023-09-12 14:35:48 +0200 |
---|---|---|
committer | Alex Auvolat <alex@adnab.me> | 2023-09-12 14:35:48 +0200 |
commit | 91e764a2bf7a479a068fdfdad020cd79b5e86b88 (patch) | |
tree | 4ab75c134805d73c5b1cf8121b7dcb551303b452 | |
parent | aa79810596d8608c1636006018f31b34197c2123 (diff) | |
download | garage-91e764a2bf7a479a068fdfdad020cd79b5e86b88.tar.gz garage-91e764a2bf7a479a068fdfdad020cd79b5e86b88.zip |
fix hang on shutdown
-rw-r--r-- | src/rpc/system.rs | 18 |
1 files changed, 9 insertions, 9 deletions
diff --git a/src/rpc/system.rs b/src/rpc/system.rs index 36b58311..4daa5ba9 100644 --- a/src/rpc/system.rs +++ b/src/rpc/system.rs @@ -9,10 +9,10 @@ use std::time::{Duration, Instant}; use arc_swap::ArcSwap; use async_trait::async_trait; -use futures::{join, select}; -use futures_util::future::*; +use futures::join; use serde::{Deserialize, Serialize}; use sodiumoxide::crypto::sign::ed25519; +use tokio::select; use tokio::sync::watch; use tokio::sync::Mutex; @@ -702,7 +702,7 @@ impl System { async fn status_exchange_loop(&self, mut stop_signal: watch::Receiver<bool>) { while !*stop_signal.borrow() { - let restart_at = tokio::time::sleep(STATUS_EXCHANGE_INTERVAL); + let restart_at = Instant::now() + STATUS_EXCHANGE_INTERVAL; self.update_local_status(); let local_status: NodeStatus = self.local_status.load().as_ref().clone(); @@ -711,13 +711,14 @@ impl System { .broadcast( &self.system_endpoint, SystemRpc::AdvertiseStatus(local_status), - RequestStrategy::with_priority(PRIO_HIGH), + RequestStrategy::with_priority(PRIO_HIGH) + .with_custom_timeout(STATUS_EXCHANGE_INTERVAL), ) .await; select! { - _ = restart_at.fuse() => {}, - _ = stop_signal.changed().fuse() => {}, + _ = tokio::time::sleep_until(restart_at.into()) => {}, + _ = stop_signal.changed() => {}, } } } @@ -799,10 +800,9 @@ impl System { #[cfg(feature = "kubernetes-discovery")] tokio::spawn(self.clone().advertise_to_kubernetes()); - let restart_at = tokio::time::sleep(DISCOVERY_INTERVAL); select! { - _ = restart_at.fuse() => {}, - _ = stop_signal.changed().fuse() => {}, + _ = tokio::time::sleep(DISCOVERY_INTERVAL) => {}, + _ = stop_signal.changed() => {}, } } } |