graceful api shutdown (#3677)

2022-01-07 14:24:54 +01:00
parent 2237f42144
commit c92d2e9fba
8 changed files with 134 additions and 97 deletions
@@ -38,12 +38,15 @@ use crate::router::ResponseFuture;
 use crate::router::Router;
 use crate::util::to_base64;
 use crate::util::RwLock;
+use crate::util::StopState;
 use crate::web::*;
 use easy_jsonrpc_mw::{Handler, MaybeReply};
+use futures::channel::oneshot;
 use hyper::{Body, Request, Response, StatusCode};
 use serde::Serialize;
 use std::net::SocketAddr;
 use std::sync::{Arc, Weak};
+use std::thread;

 /// Listener version, providing same API but listening for requests on a
 /// port and wrapping the calls
@@ -56,6 +59,8 @@ pub fn node_apis<B, P>(
 	api_secret: Option<String>,
 	foreign_api_secret: Option<String>,
 	tls_config: Option<TLSConfig>,
+	api_chan: &'static mut (oneshot::Sender<()>, oneshot::Receiver<()>),
+	stop_state: Arc<StopState>,
 ) -> Result<(), Error>
 where
 	B: BlockChain + 'static,
@@ -104,10 +109,24 @@ where
 	let mut apis = ApiServer::new();
 	warn!("Starting HTTP Node APIs server at {}.", addr);
 	let socket_addr: SocketAddr = addr.parse().expect("unable to parse socket address");
-	let api_thread = apis.start(socket_addr, router, tls_config);
+	let api_thread = apis.start(socket_addr, router, tls_config, api_chan);

 	warn!("HTTP Node listener started.");

+	thread::Builder::new()
+		.name("api_monitor".to_string())
+		.spawn(move || {
+			// monitor for stop state is_stopped
+			loop {
+				std::thread::sleep(std::time::Duration::from_millis(100));
+				if stop_state.is_stopped() {
+					apis.stop();
+					break;
+				}
+			}
+		})
+		.ok();
+
 	match api_thread {
 		Ok(_) => Ok(()),
 		Err(e) => {
@@ -186,10 +186,11 @@ impl ApiServer {
 		addr: SocketAddr,
 		router: Router,
 		conf: Option<TLSConfig>,
+		api_chan: &'static mut (oneshot::Sender<()>, oneshot::Receiver<()>),
 	) -> Result<thread::JoinHandle<()>, Error> {
 		match conf {
-			Some(conf) => self.start_tls(addr, router, conf),
-			None => self.start_no_tls(addr, router),
+			Some(conf) => self.start_tls(addr, router, conf, api_chan),
+			None => self.start_no_tls(addr, router, api_chan),
 		}
 	}

@@ -198,6 +199,7 @@ impl ApiServer {
 		&mut self,
 		addr: SocketAddr,
 		router: Router,
+		api_chan: &'static mut (oneshot::Sender<()>, oneshot::Receiver<()>),
 	) -> Result<thread::JoinHandle<()>, Error> {
 		if self.shutdown_sender.is_some() {
 			return Err(ErrorKind::Internal(
@@ -205,18 +207,26 @@ impl ApiServer {
 			)
 			.into());
 		}
-		let (tx, _rx) = oneshot::channel::<()>();
+		let rx = &mut api_chan.1;
+		let tx = &mut api_chan.0;
+
+		// Jones's trick to update memory
+		let m = oneshot::channel::<()>();
+		let tx = std::mem::replace(tx, m.0);
 		self.shutdown_sender = Some(tx);
+
 		thread::Builder::new()
 			.name("apis".to_string())
 			.spawn(move || {
 				let server = async move {
-					let server = Server::bind(&addr).serve(make_service_fn(move |_| {
-						let router = router.clone();
-						async move { Ok::<_, Infallible>(router) }
-					}));
-					// TODO graceful shutdown is unstable, investigate
-					//.with_graceful_shutdown(rx)
+					let server = Server::bind(&addr)
+						.serve(make_service_fn(move |_| {
+							let router = router.clone();
+							async move { Ok::<_, Infallible>(router) }
+						}))
+						.with_graceful_shutdown(async {
+							rx.await.ok();
+						});

 					server.await
 				};
@@ -238,6 +248,7 @@ impl ApiServer {
 		addr: SocketAddr,
 		router: Router,
 		conf: TLSConfig,
+		api_chan: &'static mut (oneshot::Sender<()>, oneshot::Receiver<()>),
 	) -> Result<thread::JoinHandle<()>, Error> {
 		if self.shutdown_sender.is_some() {
 			return Err(ErrorKind::Internal(
@@ -246,6 +257,14 @@ impl ApiServer {
 			.into());
 		}

+		let rx = &mut api_chan.1;
+		let tx = &mut api_chan.0;
+
+		// Jones's trick to update memory
+		let m = oneshot::channel::<()>();
+		let tx = std::mem::replace(tx, m.0);
+		self.shutdown_sender = Some(tx);
+
 		let acceptor = TlsAcceptor::from(conf.build_server_config()?);

 		thread::Builder::new()
@@ -258,12 +277,14 @@ impl ApiServer {
 						.and_then(move |s| acceptor.accept(s))
 						.filter(|r| r.is_ok());

-					let server = Server::builder(accept::from_stream(listener)).serve(
-						make_service_fn(move |_| {
+					let server = Server::builder(accept::from_stream(listener))
+						.serve(make_service_fn(move |_| {
 							let router = router.clone();
 							async move { Ok::<_, Infallible>(router) }
-						}),
-					);
+						}))
+						.with_graceful_shutdown(async {
+							rx.await.ok();
+						});

 					server.await
 				};
@@ -281,9 +302,10 @@ impl ApiServer {
 	/// Stops the API server, it panics in case of error
 	pub fn stop(&mut self) -> bool {
 		if self.shutdown_sender.is_some() {
-			// TODO re-enable stop after investigation
-			//let tx = mem::replace(&mut self.shutdown_sender, None).unwrap();
-			//tx.send(()).expect("Failed to stop API server");
+			let tx = self.shutdown_sender.as_mut().unwrap();
+			let m = oneshot::channel::<()>();
+			let tx = std::mem::replace(tx, m.0);
+			tx.send(()).expect("Failed to stop API server");
 			info!("API server has been stopped");
 			true
 		} else {
@@ -2,6 +2,7 @@ use grin_api as api;
 use grin_util as util;

 use crate::api::*;
+use futures::channel::oneshot;
 use hyper::{Body, Request, StatusCode};
 use std::net::SocketAddr;
 use std::sync::atomic::{AtomicUsize, Ordering};
@@ -71,7 +72,9 @@ fn test_start_api() {
 	router.add_middleware(counter.clone());
 	let server_addr = "127.0.0.1:14434";
 	let addr: SocketAddr = server_addr.parse().expect("unable to parse server address");
-	assert!(server.start(addr, router, None).is_ok());
+	let api_chan: &'static mut (oneshot::Sender<()>, oneshot::Receiver<()>) =
+		Box::leak(Box::new(oneshot::channel::<()>()));
+	assert!(server.start(addr, router, None, api_chan).is_ok());
 	let url = format!("http://{}/v1/", server_addr);
 	let index = request_with_retry(url.as_str()).unwrap();
 	assert_eq!(index.len(), 2);
@@ -96,7 +99,9 @@ fn test_start_api_tls() {
 	let router = build_router();
 	let server_addr = "0.0.0.0:14444";
 	let addr: SocketAddr = server_addr.parse().expect("unable to parse server address");
-	assert!(server.start(addr, router, Some(tls_conf)).is_ok());
+	let api_chan: &'static mut (oneshot::Sender<()>, oneshot::Receiver<()>) =
+		Box::leak(Box::new(oneshot::channel::<()>()));
+	assert!(server.start(addr, router, Some(tls_conf), api_chan).is_ok());
 	let index = request_with_retry("https://yourdomain.com:14444/v1/").unwrap();
 	assert_eq!(index.len(), 2);
 	assert!(!server.stop());