Fix threading and closing of servers

http/src/lib.rs

@@ -43,7 +43,7 @@ use std::thread;
 use parking_lot::Mutex;
 
 use crate::jsonrpc::futures::sync::oneshot;
-use crate::jsonrpc::futures::{self, future, Future, Stream};
+use crate::jsonrpc::futures::{self, Future, Stream};
 use crate::jsonrpc::MetaIoHandler;
 use crate::server_utils::reactor::{Executor, UninitializedExecutor};
 use hyper::{server, Body};
@@ -481,89 +481,85 @@ fn serve<M: jsonrpc::Metadata, S: jsonrpc::Middleware<M>>(
 	max_request_body_size: usize,
 ) {
 	let (shutdown_signal, local_addr_tx, done_tx) = signals;
-	executor.spawn(
-		future::lazy(move || {
-			let handle = tokio::reactor::Handle::default();
-
-			let bind = move || {
-				let listener = match addr {
-					SocketAddr::V4(_) => net2::TcpBuilder::new_v4()?,
-					SocketAddr::V6(_) => net2::TcpBuilder::new_v6()?,
-				};
-				configure_port(reuse_port, &listener)?;
-				listener.reuse_address(true)?;
-				listener.bind(&addr)?;
-				let listener = listener.listen(1024)?;
-				let listener = tokio::net::TcpListener::from_std(listener, &handle)?;
-				// Add current host to allowed headers.
-				// NOTE: we need to use `l.local_addr()` instead of `addr`
-				// it might be different!
-				let local_addr = listener.local_addr()?;
-
-				Ok((listener, local_addr))
+	executor.spawn({
+		let handle = tokio::reactor::Handle::default();
+
+		let bind = move || {
+			let listener = match addr {
+				SocketAddr::V4(_) => net2::TcpBuilder::new_v4()?,
+				SocketAddr::V6(_) => net2::TcpBuilder::new_v6()?,
 			};
+			configure_port(reuse_port, &listener)?;
+			listener.reuse_address(true)?;
+			listener.bind(&addr)?;
+			let listener = listener.listen(1024)?;
+			let listener = tokio::net::TcpListener::from_std(listener, &handle)?;
+			// Add current host to allowed headers.
+			// NOTE: we need to use `l.local_addr()` instead of `addr`
+			// it might be different!
+			let local_addr = listener.local_addr()?;
+
+			Ok((listener, local_addr))
+		};
 
-			let bind_result = match bind() {
-				Ok((listener, local_addr)) => {
-					// Send local address
-					match local_addr_tx.send(Ok(local_addr)) {
-						Ok(_) => futures::future::ok((listener, local_addr)),
-						Err(_) => {
-							warn!(
-								"Thread {:?} unable to reach receiver, closing server",
-								thread::current().name()
-							);
-							futures::future::err(())
-						}
+		let bind_result = match bind() {
+			Ok((listener, local_addr)) => {
+				// Send local address
+				match local_addr_tx.send(Ok(local_addr)) {
+					Ok(_) => futures::future::ok((listener, local_addr)),
+					Err(_) => {
+						warn!(
+							"Thread {:?} unable to reach receiver, closing server",
+							thread::current().name()
+						);
+						futures::future::err(())
 					}
 				}
-				Err(err) => {
-					// Send error
-					let _send_result = local_addr_tx.send(Err(err));
+			}
+			Err(err) => {
+				// Send error
+				let _send_result = local_addr_tx.send(Err(err));
 
-					futures::future::err(())
-				}
-			};
+				futures::future::err(())
+			}
+		};
 
-			bind_result.and_then(move |(listener, local_addr)| {
-				let allowed_hosts = server_utils::hosts::update(allowed_hosts, &local_addr);
-
-				let mut http = server::conn::Http::new();
-				http.keep_alive(keep_alive);
-				let tcp_stream = SuspendableStream::new(listener.incoming());
-
-				tcp_stream
-					.for_each(move |socket| {
-						let service = ServerHandler::new(
-							jsonrpc_handler.clone(),
-							cors_domains.clone(),
-							cors_max_age,
-							allowed_headers.clone(),
-							allowed_hosts.clone(),
-							request_middleware.clone(),
-							rest_api,
-							health_api.clone(),
-							max_request_body_size,
-							keep_alive,
-						);
-						tokio::spawn(
-							http.serve_connection(socket, service)
-								.map_err(|e| error!("Error serving connection: {:?}", e)),
-						);
-						Ok(())
-					})
-					.map_err(|e| {
-						warn!("Incoming streams error, closing sever: {:?}", e);
-					})
-					.select(shutdown_signal.map_err(|e| {
-						debug!("Shutdown signaller dropped, closing server: {:?}", e);
-					}))
-					.map(|_| ())
-					.map_err(|_| ())
-			})
+		bind_result.and_then(move |(listener, local_addr)| {
+			let allowed_hosts = server_utils::hosts::update(allowed_hosts, &local_addr);
+
+			let mut http = server::conn::Http::new();
+			http.keep_alive(keep_alive);
+			let tcp_stream = SuspendableStream::new(listener.incoming());
+
+			tcp_stream
+				.for_each(move |socket| {
+					let service = ServerHandler::new(
+						jsonrpc_handler.clone(),
+						cors_domains.clone(),
+						cors_max_age,
+						allowed_headers.clone(),
+						allowed_hosts.clone(),
+						request_middleware.clone(),
+						rest_api,
+						health_api.clone(),
+						max_request_body_size,
+						keep_alive,
+					);
+
+					http.serve_connection(socket, service)
+						.map_err(|e| error!("Error serving connection: {:?}", e))
+				})
+				.map_err(|e| {
+					warn!("Incoming streams error, closing sever: {:?}", e);
+				})
+				.select(shutdown_signal.map_err(|e| {
+					debug!("Shutdown signaller dropped, closing server: {:?}", e);
+				}))
+				.map(|_| ())
+				.map_err(|_| ())
 		})
-		.and_then(|_| done_tx.send(())),
-	);
+		.and_then(|_| done_tx.send(()))
+	});
 }
 
 #[cfg(unix)]

ipc/src/server.rs

@@ -7,7 +7,7 @@ use tokio_service::{self, Service as TokioService};
 
 use crate::server_utils::{
 	codecs, reactor, session,
-	tokio::{self, reactor::Handle, runtime::TaskExecutor},
+	tokio::{reactor::Handle, runtime::TaskExecutor},
 	tokio_codec::Framed,
 };
 use parking_lot::Mutex;
@@ -228,9 +228,7 @@ impl<M: Metadata, S: Middleware<M>> ServerBuilder<M, S> {
 					Ok(())
 				});
 
-				tokio::spawn(writer);
-
-				Ok(())
+				writer
 			});
 			start_signal
 				.send(Ok(()))

server-utils/Cargo.toml

@@ -16,7 +16,6 @@ globset = "0.4"
 jsonrpc-core = { version = "13.2", path = "../core" }
 lazy_static = "1.1.0"
 log = "0.4"
-num_cpus = "1.8"
 tokio = { version = "0.1" }
 tokio-codec = { version = "0.1" }
 unicase = "2.0"

server-utils/src/reactor.rs

@@ -1,9 +1,7 @@
 //! Event Loop Executor
 //! Either spawns a new event loop, or re-uses provided one.
 
-use num_cpus;
-use std::sync::mpsc;
-use std::{io, thread};
+use std::io;
 use tokio;
 
 use crate::core::futures::{self, Future};
@@ -82,7 +80,7 @@ impl Executor {
 pub struct RpcEventLoop {
 	executor: tokio::runtime::TaskExecutor,
 	close: Option<futures::Complete<()>>,
-	handle: Option<thread::JoinHandle<()>>,
+	handle: Option<tokio::runtime::Shutdown>,
 }
 
 impl Drop for RpcEventLoop {
@@ -100,42 +98,21 @@ impl RpcEventLoop {
 	/// Spawns a new named thread with the `EventLoop`.
 	pub fn with_name(name: Option<String>) -> io::Result<Self> {
 		let (stop, stopped) = futures::oneshot();
-		let (tx, rx) = mpsc::channel();
-		let mut tb = thread::Builder::new();
+
+		let mut tb = tokio::runtime::Builder::new();
+		tb.core_threads(1);
+
 		if let Some(name) = name {
-			tb = tb.name(name);
+			tb.name_prefix(name);
 		}
 
-		let handle = tb
-			.spawn(move || {
-				let core_threads = match num_cpus::get_physical() {
-					1 => 1,
-					2..=4 => 2,
-					_ => 3,
-				};
-
-				let runtime = tokio::runtime::Builder::new()
-					.core_threads(core_threads)
-					.name_prefix("jsonrpc-eventloop-")
-					.build();
-
-				match runtime {
-					Ok(mut runtime) => {
-						tx.send(Ok(runtime.executor())).expect("Rx is blocking upper thread.");
-						let terminate = futures::empty().select(stopped).map(|_| ()).map_err(|_| ());
-						runtime.spawn(terminate);
-						runtime.shutdown_on_idle().wait().unwrap();
-					}
-					Err(err) => {
-						tx.send(Err(err)).expect("Rx is blocking upper thread.");
-					}
-				}
-			})
-			.expect("Couldn't spawn a thread.");
-
-		let exec = rx.recv().expect("tx is transfered to a newly spawned thread.");
-
-		exec.map(|executor| RpcEventLoop {
+		let mut runtime = tb.build()?;
+		let executor = runtime.executor();
+		let terminate = futures::empty().select(stopped).map(|_| ()).map_err(|_| ());
+		runtime.spawn(terminate);
+		let handle = runtime.shutdown_on_idle();
+
+		Ok(RpcEventLoop {
 			executor,
 			close: Some(stop),
 			handle: Some(handle),
@@ -148,11 +125,11 @@ impl RpcEventLoop {
 	}
 
 	/// Blocks current thread and waits until the event loop is finished.
-	pub fn wait(mut self) -> thread::Result<()> {
+	pub fn wait(mut self) -> Result<(), ()> {
 		self.handle
 			.take()
-			.expect("Handle is always set before self is consumed.")
-			.join()
+			.ok_or(())?
+			.wait()
 	}
 
 	/// Finishes this event loop.

tcp/src/server.rs

@@ -137,9 +137,7 @@ impl<M: Metadata, S: Middleware<M> + 'static> ServerBuilder<M, S> {
 						Ok(())
 					});
 
-					tokio::spawn(writer);
-
-					Ok(())
+					writer
 				});
 
 				Ok(server)