TCP vs HTTP server

Comparison

http local
In 60.0000814 seconds we received 665953 responses and 0 errors
Mean 11099.201608749818 requests/sec

http remote
In 60.0109771 seconds we received 1224 responses and 0 errors
Mean 20.396268468689872 requests/sec

tcp nodelay local 
In 60.0000037 seconds we received 3152850 responses and 0 errors
Mean 52547.49675957103 requests/sec

tcp nodelay remote
In 60.0088259 seconds we received 744 responses and 0 errors
Mean 12.398176248937409 requests/sec

tcp standard local
In 60.0000049 seconds we received 1390007 responses and 0 errors
Mean 23166.781441379517 requests/sec

tcp standard remote
In 60.0790057 seconds we received 531 responses and 0 errors
Mean 8.838361983743681 requests/sec

http_client.rs

/**
cargo.toml:
[dependencies]
clap = "2.33.*"
reqwest =  {version = "0.10.0-alpha.2", features = ['blocking']}
*/

extern crate reqwest;
use std::net::{IpAddr, SocketAddr};
use std::time::{Duration, Instant};
use reqwest::StatusCode;

fn run_and_count(remote: SocketAddr, duration: Duration) -> Result<(), ()> {
    let start_time = Instant::now();
    let mut responses: usize = 0;
    let mut errors: usize = 0;
    let data = b"Hello, Rust!";
    let uri = format!("http://{}:{}", remote.ip(), remote.port());
    let client = reqwest::blocking::Client::new();
    while start_time.elapsed() < duration {
        let response = client.post(&uri)
            .body(data.to_vec())
            .send();

        match response {
            Ok(response)=>{
                if response.status() == StatusCode::OK  {
                    match response.text() {
                        Ok(r)=>{
                            if r.as_bytes()==data{
                                responses+=1;
                            }
                            else{
                                errors+=1;
                            }
                        }
                        Err(_)=>{
                            errors += 1;
                        }
                    }
                }
                else{
                    errors += 1;
                }
            },
            Err(e)=>{
                eprintln!("Error! {:?}", e);
                errors += 1
            }
        }
    }

    let total_duration = start_time.elapsed();

    println!(
        "In {} seconds we received {} responses and {} errors",
        total_duration.as_secs_f64(),
        responses,
        errors
    );
    println!(
        "Mean {} requests/sec",
        responses as f64 / total_duration.as_secs_f64()
    );

    Ok(())
}

fn main() {
    let config = get_config();
    run_and_count(config.remote, config.duration).unwrap();
}

struct Config {
    remote: SocketAddr,
    duration: Duration,
}

fn get_config() -> Config {
    use clap::{App, Arg};
    let app = App::new("HTTP connections testing")
        .version("0.1.0")
        .arg(
            Arg::with_name("remote")
                .short("R")
                .long("remote")
                .value_name("remote")
                .help("Remote ip addr")
                .required(true)
                .takes_value(true),
        )
        .arg(
            Arg::with_name("duration")
                .short("T")
                .long("time")
                .value_name("duration")
                .help("How long to test in seconds")
                .required(true)
                .takes_value(true),
        )
        .arg(
            Arg::with_name("port")
                .short("P")
                .long("port")
                .value_name("port")
                .help("Remote port")
                .required(true)
                .takes_value(true),
        );
    let matches = app.get_matches();
    let port: u16 = matches
        .value_of("port")
        .unwrap()
        .parse()
        .expect("Port must be u16!");
    let duration: Duration = Duration::new(
        matches
            .value_of("duration")
            .unwrap()
            .parse::<u64>()
            .expect("Duration must be num"),
        0,
    );
    let ip_addr = matches
        .value_of("remote")
        .unwrap()
        .parse::<IpAddr>()
        .expect("Invalid port");
    Config {
        remote: SocketAddr::new(ip_addr, port),
        duration,
    }
}

http_server.rs

/**
cargo.toml:
[dependencies]
clap = "2.33.*"
tokio = "0.2"
hyper = "0.12"
*/

extern crate hyper;

use hyper::rt::{Future, Stream};
use hyper::service::service_fn;
use hyper::{Body, Request, Response, Server};

