Can actix_web client run in multiple threads?

[flyinmind]

For example, realize a benchmark tool with actix-web client.
Run several clients in different thread.
But failed to compile it,

let h = tokio::task::spawn(async move {
| ^^^^^^^^^^^^^^^^^^ future created by async block is not Send

Maybe,ClientResponse not implemented Send, so failed,.
Anybody know how to realize the function with actix-web client? thanks!
I use actix_web 0.13.x , rust 1.47 on windows 10.

//----------------------------------------------------------------------------------
use actix_web::client::Client;
use std::time::SystemTime;
use actix::prelude::*;
use tokio::task::{self, JoinHandle};
use std::sync::Arc;

const threadNum : usize = 8;
const runTimes : u64 = 5;

#[actix_web::main]
async fn main() {
let start = SystemTime::now();
let mut i : usize = 0;
//let sys = actix::System::new("test");
let mut handlers = Vec::<task::JoinHandle>::with_capacity(threadNum);
let client = Client::default();

while i < threadNum {
    i += 1;
    let no = i;
    let h = tokio::task::spawn(async move {
        test127(no).await
    });
    handlers.push(h);
}

let mut errNum : usize = 0;
for h in handlers {
    errNum += h.await.unwrap();
}
//sys.run();
let interval = start.elapsed().unwrap().as_millis() as u64;

println!("thread-num:{},use-time:{} ms,speed:{}, failed:{}", 
    threadNum, interval, runTimes * 1000 * (threadNum as u64) / interval, errNum);

}

async fn test127(threadNo : usize) -> usize {
let client = Client::default();
let mut errNum :usize = 0;
let mut i : u64 = 0;
while i < runTimes {
i += 1;
// Create request builder and send request

    let req = client.get("http://127.0.0.1:8080");

//if remove following code, success to compile, but it is useless
match req.send().await { // <- Wait for response
Ok(mut response) => {
match response.body().await {
Ok(buf) => {
println!("Response: {},{}", String::from_utf8(buf.to_vec()).unwrap(), threadNo);
}
Err(e) => {
errNum += 1;
//println!("Fail to get body, error:{:?}", &e);
}
}
}
Err(e) => {
errNum += 1;
//println!("Fail to get resposne, error:{:?}", &e);
}
}
}
errNum
}

[fakeshadow]

1. Since you are using atcix::main macro. It means you are running on a single thread runtime. calling tokio::task::spawn would not make your task run on other threads.

2. tokio::task::spawn requires the future you spawn on it have Send bound and it's not possible for actix_web::client::Client because it's not a Send bound type.

3. most async runtimes in Rust use M:N scheduling. It means they don't spawn new threads with every spawn function.

[flyinmind]

Sorry, I am a newbie in Rust, only one month :(. Maybe it is a stupid question.
Does it mean that Actix web client need not run in multi threads?
If not that, how to do it?
I tried to run request in a WHILE, in main thread. It is too slow relative to other tools.

[fakeshadow]

No worry. It's a common question around here.
In short yes. actix-web's client is single thread only. The common strategy is to run one client on every thread and you may need actix's actor pattern to help you set that up if you are not using it inside actix-web app.

[flyinmind]

I wrote a example, there are some problems to ask for advice.
I want to realize a micro-service framework by actix-web, so need to call other service in a actix-web service.
In the following example,

1. When calling http://127.0.0.1/hello/yourname, performance is very good, abount 70000/s;
2. When calling http://127.0.0.1/test/yourname, in the service, will use Client to call http://127.0.0.1/hello/yourname,
   performance is very bad, about 1500/s. Is it because of creating client in every request? Connection pool can't play a role, not sure.
3. If use singleton in test(req: HttpRequest), can't work when concurrently. I know, it shouldn't work, but I wish it could.
4. How to optimize the situation, user's requests --> actix-web-server --> actix-web-client --> other service? Is there an example?

use actix_web::{get, App, HttpRequest, HttpResponse, HttpServer};
//-----------------------------------------------------------------------------
mod webclient {
    use std::sync::Arc;
    use actix_web::client::Client;
    use std::time::Duration;
    use std::collections::HashMap;
    use serde_json::{Map, Value};
    use std::net::SocketAddr;

    const HEAD_CONTENT_TYPE : &str = "Content-type";

    pub struct WebClient {
        client : Client
    }

    impl WebClient {
        pub fn instance() -> &'static WebClient {
            static mut INS: Option<Arc<WebClient>> = None;
            unsafe {
                INS.get_or_insert_with(|| {
                    let client = WebClient::new();
                    Arc::new(client)
                });
                INS.as_ref().unwrap()
            }
        }
        /// Initialize a new http client.
        /// If need parse dns yourself, set dnsResolver
        pub fn new() -> Self {
            let client = Client::builder()
                .disable_redirects()
                .finish();

            WebClient{client}
        }

        /// Get request with parameters
        /// return a string
        pub async fn get(&self, url : &str, timeout : u64) -> String {
            let mut cr = self.client.get(url);
            cr = cr.timeout(Duration::from_secs(timeout));
            match cr.send().await {
                Ok(mut response) => {
                    match response.body().await {
                        Ok(buf) => String::from_utf8(buf.to_vec()).unwrap(),
                        Err(e) => format!("Fail to response after getting, err:{:?}, url:{}", &e, url)
                    }
                }
                Err(e) => format!("Fail to send body when getting, err:{:?}, url:{}", &e, url)
            }
        }
    }
}
//-----------------------------------------------------------------------------
#[get("/hello/{name}")]
async fn hello(req: HttpRequest, name: web::Path<String>) -> String {
    //println!("1==>REQ: {:?}", req);
    format!("Hello: {}!\r\n", name)
}

#[get("/test/*")]
async fn test(req: HttpRequest) -> String {
    //println!("2==>REQ: {:?}", req);
    //let client = webclient::WebClient::instance(); //**can't work when concurrently request**
    let client = webclient::WebClient::new(); //**too slow, about 1500/s vs 70000/s**
    client.get("http://127.0.0.1:8080/hello/ketty",  3).await
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    //std::env::set_var("RUST_LOG", "actix_server=info,actix_web=info");
    //env_logger::init();

    HttpServer::new(|| {
        App::new()
            .wrap(middleware::DefaultHeaders::new().header("X-Version", "0.2"))
            .wrap(middleware::Compress::default())
            .wrap(middleware::Logger::default())
            .service(hello)
            .service(test)
            // .service(web::resource("/test1.html").to(|| async { "Test\r\n" }))
    })
    .bind("127.0.0.1:8080")?
    .workers(8)
    .run()
    .await
}

[robjtede]

There are some simple rules with the client:

• do not build a new client for each request
• use app data to store one client per worker for minimum contention

[fakeshadow]

Like robjtede pointed out. You don't build anything you want to reuse with every request.

[flyinmind]

Thanks for your advice. It is ok now!
Only a blemish in an otherwise perfect thing，
In my test case, client and server are all in the same PC:

1. testclient(8 thread) --> actix-web-server ==> about 74000/s;
2. testclient(8 thread) --> actix-web-server -> actix-web-client --> actix-web-server(another url) ==> about 12000/s, more better than before(1500/s), create client every time.
   I think in test case 2, it should be half of case 1.
   Although it is acceptable, but I want to know is there any way to improve it?

use actix_web::{get, web, middleware, App, HttpRequest, /*HttpResponse,*/ HttpServer};
//-----------------------------------------------------------------------------
mod webclient {
    use actix_web::client::Client;
    use std::time::Duration;

    pub struct WebClient {
        client : Client
    }

    impl WebClient {
        /// Initialize a new http client.
        /// If need parse dns yourself, set dnsResolver
        pub fn new() -> Self {
            let client = Client::builder()
                .disable_redirects()
                .finish();

            WebClient{client}
        }

        /// Get request with parameters
        /// return a string
        pub async fn get(&self, url : &str, timeout : u64) -> String {
            let mut cr = self.client.get(url);
            cr = cr.timeout(Duration::from_secs(timeout));
            match cr.send().await {
                Ok(mut response) => {
                    match response.body().await {
                        Ok(buf) => String::from_utf8(buf.to_vec()).unwrap(),
                        Err(e) => format!("Fail to response after getting, err:{:?}, url:{}", &e, url)
                    }
                }
                Err(e) => format!("Fail to send body when getting, err:{:?}, url:{}", &e, url)
            }
        }
    }
}
//-----------------------------------------------------------------------------
#[get("/hello/{name}")]
async fn hello(_req: HttpRequest, name: web::Path<String>) -> String {
    //println!("1==>REQ: {:?}", req);
    format!("Hello: {}!\r\n", name)
}

#[get("/test/*")]
async fn test(_req: HttpRequest, client: web::Data<webclient::WebClient>) -> String {
    //println!("2==>REQ: {:?}", req);
    //let client = webclient::WebClient::new(); //**too slow, about 1500/s vs 70000/s**
    client.get("http://127.0.0.1:8080/hello/ketty", 3 as u64).await //ok about 12000/s vs 75000/s
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    //std::env::set_var("RUST_LOG", "actix_server=info,actix_web=info");
    //env_logger::init();

    HttpServer::new(|| {
        App::new()
            .wrap(middleware::DefaultHeaders::new().header("X-Version", "0.2"))
            .wrap(middleware::Compress::default())
            //.wrap(middleware::Logger::default())
            .data(webclient::WebClient::new())
            .service(hello)
            .service(test)
            // .service(web::resource("/test1.html").to(|| async { "Test\r\n" }))
    })
    .bind("127.0.0.1:8080")?
    .workers(8)
    .run()
    .await
}

[Ujang360]

I'm not sure with the exact culprit of the performance drop of that request piping in your async fn test @flyinmind, but I'm pretty sure compress-decompress happens more than once. Or maybe I'm not that sure 😀