Augmented tokio::net::Incoming with Suspendable stream (paritytech#321)

* Augmented tokio::net::Incoming with Suspendable stream

* exponential delay increase, removed duplicate polling logic

* exempt connection errors from timeout

* reset timer

* performance improvements, added concise documentation

* reset initial_delay to 10ms

Author: seunlanlege

http/src/lib.rs

@@ -53,7 +53,7 @@ use server_utils::reactor::{Executor, UninitializedExecutor};
 
 pub use server_utils::hosts::{Host, DomainsValidation};
 pub use server_utils::cors::{self, AccessControlAllowOrigin, Origin, AllowCors};
-pub use server_utils::tokio;
+pub use server_utils::{tokio, SuspendableStream};
 pub use handler::ServerHandler;
 pub use utils::{is_host_allowed, cors_allow_origin, cors_allow_headers};
 pub use response::Response;
@@ -503,8 +503,9 @@ fn serve<M: jsonrpc::Metadata, S: jsonrpc::Middleware<M>>(
 
 			let mut http = server::conn::Http::new();
 			http.keep_alive(keep_alive);
+			let tcp_stream = SuspendableStream::new(listener.incoming());
 
-			listener.incoming()
+			tcp_stream
 				.for_each(move |socket| {
 					let service = ServerHandler::new(
 						jsonrpc_handler.clone(),

server-utils/src/lib.rs

@@ -22,7 +22,9 @@ pub mod session;
 pub mod reactor;
 mod matcher;
 mod stream_codec;
+mod suspendable_stream;
 
+pub use suspendable_stream::SuspendableStream;
 pub use matcher::Pattern;
 
 /// Codecs utilities

server-utils/src/suspendable_stream.rs

@@ -0,0 +1,88 @@
+use std::time::{Duration, Instant};
+use tokio::timer::Delay;
+use std::io;
+use tokio::prelude::*;
+
+/// `Incoming` is a stream of incoming sockets
+/// Polling the stream may return a temporary io::Error (for instance if we can't open the connection because of "too many open files" limit)
+/// we use for_each combinator which:
+/// 1. Runs for every Ok(socket)
+/// 2. Stops on the FIRST Err()
+/// So any temporary io::Error will cause the entire server to terminate.
+/// This wrapper type for tokio::Incoming stops accepting new connections
+/// for a specified amount of time once an io::Error is encountered
+pub struct SuspendableStream<S> {
+	stream: S,
+	next_delay: Duration,
+	initial_delay: Duration,
+	max_delay: Duration,
+	timeout: Option<Delay>,
+}
+
+impl<S> SuspendableStream<S> {
+	/// construct a new Suspendable stream, given tokio::Incoming
+	/// and the amount of time to pause for.
+	pub fn new(stream: S) -> Self {
+		SuspendableStream {
+			stream,
+			next_delay: Duration::from_millis(20),
+			initial_delay: Duration::from_millis(10),
+			max_delay: Duration::from_secs(5),
+			timeout: None,
+		}
+	}
+}
+
+impl<S, I> Stream for SuspendableStream<S>
+	where S: Stream<Item=I, Error=io::Error>
+{
+	type Item = I;
+	type Error = ();
+
+	fn poll(&mut self) -> Result<Async<Option<Self::Item>>, ()> {
+		if let Some(mut timeout) = self.timeout.take() {
+			match timeout.poll() {
+				Ok(Async::Ready(_)) => {}
+				Ok(Async::NotReady) => {
+					self.timeout = Some(timeout);
+					return Ok(Async::NotReady);
+				}
+				Err(_) => unreachable!("Polling a delay shouldn't yield any errors; qed")
+			}
+		}
+
+		loop {
+			match self.stream.poll() {
+				Ok(item) => {
+					if self.next_delay > self.initial_delay {
+						self.next_delay = self.initial_delay;
+					}
+					return Ok(item)
+				},
+				Err(ref e) => if connection_error(e) {
+					warn!("Connection Error: {:?}", e);
+					continue
+				}
+				Err(err) => {
+					self.next_delay = if self.next_delay < self.max_delay {
+						self.next_delay * 2
+					} else {
+						self.next_delay
+					};
+					warn!("Error accepting connection: {}", err);
+					warn!("The server will stop accepting connections for {:?}", self.next_delay);
+					self.timeout = Some(Delay::new(Instant::now() + self.next_delay));
+				}
+			}
+		}
+	}
+}
+
+
+/// assert that the error was a connection error
+fn connection_error(e: &io::Error) -> bool {
+	e.kind() == io::ErrorKind::ConnectionRefused ||
+		e.kind() == io::ErrorKind::ConnectionAborted ||
+		e.kind() == io::ErrorKind::ConnectionReset
+}
+

tcp/examples/tcp.rs

@@ -1,9 +1,11 @@
 extern crate solana_jsonrpc_tcp_server as jsonrpc_tcp_server;
+extern crate env_logger;
 
 use jsonrpc_tcp_server::ServerBuilder;
 use jsonrpc_tcp_server::jsonrpc_core::*;
 
 fn main() {
+	env_logger::init();
 	let mut io = IoHandler::default();
 	io.add_method("say_hello", |_params| {
 		println!("Processing");

tcp/src/server.rs

@@ -10,6 +10,7 @@ use jsonrpc::futures::sync::{mpsc, oneshot};
 use server_utils::{
 	tokio_codec::Framed,
 	tokio, reactor, codecs,
+	SuspendableStream
 };
 
 use dispatch::{Dispatcher, SenderChannels, PeerMessageQueue};
@@ -86,7 +87,7 @@ impl<M: Metadata, S: Middleware<M> + 'static> ServerBuilder<M, S> {
 		executor.spawn(future::lazy(move || {
 			let start = move || {
 				let listener = tokio::net::TcpListener::bind(&address)?;
-				let connections = listener.incoming();
+				let connections = SuspendableStream::new(listener.incoming());
 
 				let server = connections.for_each(move |socket| {
 					let peer_addr = socket.peer_addr().expect("Unable to determine socket peer address");
@@ -152,7 +153,7 @@ impl<M: Metadata, S: Middleware<M> + 'static> ServerBuilder<M, S> {
 				Ok(server)
 			};
 
-			let stop = stop_rx.map_err(|_| std::io::ErrorKind::Interrupted.into());
+			let stop = stop_rx.map_err(|_| ());
 			match start() {
 				Ok(server) => {
 					tx.send(Ok(())).expect("Rx is blocking parent thread.");