Add SIGTERM handler for a graceful shutdown

Changes the wait_for_connection function in the Listen trait to be
non-blocking. That allows checking in the main loop if the SIGTERM
signal has been raised.

Signed-off-by: Hugues de Valon <[email protected]>

Author: hug-dev

Cargo.lock

@@ -5,13 +5,18 @@ authors = ["Ionut Mihalcea <[email protected]>",
            "Hugues de Valon <[email protected]>"]
 edition = "2018"
 
+[[bin]]
+name = "parsec"
+path = "src/bin/main.rs"
+
 [dependencies]
 parsec-interface = { git = "https://github.com/parallaxsecond/parsec-interface-rs", tag = "0.1.0"  }
 rand = "0.7.2"
 base64 = "0.10.1"
 uuid = "0.7.4"
 threadpool = "1.7.1"
 std-semaphore = "0.1.0"
+signal-hook = "0.1.10"
 
 [dev-dependencies]
 parsec-client-test = { git = "https://github.com/parallaxsecond/parsec-client-test", tag = "0.1.2"  }

Cargo.toml

@@ -138,6 +138,7 @@ This project uses the following third party crates:
 * threadpool (Apache-2.0)
 * std-semaphore (MIT and Apache-2.0)
 * num_cpus (MIT and Apache-2.0)
+* signal-hook (MIT and Apache-2.0)
 
 This project uses the following third party libraries:
 * [Mbed Crypto](https://github.com/ARMmbed/mbed-crypto) (Apache-2.0)

README.md

@@ -25,9 +25,14 @@ use parsec::providers::{
 use parsec_interface::operations_protobuf::ProtobufConverter;
 use parsec_interface::requests::AuthType;
 use parsec_interface::requests::{BodyType, ProviderID};
+use signal_hook::flag;
+use signal_hook::SIGTERM;
+use std::io::Error;
 use std::path::PathBuf;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use std::sync::RwLock;
+use std::thread;
 use std::time::Duration;
 use threadpool::Builder;
 
@@ -91,23 +96,38 @@ fn build_components() -> (FrontEndHandler, impl Listen) {
     (front_end, listener)
 }
 
-fn main() {
+fn main() -> Result<(), Error> {
     let (front_end_handler, listener) = build_components();
     // Multiple threads can not just have a reference of the front end handler because they could
     // outlive the run function. It is needed to give them all ownership of the front end handler
     // through an Arc.
     let front_end_handler = Arc::from(front_end_handler);
 
+    // Register a boolean set to true when the SIGTERM signal is received.
+    let kill_signal = Arc::new(AtomicBool::new(false));
+    flag::register(SIGTERM, kill_signal.clone())?;
+
     let threadpool = Builder::new().build();
 
     loop {
-        if let Some(stream) = listener.wait_on_connection() {
+        if kill_signal.load(Ordering::Relaxed) {
+            println!("SIGTERM signal received.");
+            break;
+        }
+
+        if let Some(stream) = listener.accept() {
             let front_end_handler = front_end_handler.clone();
             threadpool.execute(move || {
                 front_end_handler.handle_request(stream);
             });
         } else {
-            println!("Error on establishing last connection, continuing...");
+            //TODO: this value should come from the configuration.
+            thread::sleep(Duration::from_millis(10));
         }
     }
+
+    println!("Shutting down PARSEC, waiting for all threads to finish.");
+    threadpool.join();
+
+    Ok(())
 }

src/bin/main.rs

@@ -13,6 +13,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 use std::fs;
+use std::io::ErrorKind;
 use std::os::unix::net::UnixListener;
 use std::path::Path;
 use std::time::Duration;
@@ -34,7 +35,7 @@ pub struct DomainSocketListener {
     timeout: Duration,
 }
 
-impl Listen for DomainSocketListener {
+impl DomainSocketListener {
     /// Initialise the connection to the Unix socket.
     ///
     /// # Panics
@@ -53,21 +54,25 @@ impl Listen for DomainSocketListener {
             Err(err) => panic!(err),
         };
 
+        // Set the socket as non-blocking.
+        listener_val
+            .set_nonblocking(true)
+            .expect("Could not set the socket as non-blocking");
+
         self.listener = Some(listener_val);
     }
+}
 
+impl Listen for DomainSocketListener {
     fn set_timeout(&mut self, duration: Duration) {
         self.timeout = duration;
     }
 
-    fn wait_on_connection(&self) -> Option<Box<dyn ReadWrite + Send>> {
+    fn accept(&self) -> Option<Box<dyn ReadWrite + Send>> {
         if let Some(listener) = &self.listener {
-            let stream_result = listener
-                .incoming()
-                .next()
-                .expect("The Incoming iterator should never return None!");
+            let stream_result = listener.accept();
             match stream_result {
-                Ok(stream) => {
+                Ok((stream, _)) => {
                     if let Err(err) = stream.set_read_timeout(Some(self.timeout)) {
                         println!("Failed to set read timeout ({})", err);
                         None
@@ -79,7 +84,11 @@ impl Listen for DomainSocketListener {
                     }
                 }
                 Err(err) => {
-                    println!("Failed to connect with a UnixStream ({})", err);
+                    // Check if the error is because no connections are currently present.
+                    if err.kind() != ErrorKind::WouldBlock {
+                        // Only log the real errors.
+                        println!("Failed to connect with a UnixStream ({})", err);
+                    }
                     None
                 }
             }

src/front/domain_socket.rs

@@ -21,22 +21,20 @@ pub trait ReadWrite: std::io::Read + std::io::Write {}
 impl<T: std::io::Read + std::io::Write> ReadWrite for T {}
 
 pub trait Listen {
-    /// Initialise the internals of the listener.
-    fn init(&mut self);
-
     /// Set the timeout on read and write calls on any stream returned by this listener.
     fn set_timeout(&mut self, duration: Duration);
 
-    /// Blocking call that waits for incoming connections and returns a stream (a Read and Write
-    /// trait object). Requests are read from the stream and responses are written to it.
-    /// Streams returned by this method should have a timeout period as set by the `set_timeout`
-    /// method.
+    /// Non-blocking call that gets the next client connection and returns a stream
+    /// (a Read and Write trait object). Requests are read from the stream and responses are written
+    /// to it. Streams returned by this method should have a timeout period as set by the
+    /// `set_timeout` method.
+    /// If no connections are present, return `None`.
     /// If there are any errors in establishing the connection other than the missing
     /// initialization, the implementation should log them and return `None`.
     /// `Send` is needed because the stream is moved to a thread.
     ///
     /// # Panics
     ///
     /// If the listener has not been initialised before, with the `init` method.
-    fn wait_on_connection(&self) -> Option<Box<dyn ReadWrite + Send>>;
+    fn accept(&self) -> Option<Box<dyn ReadWrite + Send>>;
 }