Support multiple hyper threads. (#103)

Author: tomusdrw

http/Cargo.toml

@@ -11,6 +11,7 @@ documentation = "https://paritytech.github.io/jsonrpc/jsonrpc_http_server/index.
 
 [dependencies]
 log = "0.3"
+net2 = "0.2"
 jsonrpc-core = { version = "7.0", path = "../core" }
 jsonrpc-server-utils = { version = "7.0", path = "../server-utils" }
 hyper = { git = "https://github.com/hyperium/hyper" }

http/examples/server.rs

@@ -11,6 +11,7 @@ fn main() {
 	});
 
 	let server = ServerBuilder::new(io)
+		.threads(3)
 		.cors(DomainsValidation::AllowOnly(vec![AccessControlAllowOrigin::Any]))
 		.start_http(&"127.0.0.1:3030".parse().unwrap())
 		.expect("Unable to start RPC server");

http/src/handler.rs

@@ -10,13 +10,9 @@ use unicase::UniCase;
 use jsonrpc::{Metadata, Middleware, NoopMiddleware};
 use jsonrpc::futures::{Future, Poll, Async, BoxFuture, Stream};
 use response::Response;
-use server_utils::{cors, hosts};
+use server_utils::cors;
 
-use {utils, RequestMiddleware, RequestMiddlewareAction};
-
-
-type AllowedHosts = Option<Vec<hosts::Host>>;
-type CorsDomains = Option<Vec<cors::AccessControlAllowOrigin>>;
+use {utils, RequestMiddleware, RequestMiddlewareAction, CorsDomains, AllowedHosts};
 
 /// jsonrpc http request handler.
 pub struct ServerHandler<M: Metadata = (), S: Middleware<M> = NoopMiddleware> {
@@ -119,6 +115,7 @@ impl<M> RpcPollState<M> {
 		}
 	}
 }
+
 enum RpcHandlerState<M> {
 	ReadingHeaders {
 		request: server::Request,

http/src/lib.rs

@@ -23,6 +23,7 @@
 extern crate unicase;
 extern crate jsonrpc_core as jsonrpc;
 extern crate jsonrpc_server_utils as server_utils;
+extern crate net2;
 
 pub extern crate hyper;
 
@@ -39,6 +40,7 @@ use std::net::SocketAddr;
 use hyper::server;
 use jsonrpc::MetaIoHandler;
 use jsonrpc::futures::{self, Future, IntoFuture, BoxFuture, Stream};
+use jsonrpc::futures::sync::oneshot;
 use server_utils::reactor::{Remote, UninitializedRemote};
 
 pub use server_utils::hosts::{Host, DomainsValidation};
@@ -198,15 +200,18 @@ impl<M: jsonrpc::Metadata, S: jsonrpc::Middleware<M>> Clone for Rpc<M, S> {
 	}
 }
 
+type AllowedHosts = Option<Vec<Host>>;
+type CorsDomains = Option<Vec<AccessControlAllowOrigin>>;
 
 /// Convenient JSON-RPC HTTP Server builder.
 pub struct ServerBuilder<M: jsonrpc::Metadata = (), S: jsonrpc::Middleware<M> = jsonrpc::NoopMiddleware> {
 	handler: Arc<MetaIoHandler<M, S>>,
 	remote: UninitializedRemote,
 	meta_extractor: Arc<MetaExtractor<M>>,
 	request_middleware: Arc<RequestMiddleware>,
-	cors_domains: Option<Vec<AccessControlAllowOrigin>>,
-	allowed_hosts: Option<Vec<Host>>,
+	cors_domains: CorsDomains,
+	allowed_hosts: AllowedHosts,
+	threads: usize,
 }
 
 const SENDER_PROOF: &'static str = "Server initialization awaits local address.";
@@ -230,15 +235,33 @@ impl<M: jsonrpc::Metadata, S: jsonrpc::Middleware<M>> ServerBuilder<M, S> {
 			request_middleware: Arc::new(NoopRequestMiddleware::default()),
 			cors_domains: None,
 			allowed_hosts: None,
+			threads: 1,
 		}
 	}
 
 	/// Utilize existing event loop remote to poll RPC results.
+	/// Applies only to 1 of the threads. Other threads will spawn their own Event Loops.
 	pub fn event_loop_remote(mut self, remote: tokio_core::reactor::Remote) -> Self {
 		self.remote = UninitializedRemote::Shared(remote);
 		self
 	}
 
