Slow parallel client performance

[weberc2]

Hi, I'm new to Rust, so apologies in advance if these observations boil down to an error on my part. I recently ported a small application from Go to Rust, and the Go version is outperforming the Rust version by a factor of 12 (2.5s vs 30s).

In the initial Rust version, I'm wrapping the client in an Arc pointer and cloning it once for each thread; however, if I simply create a new client per thread, I see a 4X speedup (7s vs 30s). This is surprising since the Hyper docs advertise this method.

Even with the 4X speedup, the Go version is still several times faster than the Rust version. I figured the problem was with my implementation, but some folks at the Rust forum suggested I file a bug against Hyper.

Timings were observed on my MacBook Pro with OS 10.10.5. Rust version 1.8.0 (with --release) and Go version 1.6.1.

Here's the initial Rust version:

extern crate hyper;
extern crate rustc_serialize;

use std::thread;
use std::io::Read;
use std::ops::DerefMut;
use std::sync::Mutex;
use std::sync::MutexGuard;
use std::sync::LockResult;
use std::sync::Arc;

use hyper::Client;
use hyper::header::Connection;
use rustc_serialize::json;

#[derive(RustcDecodable, RustcEncodable)]
struct Story {
    by: String,
    id: i32,
    score: i32,
    time: i32,
    title: String,
}

fn next(cursor: &mut Arc<Mutex<usize>>) -> usize {
    let result: LockResult<MutexGuard<usize>> = cursor.lock();
    let mut guard: MutexGuard<usize> = result.unwrap();
    let mut temp = guard.deref_mut();
    *temp = *temp+1;
    return *temp;
}

fn main() {
    let client = Arc::new(Client::new());
    let url = "https://hacker-news.firebaseio.com/v0/topstories.json";
    let mut res = client.get(url).header(Connection::close()).send().unwrap();
    let mut body = String::new();
    res.read_to_string(&mut body).unwrap();
    let vec: Vec<i32> = json::decode(body.as_str()).unwrap();
    let lock: Arc<Mutex<usize>> = Arc::new(Mutex::new(0));

    let mut handles: Vec<thread::JoinHandle<()>> = Vec::new();
    for _ in 0..8 {
        let client2 = client.clone();
        let mut lock2 = lock.clone();
        let vec2 = vec.clone();
        handles.push(thread::spawn(move|| {
            loop {
                let cursor = next(&mut lock2);

                if cursor >= vec2.len() {
                    break;
                }

                let url = format!(
                    "https://hacker-news.firebaseio.com/v0/item/{}.json",
                    vec2[cursor],
                );

                let mut res = client2.get(url.as_str())
                    .header(Connection::close())
                    .send()
                    .unwrap();

                let mut body = String::new();
                res.read_to_string(&mut body).unwrap();
                let story: Story = json::decode(body.as_str()).unwrap();
                println!("{}", story.title);
            };
        }));
    }

    for handle in handles.into_iter() {
        handle.join().unwrap();
    }
}

And the Go version:

package main

import (
    "encoding/json"
    "fmt"
    "io/ioutil"
    "net/http"
    "sync"
)

type Story struct {
    Title string
}

func main() {
    rsp, err := http.Get("https://hacker-news.firebaseio.com/v0/topstories.json")
    if err != nil {
        panic(err)
    }
    defer rsp.Body.Close()

    data, err := ioutil.ReadAll(rsp.Body)
    if err != nil {
        panic(err)
    }

    var ids []int
    if err := json.Unmarshal(data, &ids); err != nil {
        panic(err)
    }

    var cursor int
    var mutex sync.Mutex

    next := func() int {
        mutex.Lock()
        defer mutex.Unlock()
        temp := cursor
        cursor++
        return temp
    }

    wg := sync.WaitGroup{}
    for i := 0; i < 8; i++ {
        wg.Add(1)
        go func() {
            for cursor := next(); cursor < len(ids); cursor = next() {
                url := fmt.Sprintf(
                    "https://hacker-news.firebaseio.com/v0/item/%d.json",
                    ids[cursor],
                )
                rsp, err := http.Get(url)
                if err != nil {
                    panic(err)
                }
                defer rsp.Body.Close()

                data, err := ioutil.ReadAll(rsp.Body)
                if err != nil {
                    panic(err)
                }
                var story Story
                if err := json.Unmarshal(data, &story); err != nil {
                    panic(err)
                }
                fmt.Println(story.Title)
            }
            wg.Done()
        }()
    }

    wg.Wait()
}

