From aa79810596d8608c1636006018f31b34197c2123 Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Tue, 12 Sep 2023 14:35:19 +0200 Subject: [PATCH 1/2] Fix error when none of S3/K2V/WEB/ADMIN server is started (fix #613) --- src/garage/server.rs | 23 +++++++++++++++-------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/src/garage/server.rs b/src/garage/server.rs index 958089c6..472616c7 100644 --- a/src/garage/server.rs +++ b/src/garage/server.rs @@ -130,20 +130,27 @@ pub async fn run_server(config_file: PathBuf, secrets: Secrets) -> Result<(), Er warn!("This Garage version is built without the metrics feature"); } - // Stuff runs + if servers.is_empty() { + // Nothing runs except netapp (not in servers) + // Await shutdown signal before proceeding to shutting down netapp + wait_from(watch_cancel).await; + } else { + // Stuff runs - // When a cancel signal is sent, stuff stops + // When a cancel signal is sent, stuff stops - // Collect stuff - for (desc, join_handle) in servers { - if let Err(e) = join_handle.await? { - error!("{} server exited with error: {}", desc, e); - } else { - info!("{} server exited without error.", desc); + // Collect stuff + for (desc, join_handle) in servers { + if let Err(e) = join_handle.await? { + error!("{} server exited with error: {}", desc, e); + } else { + info!("{} server exited without error.", desc); + } } } // Remove RPC handlers for system to break reference cycles + info!("Deregistering RPC handlers for shutdown..."); garage.system.netapp.drop_all_handlers(); opentelemetry::global::shutdown_tracer_provider(); From 91e764a2bf7a479a068fdfdad020cd79b5e86b88 Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Tue, 12 Sep 2023 14:35:48 +0200 Subject: [PATCH 2/2] fix hang on shutdown --- src/rpc/system.rs | 18 +++++++++--------- 1 file 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) { 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() => {}, } } }