fn main() {
    let config = get_config();
    let addr = ([0u8; 4], config.port).into();

    let new_svc = || {
        service_fn(|req: Request<Body>| {
            let response = req.into_body().concat2().map(|chunk| {
                let buffer: Vec<u8> = chunk.to_owned();
                match std::str::from_utf8(&buffer){
                    Ok(_) => Response::new(Body::from(buffer)),
                    Err(_) => Response::new(Body::from("Invalid data")),
                }

            });
            Box::new(response)
        })
    };

    let server = Server::bind(&addr)
        .serve(new_svc)
        .map_err(|e| eprintln!("server error: {}", e));

    hyper::rt::run(server);
}

struct Config {
    port: u16,
}

fn get_config() -> Config {
    use clap::{App, Arg};
    let app = App::new("HTTP echo server").version("0.1.0").arg(
        Arg::with_name("port")
            .short("P")
            .long("port")
            .value_name("port")
            .help("Remote port")
            .required(true)
            .takes_value(true),
    );
    let matches = app.get_matches();
    let port: u16 = matches
        .value_of("port")
        .unwrap()
        .parse()
        .expect("Port must be u16!");
    Config { port }
}

tcp_client.rs

/**
cargo.toml:
[dependencies]
clap = "2.33.*"
*/

use std::io::{Read, Write};
use std::net::{IpAddr, SocketAddr, TcpStream};
use std::time::{Duration, Instant};

fn run_and_count(remote: SocketAddr, duration: Duration) -> Result<(), String> {
    let start_time = Instant::now();
    let mut responses: usize = 0;
    let mut errors: usize = 0;
    let data = b"Hello, Rust!";
    let combined: Vec<u8> = (data.len() as u64)
        .to_le_bytes()
        .iter()
        .chain(data.iter())
        .cloned()
        .collect();
    let mut response_buffer = vec![0; combined.len()];
    let mut conn = TcpStream::connect(remote).map_err(|e| {
        eprintln!("Connection error {:?}", e);
        "Failed to connect".to_string()
    })?;
    conn.set_nodelay(true).map_err(|e|{
        eprintln!("Connection error {:?}", e);
        "Failed to nodelay".to_string()
    })?;
    while start_time.elapsed() < duration {
        match conn.write_all(&combined) {
            Ok(_) => {}
            Err(_) => {
                errors += 1;
                println!("Increase on write!");
                conn = TcpStream::connect(remote).map_err(|e| {
                    eprintln!("Connection error {:?}", e);
                    "Failed to connect".to_string()
                })?;
                conn.set_nodelay(true).map_err(|e|{
                    eprintln!("Connection error {:?}", e);
                    "Failed to nodelay".to_string()
                })?;
                continue;
            }
        };
        match conn.read_exact(&mut response_buffer) {
            Ok(_) => {}
            Err(e) => {
                errors += 1;
                println!("Increase on read! {:?}", e);
                conn = TcpStream::connect(remote).map_err(|e| {
                    eprintln!("Connection error {:?}", e);
                    "Failed to connect".to_string()
                })?;
                conn.set_nodelay(true).map_err(|e|{
                    eprintln!("Connection error {:?}", e);
                    "Failed to nodelay".to_string()
                })?;
                continue;
            }
        }
        if combined == response_buffer {
            responses += 1;
        } else {
            errors += 1;
            println!("Increase on check!");
            conn = TcpStream::connect(remote).map_err(|e| {
                eprintln!("Connection error {:?}", e);
                "Failed to connect".to_string()
            })?;
            conn.set_nodelay(true).map_err(|e|{
                eprintln!("Connection error {:?}", e);
                "Failed to nodelay".to_string()
            })?;
        }
    }

    let total_duration = start_time.elapsed();

    println!(
        "In {} seconds we received {} responses and {} errors",
        total_duration.as_secs_f64(),
        responses,
        errors
    );
    println!(
        "Mean {} requests/sec",
        responses as f64 / total_duration.as_secs_f64()
    );

    Ok(())
}

fn main() {
    let config = get_config();
    run_and_count(config.remote, config.duration).unwrap();
}

struct Config {
    remote: SocketAddr,
    duration: Duration,
}

