fix: exit with error if gRPC port is busy

By default, the panic behaviour is to unwind, meaning we don't exit.
In case we fail to even start gRPC we should exit, as there's nothing
to do.

In this commit, we now internally raise a signal SIGUSR1 which then
triggers the shutdown of the reactor et all.

todo:
We should also consider catching any panic and do the same, or is
there any case where we'd want to unwind and remain running?

Signed-off-by: Tiago Castro <[email protected]>

Author: tiagolobocastro

io-engine/src/bin/io-engine.rs

@@ -167,9 +167,12 @@ fn start_tokio_runtime(args: &MayastorCliArgs) {
                 futures.push(Registration::run().boxed());
             }
 
-            futures::future::try_join_all(futures)
-                .await
-                .expect("runtime exited in the normal state");
+            if let Err(error) = futures::future::try_join_all(futures).await {
+                error!(error, "tokio runtime exited unexpectedly");
+                // todo: force process abort?
+                signal_hook::low_level::raise(signal_hook::consts::SIGUSR1)
+                    .expect("failed to raise internal error");
+            };
         });
     });
 }
@@ -313,5 +316,11 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     ms.fini();
     ms.event(EventAction::Start).generate();
+
+    if MayastorEnvironment::internal_aborted() {
+        tracing::error!("Exitting with error due to an internal abort request");
+        std::process::exit(1);
+    }
+
     Ok(())
 }

io-engine/src/core/env.rs