+	/// Sets number of threads of the server to run.
+	/// Panics when set to `0`.
+	#[cfg(not(unix))]
+	pub fn threads(mut self, _threads: usize) -> Self {
+		warn!("Multi-threaded server is not available on Windows. Falling back to single thread.");
+		self
+	}
+
+	/// Sets number of threads of the server to run.
+	/// Panics when set to `0`.
+	#[cfg(unix)]
+	pub fn threads(mut self, threads: usize) -> Self {
+		self.threads = threads;
+		self
+	}
+
 	/// Configures a list of allowed CORS origins.
 	pub fn cors(mut self, cors_domains: DomainsValidation<AccessControlAllowOrigin>) -> Self {
 		self.cors_domains = cors_domains.into();
@@ -274,88 +297,160 @@ impl<M: jsonrpc::Metadata, S: jsonrpc::Middleware<M>> ServerBuilder<M, S> {
 		let cors_domains = self.cors_domains;
 		let request_middleware = self.request_middleware;
 		let allowed_hosts = self.allowed_hosts;
-
-		let eloop = self.remote.initialize()?;
 		let jsonrpc_handler = Rpc {
 			handler: self.handler,
 			extractor: self.meta_extractor,
 		};
+		let reuse_port = self.threads > 1;
 
 		let (local_addr_tx, local_addr_rx) = mpsc::channel();
-		let (close, shutdown_signal) = futures::sync::oneshot::channel();
-		let addr = addr.to_owned();
-		// TODO [ToDr] Consider spawning many threads like in minihttp?
-		eloop.remote().spawn(move |handle| {
-			let handle1 = handle.clone();
-			let bind = move || {
-				let listener = tokio_core::net::TcpListener::bind(&addr, &handle1)?;
-				// 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
-					local_addr_tx.send(Ok(local_addr)).expect(SENDER_PROOF);
-
-					futures::future::ok((listener, local_addr))
-				},
-				Err(err) => {
-					// Send error
-					local_addr_tx.send(Err(err)).expect(SENDER_PROOF);
-
-					futures::future::err(())
-				}
-			};
-
-			let handle = handle.clone();
-			bind_result.and_then(move |(listener, local_addr)| {
-				let allowed_hosts = server_utils::hosts::update(allowed_hosts, &local_addr);
-
-				let http = server::Http::new();
-				listener.incoming()
-					.for_each(move |(socket, addr)| {
-						http.bind_connection(&handle, socket, addr, ServerHandler::new(
-							jsonrpc_handler.clone(),
-							cors_domains.clone(),
-							allowed_hosts.clone(),
-							request_middleware.clone(),
-						));
-						Ok(())
-					})
-					.map_err(|e| {
-						warn!("Incoming streams error, closing sever: {:?}", e);
-					})
-					.select(shutdown_signal.map_err(|e| {
-						warn!("Shutdown signaller dropped, closing server: {:?}", e);
-					}))
-					.map(|_| ())
-					.map_err(|_| ())
-			})
-		});
+		let (close, shutdown_signal) = oneshot::channel();
+		let eloop = self.remote.init_with_name("http.worker0")?;
+		serve(
+			(shutdown_signal, local_addr_tx),
+			eloop.remote(),
+			addr.to_owned(),
+			cors_domains.clone(),
+			request_middleware.clone(),
+			allowed_hosts.clone(),
+			jsonrpc_handler.clone(),
+			reuse_port,
+		);
+		let handles = (0..self.threads - 1).map(|i| {
+			let (local_addr_tx, local_addr_rx) = mpsc::channel();
+			let (close, shutdown_signal) = oneshot::channel();
+			let eloop = UninitializedRemote::Unspawned.init_with_name(format!("http.worker{}", i + 1))?;
+			serve(
+				(shutdown_signal, local_addr_tx),
+				eloop.remote(),
+				addr.to_owned(),
+				cors_domains.clone(),
+				request_middleware.clone(),
+				allowed_hosts.clone(),
+				jsonrpc_handler.clone(),
+				reuse_port,
+			);
+			Ok((eloop, close, local_addr_rx))
+		}).collect::<io::Result<Vec<_>>>()?;
 
 		// Wait for server initialization
-		let local_addr: io::Result<SocketAddr> = local_addr_rx.recv().map_err(|_| {
-			Error::Io(io::Error::new(io::ErrorKind::Interrupted, ""))
-		})?;
+		let local_addr = recv_address(local_addr_rx);
+		// Wait for other threads as well.
+		let mut handles = handles.into_iter().map(|(eloop, close, local_addr_rx)| {
+			let _ = recv_address(local_addr_rx)?;
+			Ok((eloop, close))
+		}).collect::<io::Result<(Vec<_>)>>()?;
+		handles.push((eloop, close));
+		let (remotes, close) = handles.into_iter().unzip();
 
 		Ok(Server {
 			address: local_addr?,
-			remote: Some(eloop),
+			remote: Some(remotes),
 			close: Some(close),
 		})
 	}
 }
 