[teburd]

It always must be asked, you were comparing a release build right? cargo build --release

[weberc2]

@bfrog Yes, sorry for omitting that detail.

[teburd]

Well one thing seems pretty different almost right away, you are explicitly enforcing a new connection on each request in the rust version, in the go version your leaving the decision whether to keep-alive the connection and reuse it up to the implementation. Removing the .header(Connection::close()) changes the program running time on my machine from 10's of seconds to 2.7s.

#[macro_use] extern crate log;
extern crate env_logger;
extern crate hyper;
extern crate rustc_serialize;

use std::thread;
use std::io::Read;
use std::ops::DerefMut;
use std::sync::Mutex;
use std::sync::MutexGuard;
use std::sync::LockResult;
use std::sync::Arc;

use hyper::Client;
use rustc_serialize::json;

#[derive(RustcDecodable, RustcEncodable)]
struct Story {
    by: String,
    id: i32,
    score: i32,
    time: i32,
    title: String,
}

fn next(cursor: &mut Arc<Mutex<usize>>) -> usize {
    let result: LockResult<MutexGuard<usize>> = cursor.lock();
    let mut guard: MutexGuard<usize> = result.unwrap();
    let mut temp = guard.deref_mut();
    *temp = *temp+1;
    return *temp;
}

fn main() {
    env_logger::init().unwrap();
    let client = Arc::new(Client::new());
    let url = "https://hacker-news.firebaseio.com/v0/topstories.json";
    let mut res = client.get(url).send().unwrap();
    let mut body = String::new();
    res.read_to_string(&mut body).unwrap();
    let vec: Vec<i32> = json::decode(body.as_str()).unwrap();
    let lock: Arc<Mutex<usize>> = Arc::new(Mutex::new(0));

    let mut handles: Vec<thread::JoinHandle<()>> = Vec::new();
    for _ in 0..8 {
        let client2 = client.clone();
        let mut lock2 = lock.clone();
        let vec2 = vec.clone();
        handles.push(thread::spawn(move|| {
            loop {
                let cursor = next(&mut lock2);

                if cursor >= vec2.len() {
                    break;
                }

                let url = format!(
                    "https://hacker-news.firebaseio.com/v0/item/{}.json",
                    vec2[cursor],
                    );

                let mut res = client2.get(url.as_str())
                    .send()
                    .unwrap();

                let mut body = String::new();
                res.read_to_string(&mut body).unwrap();
                let story: Story = json::decode(body.as_str()).unwrap();
                println!("{}", story.title);
            };
        }));
    }

    for handle in handles.into_iter() {
        handle.join().unwrap();
    }
}

[seanmonstar]

I should really change the example of setting a different header. I chose Connection because I thought it was clear what that header does. But I see it being used everywhere just because the example uses it :_(

On Wed, May 4, 2016, 8:55 AM Tom Burdick notifications@github.com wrote:

Well one thing seems pretty different almost right away, your explicitly enforcing a new connection on each request in the rust version, in the go version your leaving the decision whether to keep-alive the connection and reuse it up to the implementation. Removing the .header(Connection::close()) changes the program running time on my machine from 10's of seconds to 2.7s

— You are receiving this because you are subscribed to this thread. Reply to this email directly or view it on GitHub https://github.com/hyperium/hyper/issues/777#issuecomment-216910480

[weberc2]

Removing the Connection header and not sharing the client bring the time back down to parity with Go. It does appear it was the connection header, and yes, I copied it from the example. It seems the biggest penalty was still sharing the client; is this penalty expected as well?

[seanmonstar]

The penalty when sharing the Client is due to accidentally keeping a Mutex locked while waiting on DNS. The Mutex locks the map of pooled connections. A patch exists that unlocks the Mutex while waiting on DNS, so that other threads may access the pool quicker.

On Wed, May 4, 2016, 10:05 AM weberc2 notifications@github.com wrote:

Removing the Connection header and not sharing the client bring the time back down to parity with Go. It does appear it was the connection header, and yes, I copied it from the example. It seems the biggest penalty was still sharing the client; is this penalty expected as well?

— You are receiving this because you commented.

Reply to this email directly or view it on GitHub https://github.com/hyperium/hyper/issues/777#issuecomment-216933562

[weberc2]

Cool. Thanks for the input. I'll close this issue since the shared client penalty is a known issue.

[seanmonstar]

Which should be released in 0.9.2 today https://github.com/hyperium/hyper/commit/5fcc04a6cd3dd2962eaefdf23133f88e9242e8b3