multithreaded_echo_server.rs

#[macro_use]
extern crate log;
extern crate env_logger;
extern crate num_cpus;
extern crate net2;
extern crate tokio_core;
extern crate tokio_io;
extern crate futures;

use futures::Future;
use futures::stream::Stream;
use std::net::SocketAddr;
use std::thread;
use std::io;
use tokio_core::reactor::{Core, Handle};
use tokio_core::net::TcpListener;
use tokio_io::{AsyncRead, AsyncWrite};

fn main() {
    env_logger::init().unwrap();
    info!("Server started");

    let addr = "127.0.0.1:7777".parse::<SocketAddr>().unwrap();
    info!("Using address: {}", addr);

    let thread_count = num_cpus::get();
    debug!("Using {} threads", thread_count);

    for i in 1..thread_count {
        thread::Builder::new()
            .name(format!("server_thread-{}", i))
            .spawn(move || server_main(addr));
    }

    server_main(addr);
}

fn thread_name() -> String {
    thread::current().name().unwrap_or("unnamed").into()
}

fn server_main(addr: SocketAddr) {
    debug!("Thread `{}` started", thread_name());

    let mut core = Core::new().unwrap();
    let handle = core.handle();
    let listener = listener(&addr, &handle).unwrap();

    let server = listener.incoming().for_each(move |(socket, _)| {
        let (reader, writer) = socket.split();
        let msg = tokio_io::io::copy(reader, writer)
            .map(move |(amount, _, _)| {
                debug!("Thread `{}` wrote {} bytes to {}", thread_name(), amount, addr)
            })
            .map_err(|e| {
                error!("Thread `{}` error while copying: {}", thread_name(), e);
            });

        handle.spawn(msg);

        Ok(())
    });

    core.run(server).unwrap();
    debug!("Thread `{}` shutdown", thread_name());
}

fn listener(addr: &SocketAddr, handle: &Handle) -> io::Result<TcpListener> {
    let builder = match *addr {
        SocketAddr::V4(_) => net2::TcpBuilder::new_v4()?,
        SocketAddr::V6(_) => net2::TcpBuilder::new_v6()?,
    };

    builder
        .reuse_address(true)?
        .bind(addr)?
        .listen(1024)
        .and_then(|l| {
            TcpListener::from_listener(l, addr, handle)
        })
}