@@ -5,7 +5,7 @@ use std::{
     pin::Pin,
     str::FromStr,
     sync::{
-        atomic::{AtomicBool, Ordering::SeqCst},
+        atomic::{AtomicI32, Ordering::SeqCst},
         Arc, Mutex,
     },
     time::Duration,
@@ -14,7 +14,7 @@ use std::{
 use byte_unit::{Byte, Unit};
 use clap::Parser;
 use events_api::event::EventAction;
-use futures::{channel::oneshot, future};
+use futures::{channel::oneshot, future, TryFutureExt};
 use http::Uri;
 use once_cell::sync::{Lazy, OnceCell};
 use snafu::Snafu;
@@ -381,7 +381,7 @@ impl MayastorCliArgs {
 pub static GLOBAL_RC: Lazy<Arc<Mutex<i32>>> = Lazy::new(|| Arc::new(Mutex::new(-1)));
 
 /// keep track if we have received a signal already
-pub static SIG_RECEIVED: Lazy<AtomicBool> = Lazy::new(|| AtomicBool::new(false));
+pub static SIG_RECEIVED: Lazy<AtomicI32> = Lazy::new(|| AtomicI32::new(0));
 
 #[derive(Debug, Snafu)]
 pub enum EnvError {
@@ -575,12 +575,12 @@ unsafe extern "C" fn signal_trampoline(_: *mut c_void) {
 
 /// called on SIGINT and SIGTERM
 extern "C" fn mayastor_signal_handler(signo: i32) {
-    if SIG_RECEIVED.load(SeqCst) {
+    if SIG_RECEIVED.load(SeqCst) > 0 {
         return;
     }
 
-    warn!("Received SIGNO: {}", signo);
-    SIG_RECEIVED.store(true, SeqCst);
+    warn!("Received SIGNO: {signo}");
+    SIG_RECEIVED.store(signo, SeqCst);
     unsafe {
         if let Some(mth) = Mthread::primary_safe() {
             spdk_thread_send_critical_msg(mth.as_ptr(), Some(signal_trampoline));
@@ -669,19 +669,16 @@ impl MayastorEnvironment {
 
     /// configure signal handling
     fn install_signal_handlers(&self) {
-        unsafe {
-            signal_hook::low_level::register(signal_hook::consts::SIGTERM, || {
-                mayastor_signal_handler(1)
-            })
-        }
-        .unwrap();
-
-        unsafe {
-            signal_hook::low_level::register(signal_hook::consts::SIGINT, || {
-                mayastor_signal_handler(1)
-            })
+        for signal in [
+            signal_hook::consts::SIGTERM,
+            signal_hook::consts::SIGINT,
+            signal_hook::consts::SIGUSR1,
+        ] {
+            unsafe {
+                signal_hook::low_level::register(signal, move || mayastor_signal_handler(signal))
+            }
+            .expect("Failed to install handler for signal {signal}");
         }
-        .unwrap();
     }
 
     /// construct an array of options to be passed to EAL and start it
@@ -1128,7 +1125,7 @@ impl MayastorEnvironment {
     where
         F: FnOnce() + 'static,
     {
-        type FutureResult = Result<(), ()>;
+        type FutureResult = Result<(), Box<dyn std::error::Error + Send + Sync>>;
         let node_name = self.node_name.clone();
         let node_nqn = self.node_nqn.clone();
 
@@ -1166,7 +1163,9 @@ impl MayastorEnvironment {
                 )));
             }
             futures.push(Box::pin(subsys::Registration::run()));
-            futures.push(Box::pin(master));
+            futures.push(Box::pin(
+                master.map_err(|_| std::io::Error::other("").into()),
+            ));
             let _out = future::try_join_all(futures).await;
             info!("reactors stopped");
             ms.fini();
@@ -1179,6 +1178,11 @@ impl MayastorEnvironment {
     pub fn make_hostnqn(&self) -> Option<String> {
         self.node_nqn.clone()
     }
+
+    /// Check if we had to internally aborted.
+    pub fn internal_aborted() -> bool {
+        SIG_RECEIVED.load(SeqCst) == signal_hook::consts::SIGUSR1
+    }
 }
 
 fn make_hostnqn(node_name: Option<&String>) -> Option<String> {

io-engine/src/grpc/server.rs

@@ -54,7 +54,7 @@ impl MayastorGrpcServer {
         endpoint: std::net::SocketAddr,
         rpc_addr: String,
         api_versions: Vec<ApiVersion>,
-    ) -> Result<(), ()> {
+    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         let mut rcv_chan = Box::pin(Self::get_or_init().rcv_chan.clone());
 
         let address = Cow::from(rpc_addr);
@@ -64,10 +64,15 @@ impl MayastorGrpcServer {
 
         let enable_v0 = api_versions.contains(&ApiVersion::V0).then_some(true);
         let enable_v1 = api_versions.contains(&ApiVersion::V1).then_some(true);
-        info!(
-            "{:?} gRPC server configured at address {}",
-            api_versions, endpoint
-        );
+
+        let incoming =
+            tonic::transport::server::TcpIncoming::new(endpoint, true, None).map_err(|e| {
+                std::io::Error::new(
+                    std::io::ErrorKind::AddrInUse,
+                    format!("Failed to bind gRPC socket to {endpoint}: {e}"),
+                )
+            })?;
+        info!("{api_versions:?} gRPC server configured at address {endpoint}");
         let svc = Server::builder()
             .add_optional_service(
                 enable_v1.map(|_| v1::bdev::BdevRpcServer::new(BdevService::new())),
@@ -115,7 +120,7 @@ impl MayastorGrpcServer {
                 enable_v0.map(|_| JsonRpcServer::new(JsonRpcSvc::new(address.clone()))),
             )
             .add_optional_service(enable_v0.map(|_| BdevRpcServer::new(BdevSvc::new())))
-            .serve(endpoint);
+            .serve_with_incoming(incoming);
 
         select! {
             result = svc.fuse() => {
@@ -125,8 +130,8 @@ impl MayastorGrpcServer {
                         Ok(())
                     }
                     Err(e) => {
-                        error!("gRPC server failed with error: {}", e);
-                        Err(())
+                        error!("gRPC server failed with error: {e:#?}");
+                        Err(e.into())
                     }
                 }
             },

io-engine/src/subsys/registration/registration_grpc.rs

@@ -189,7 +189,7 @@ impl Registration {
 
     /// runner responsible for registering and
     /// de-registering the mayastor instance on shutdown
-    pub async fn run() -> Result<(), ()> {
+    pub async fn run() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         if let Some(registration) = GRPC_REGISTRATION.get() {
             registration.clone().run_loop().await;
         }