fn get_config() -> Config {
    use clap::{App, Arg};
    let app = App::new("Tcp connections testing")
        .version("0.1.0")
        .arg(
            Arg::with_name("remote")
                .short("R")
                .long("remote")
                .value_name("remote")
                .help("Remote ip addr")
                .required(true)
                .takes_value(true),
        )
        .arg(
            Arg::with_name("duration")
                .short("T")
                .long("time")
                .value_name("duration")
                .help("How long to test in seconds")
                .required(true)
                .takes_value(true),
        )
        .arg(
            Arg::with_name("port")
                .short("P")
                .long("port")
                .value_name("port")
                .help("Remote port")
                .required(true)
                .takes_value(true),
        );
    let matches = app.get_matches();
    let port: u16 = matches
        .value_of("port")
        .unwrap()
        .parse()
        .expect("Port must be u16!");
    let duration: Duration = Duration::new(
        matches
            .value_of("duration")
            .unwrap()
            .parse::<u64>()
            .expect("Duration must be num"),
        0,
    );
    let ip_addr = matches
        .value_of("remote")
        .unwrap()
        .parse::<IpAddr>()
        .expect("Invalid port");
    Config {
        remote: SocketAddr::new(ip_addr, port),
        duration,
    }
}

tcp_server.rs

/**
cargo.toml:
[dependencies]
tokio = { version = "0.2", features = ["full"] }
clap = "2.33.*"
*/

use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use tokio;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::runtime::Runtime;

async fn echo_and_log(mut socket: TcpStream) -> Result<(), ()> {
    socket.set_nodelay(true);
    let mut buff: Vec<u8> = Vec::new();
    let mut temp_buff = [0u8; 1024];
    loop {
        buff.clear();
        let mut len_buff = [0u8; std::mem::size_of::<u64>()];
        {
            let mut current_read = 0;
            while current_read < len_buff.len() {
                let received = socket
                    .read(&mut len_buff[current_read..])
                    .await
                    .map_err(|e| {
                        eprintln!("Failed to get next len: {:?}", e);
                    })?;
                if received == 0 {
                    return Ok(());
                }
                current_read += received;
            }
        }
        let will_read = u64::from_le_bytes(len_buff) as usize;
        while buff.len() < will_read {
            let to_read = std::cmp::min(temp_buff.len(), will_read - buff.len());
            let received = socket.read(&mut temp_buff[..to_read]).await.map_err(|e| {
                eprintln!("Failed to get next len: {:?}", e);
            })?;
            buff.extend_from_slice(&temp_buff[..received]);
        }
        match std::str::from_utf8(&buff) {
            Ok(_) => {
                let mut answer = Vec::with_capacity(buff.len() + std::mem::size_of::<u64>());
                let len_buff = (buff.len() as u64).to_le_bytes();
                answer.extend_from_slice(&len_buff);
                answer.extend_from_slice(&buff);
                socket.write_all(&answer).await.map_err(|e| {
                    eprintln!("Failed to echo {:?}", e);
                })?;
            }
            Err(_) => {
                eprintln!("Invalid utf-8 packet received!");
                let answer = b"InvalidData!";
                let len_buff = (answer.len() as u64).to_le_bytes();
                socket.write_all(&len_buff).await.map_err(|e| {
                    eprintln!("Failed to answer {:?}", e);
                })?;
                socket.write_all(answer).await.map_err(|e| {
                    eprintln!("Failed to answer {:?}", e);
                })?;
                return Err(());
            }
        };

        buff.clear();
    }
}

async fn run_server(addr: SocketAddr) -> Result<(), std::io::Error> {
    let mut listener = TcpListener::bind(addr).await.unwrap();
    println!("Starting to listen");
    loop {
        let (socket, _receiver_addr) = listener.accept().await?;
        tokio::spawn(echo_and_log(socket));
    }
}

fn main() {
    let config = get_config();
    let mut runtime: Runtime = tokio::runtime::Builder::new()
        .threaded_scheduler()
        .num_threads(10)
        .thread_name("tokio-worker-")
        .enable_all()
        .build()
        .unwrap();
    let server = run_server(SocketAddr::V4(SocketAddrV4::new(
        Ipv4Addr::from([0u8; 4]),
        config.port,
    )));
    runtime.block_on(server).expect("Error during run!");
}

struct Config {
    port: u16
}

fn get_config() -> Config {
    use clap::{App, Arg};
    let app = App::new("Tcp echo server")
        .version("0.1.0")
        .arg(
            Arg::with_name("port")
                .short("P")
                .long("port")
                .value_name("port")
                .help("Remote port")
                .required(true)
                .takes_value(true),
        );
    let matches = app.get_matches();
    let port: u16 = matches
        .value_of("port")
        .unwrap()
        .parse()
        .expect("Port must be u16!");
    Config {
        port
    }
}