CODEWITHSUNDEEP
multithreadingrustrayon
Vladimir
Vladimir

Reputation: 292

Rust rayon tcp blocking

I'm writing a program which executes a command on server using ssh and gets the output.

The part I don't understand is lower in the code.

If the function waits and then returns a string, it works as expected, but if work with TCP it starts performing very bad. I expect that using 100 threads on 100 hosts will perform 100 times faster, because will open 100 sockets simultaneously.

In the sleep version, changing poolThreads directly affects the time of execution. In the version with TCP streams, changing pool from 1 to 100 with 100 hosts only speeds it up from 90 to 67 because some hosts are offline.

I read the documentation, but cannot find anything to help me.

use clap::{App, Arg};
use rayon::prelude::*;
use rayon::ThreadPoolBuilder;
use ssh2::Session;
use std::fmt::Display;
use std::io::prelude::*;
use std::io::{BufReader, Read};
use std::net::{TcpStream, ToSocketAddrs};
use std::thread::sleep;
use std::time::Duration;

fn process_host<A>(hostname: A) -> Result<String, String>
where
    A: ToSocketAddrs + Display,
{
    sleep(Duration::from_secs(1));
    return Ok(hostname.to_string());
    // here is the problem
    // -----------------------------------------------------------
    let tcp = match TcpStream::connect(&hostname) {
        Ok(a) => a,
        Err(e) => {
            return Err(format!("{}:{}", hostname, e).to_string());
        }
    };
    let mut sess = match Session::new() {
        Ok(a) => a,
        Err(e) => {
            // todo logging
            return Err(format!("{}:{}", hostname, e).to_string());
        }
    };
    sess.set_tcp_stream(tcp);
    match sess.handshake() {
        Ok(a) => a,
        Err(e) => {
            return Err(format!("{}:{}", hostname, e).to_string());
        }
    };
    Ok(format!("{}", hostname))
}


fn main() {

    let hosts = vec!["aaaaa:22", "bbbbbbb:22"];
    let pool = ThreadPoolBuilder::new()
        .num_threads(10)
        .build()
        .expect("failed creating pool");
    pool.install(|| {
        hosts
            .par_iter()
            .map(|x| process_host(x))
            .for_each(|x| println!("{:?}", x))
    });
}

Upvotes: 0

Views: 325

Answers (1)

Vladimir
Vladimir

Reputation: 292

To debug a such a problem, you need to analysis, where your program wastes its time. There're to ways: profiling and analyzing of TCP connections.
I suggest you 2 way, cause it's much easier. Dump traffic with wireshark, filter it by port 22. After this, use conversations tab. Here you can sort connections by time, and can see, that program doesn't speedup because of no time limit for ssh connection.

Upvotes: 0

Related Questions