+fn recv_address(local_addr_rx: mpsc::Receiver<io::Result<SocketAddr>>) -> io::Result<SocketAddr> {
+	local_addr_rx.recv().map_err(|_| {
+		io::Error::new(io::ErrorKind::Interrupted, "")
+	})?
+}
+
+fn serve<M: jsonrpc::Metadata, S: jsonrpc::Middleware<M>>(
+	signals: (oneshot::Receiver<()>, mpsc::Sender<io::Result<SocketAddr>>),
+	remote: tokio_core::reactor::Remote,
+	addr: SocketAddr,
+	cors_domains: CorsDomains,
+	request_middleware: Arc<RequestMiddleware>,
+	allowed_hosts: AllowedHosts,
+	jsonrpc_handler: Rpc<M, S>,
+	reuse_port: bool,
+) {
+	let (shutdown_signal, local_addr_tx) = signals;
+	remote.spawn(move |handle| {
+		let handle1 = handle.clone();
+		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_core::net::TcpListener::from_listener(listener, &addr, &handle1)?;
+			// 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
+				local_addr_tx.send(Ok(local_addr)).expect(SENDER_PROOF);
+
+				futures::future::ok((listener, local_addr))
+			},
+			Err(err) => {
+				// Send error
+				local_addr_tx.send(Err(err)).expect(SENDER_PROOF);
+
+				futures::future::err(())
+			}
+		};
+
+		let handle = handle.clone();
+		bind_result.and_then(move |(listener, local_addr)| {
+			let allowed_hosts = server_utils::hosts::update(allowed_hosts, &local_addr);
+
+			let http = server::Http::new();
+			listener.incoming()
+				.for_each(move |(socket, addr)| {
+					http.bind_connection(&handle, socket, addr, ServerHandler::new(
+						jsonrpc_handler.clone(),
+						cors_domains.clone(),
+						allowed_hosts.clone(),
+						request_middleware.clone(),
+					));
+					Ok(())
+				})
+				.map_err(|e| {
+					warn!("Incoming streams error, closing sever: {:?}", e);
+				})
+				.select(shutdown_signal.map_err(|e| {
+					warn!("Shutdown signaller dropped, closing server: {:?}", e);
+				}))
+				.map(|_| ())
+				.map_err(|_| ())
+		})
+	});
+}
+
+#[cfg(unix)]
+fn configure_port(reuse: bool, tcp: &net2::TcpBuilder) -> io::Result<()> {
+    use net2::unix::*;
+
+    if reuse {
+        try!(tcp.reuse_port(true));
+    }
+
+    Ok(())
+}
+
+#[cfg(not(unix))]
+fn configure_port(_reuse: bool, _tcp: &net2::TcpBuilder) -> io::Result<()> {
+    Ok(())
+}
+
 /// jsonrpc http server instance
 pub struct Server {
 	address: SocketAddr,
-	remote: Option<Remote>,
-	close: Option<futures::sync::oneshot::Sender<()>>,
+	remote: Option<Vec<Remote>>,
+	close: Option<Vec<oneshot::Sender<()>>>,
 }
 
 const PROOF: &'static str = "Server is always Some until self is consumed.";
@@ -367,19 +462,28 @@ impl Server {
 
 	/// Closes the server.
 	pub fn close(mut self) {
-		let _ = self.close.take().expect(PROOF).send(());
-		self.remote.take().expect(PROOF).close();
+		for close in self.close.take().expect(PROOF) {
+			let _ = close.send(());
+		}
+
+		for remote in self.remote.take().expect(PROOF) {
+			remote.close();
+		}
 	}
 
 	/// Will block, waiting for the server to finish.
 	pub fn wait(mut self) {
-		self.remote.take().expect(PROOF).wait();
+		for remote in self.remote.take().expect(PROOF) {
+			remote.wait();
+		}
 	}
 }
 
 impl Drop for Server {
 	fn drop(&mut self) {
-		self.remote.take().map(|remote| remote.close());
+		self.remote.take().map(|remotes| {
+			for remote in remotes { remote.close(); }
+		});
 	}
 }
 

minihttp/src/lib.rs

@@ -133,7 +133,7 @@ impl<M: jsonrpc::Metadata, S: jsonrpc::Middleware<M>> ServerBuilder<M, S> {
 		}
 	}
 
-	/// Sets number of threads of the server to run.
+	/// Sets number of threads of the server to run. (not available for windows)
 	/// Panics when set to `0`.
 	pub fn threads(mut self, threads: usize) -> Self {
 		assert!(threads > 0);