hyper: How do you write a resilient HTTP Hyper server that does not crash with "too many open files"?

Using tk-listen I can now mitigate the problem and the server does not crash anymore when it has only a few file descriptors with ulimit -n 50. Yay!

Here is the full source for a resilient Hyper echo server:

extern crate futures;
extern crate hyper;
extern crate service_fn;
extern crate tk_listen;
extern crate tokio_core;

use futures::Stream;
use hyper::header::{ContentLength, ContentType};
use hyper::server::{Http, Response};
use service_fn::service_fn;
use std::time::Duration;
use tk_listen::ListenExt;
use tokio_core::net::TcpListener;
use tokio_core::reactor::Core;

const TEXT: &'static str = "Hello, World!";

fn main() {
    let addr = ([127, 0, 0, 1], 3000).into();

    let mut core = Core::new().unwrap();
    let handle = core.handle();
    let handle2 = core.handle();
    let http = Http::new();
    let listener = TcpListener::bind(&addr, &handle).unwrap();

    let server = listener
        .incoming()
        .sleep_on_error(Duration::from_millis(10), &handle2)
        .map(move |(sock, addr)| {
            let hello = service_fn(|_req| {
                Ok(
                    Response::<hyper::Body>::new()
                        .with_header(ContentLength(TEXT.len() as u64))
                        .with_header(ContentType::plaintext())
                        .with_body(TEXT),
                )
            });
            http.bind_connection(
                &handle,
                sock,
                addr,
                hello,
            );
            Ok(())
        })
        // Maximum of 10,000 connections simultaneously.
        .listen(10_000);

    core.run(server).unwrap();
}

Now calling ab -c 1000 -n 100000 http://localhost:3000/ in a new shell works, but it does not finish. The last ~200 of 100k requests never finish and at some point ab exists with

Completed 10000 requests
Completed 20000 requests
Completed 30000 requests
Completed 40000 requests
Completed 50000 requests
Completed 60000 requests
Completed 70000 requests
Completed 80000 requests
Completed 90000 requests
apr_socket_recv: Connection reset by peer (104)
Total of 99798 requests completed

While ab is in progress I can successfully reach the server manually in my browser, so this might not be a big problem. And it is certainly an improvement by not crashing the server 😃

@seanmonstar: what do you think of using tk-listen as a dependency in Hyper and patching server/mod.rs to do something similar?

An alternative solution to this is using std::net::TcpListener to accept connections, as in the Tokio multi threaded server example: https://github.com/tokio-rs/tokio-core/blob/master/examples/echo-threads.rs

Advantages:

• no need for tk-listen
• server is resilient and does not crash
• slight performance increase because request handling is distributed on worker threads with their own Tokio core event loop

The downside is that you have more code in your server that you need to reason about and maintain.

The tinyhttp example in tokio-core has the same vulnerability, also filed an issue there.

Maybe this is not quite to the point, but the thread reminded me of https://crates.io/crates/tk-listen

for socket in listener.incoming() {
    let socket = match socket {
        Ok(socket) => socket,
        // Ignore socket errors like "Too many open files" on the OS
        // level. Just continue with the next request.
        Err(_) => continue,
    };
    // ...  
}

Just to clarify tailhook’s comment: this code right here will spin the accept loop hard, since the EMFILE error doesn’t remove the socket from the acceptor’s queue. You might want to sleep the thread for a few milliseconds or something there.

What about filtering out Errs?

let server = listener
    .incoming()
    .then(|x| future::ok::<_, ()>(x.ok()))
    .filter_map(|x| x)
    .for_each(|(sock, addr)| {
        println!("{:?} {:?}", sock, addr);
        Ok(())
    });

I’m trying the or_else() future as @carllerche recommended.

Starting from:

let server = listener.incoming().for_each(move |(sock, addr)| {
    http.bind_connection(
        &handle,
        sock,
        addr,
        Proxy {
            port: port,
            upstream_port: upstream_port,
            client: client.clone(),
        },
    );
    Ok(())
});

Attempt 1: just insert the or_else() and see what the compiler tells us:

let server = listener.incoming().or_else(|e| {
                println!("{:?}", e);
            }).for_each(move |(sock, addr)| { ... });

error[E0277]: the trait bound `(): futures::Future` is not satisfied
   --> src/lib.rs:169:46
    |
169 |             let server = listener.incoming().or_else(|e| {
    |                                              ^^^^^^^ the trait `futures::Future` is not implemented for `()`
    |
    = note: required because of the requirements on the impl of `futures::IntoFuture` for `()`

I was hoping the compiler would give me a hint about the return type I have to produce in my closure, but that is not helpful. I’m not using () anywhere, so what is she talking about? Looking at the docs at https://docs.rs/futures/0.1.16/futures/stream/trait.Stream.html#method.or_else there is no example and the type only says I need to return an U which is IntoFuture<Item = Self::Item>.

Attempt 2: Return an empty Ok tuple as the for_each() does

let server = listener.incoming().or_else(|e| {
                println!("{:?}", e);
                Ok(())
            }).for_each(move |(sock, addr)| { ... });

error[E0271]: type mismatch resolving `<std::result::Result<(), _> as futures::IntoFuture>::Item == (tokio_core::net::TcpStream, std::net::SocketAddr)`
   --> src/lib.rs:169:46
    |
169 |             let server = listener.incoming().or_else(|e| {
    |                                              ^^^^^^^ expected (), found tuple
    |
    = note: expected type `()`
               found type `(tokio_core::net::TcpStream, std::net::SocketAddr)`

Attempt 3: Return an Err:

let server = listener.incoming().or_else(|e| {
                println!("{:?}", e);
                Err(e)
            }).for_each(move |(sock, addr)| { ... });

At least it compiles!!!

But it does not solve the problem: returning an error here bubbles up and crashes my server as before. With the only difference of the additional print statement.

Attempt 4: Return an empty future, assuming it does nothing and Incoming continues with the next connection attempt:

let server = listener.incoming().or_else(|e| -> futures::future::Empty<_, std::io::Error> {
    println!("{:?}", e);
    futures::future::empty()
}).for_each(move |(sock, addr)| { ... });

That compiles, but as soon as the first IO error happens the server does not respond anymore. Looking at the docs: https://docs.rs/futures/0.1.16/futures/future/struct.Empty.html it says “A future which is never resolved.”. Aha, so that is probably blocking my server. So this is not really an Empty future and should be renamed to “AlwaysBlockingDoingNothing”.

Attempt 5: Let’s try the or_else() after the for_each():

let server = listener.incoming().for_each(...).or_else(|e| -> Result<_> {
    println!("{:?}", e);
    Ok(())
});

This compiles, but does not swallow the error. The server still crashes except for the additional print statement.

At this point I’m running out of ideas. How can I swallow the IO error and make the incoming future continue?

@seanmonstar that contract isn’t actually accurate. Stream returning Err is implementation specific. See: https://github.com/alexcrichton/futures-rs/issues/206.

The Incoming stream is intended to allow polling after an error is returned. None represents the final state.