actix/actix

apache bench keep alive with streaming response

FYI to avoid questions,

• I do thing that ab is quite compact and eats very few resources compared to siege/wrk and most certainly jmeter ;)
• absolutely no problem with curl/wget keep alive gets

assuming such simple server that outputs 10x blah

use actix_web::dev::HttpResponseBuilder;
use actix_web::{middleware, web, App, HttpRequest, HttpResponse, HttpServer};

struct LiveStream {
    offset: i64,
}

impl LiveStream {
    pub fn new() -> LiveStream {
        LiveStream { offset: 10 }
    }
}
// https://gitter.im/actix/actix-web?at=5f94900e7be0d67d2794c0df
impl futures::stream::Stream for LiveStream {
    type Item = Result<web::Bytes, ()>;

    fn poll_next(mut self: std::pin::Pin<&mut Self>, cx: &mut futures::task::Context<'_>) -> futures::task::Poll<Option<Self::Item>> {
        if self.offset <= 0 {
            return futures::task::Poll::Ready(None);
        }
        let bytes = web::Bytes::from("blah");
        return futures::task::Poll::Ready(Some(Ok(bytes)));
    }
}

#[actix_web::get("/foo/{path}")]
fn foo(_: HttpRequest) -> HttpResponse {
    return HttpResponseBuilder::new(actix_web::http::StatusCode::OK)
        .header(actix_web::http::header::CONTENT_LENGTH, 10 * 4 as usize)
        .streaming(LiveStream::new());
}

#[actix_web::get("/bar/{path}")]
fn bar(_: HttpRequest) -> HttpResponse {
    let (tx, rx_body) = actix_utils::mpsc::channel::<actix_web::Result<web::Bytes, actix_web::Error>>();
    println!("starting response");
    actix_web::rt::spawn(async move {
        let mut offset: i64 = 10;

        while offset > 0 {
            offset -= 1;
            if let Err(_) = tx.send(Ok::<_, actix_web::Error>(web::Bytes::from("blah"))) {
                return;
            }
            futures_timer::Delay::new(std::time::Duration::from_micros(1)).await; // required
        }
        println!("response finished");
        drop(tx);
    });

    return HttpResponseBuilder::new(actix_web::http::StatusCode::OK).streaming(rx_body);
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    if std::env::var("RUST_LOG").is_err() {
        std::env::set_var("RUST_LOG", "actix_web=debug");
    }
    env_logger::init();
    let plain_server = HttpServer::new(|| App::new().wrap(middleware::Logger::default()).service(bar).service(foo))
        .keep_alive(16)
        .max_connection_rate(512)
        .max_connections(40000)
        .bind("0.0.0.0:8080")?
        .workers(600)
        .run();

    let servers = vec![plain_server];
    futures::future::join_all(servers).await;

    Ok(())
}

lets say you:
ab -k -c 1 -n 20 "http://127.0.0.1:8080/bar/xxx"
ab -k -c 1 -n 20 "http://127.0.0.1:8080/foo/xxx"
we receive log:

starting response
response finished
[2021-04-09T17:32:59Z INFO  actix_web::middleware::logger] 127.0.0.1:56404 "GET /bar/xxx HTTP/1.0" 200 40 "-" "ApacheBench/2.3" 0.001095
starting response
response finished
[2021-04-09T17:32:59Z INFO  actix_web::middleware::logger] 127.0.0.1:56404 "GET /bar/xxx HTTP/1.0" 200 40 "-" "ApacheBench/2.3" 0.000665
starting response
response finished
[2021-04-09T17:33:15Z INFO  actix_web::middleware::logger] 127.0.0.1:56416 "GET /bar/xxx HTTP/1.0" 200 40 "-" "ApacheBench/2.3" 0.001046
starting response
response finished
[2021-04-09T17:33:15Z INFO  actix_web::middleware::logger] 127.0.0.1:56416 "GET /bar/xxx HTTP/1.0" 200 40 "-" "ApacheBench/2.3" 0.000665
starting response
response finished
[2021-04-09T17:33:31Z INFO  actix_web::middleware::logger] 127.0.0.1:56440 "GET /bar/xxx HTTP/1.0" 200 40 "-" "ApacheBench/2.3" 0.001854
...

ab sems to be timeouting after 2nd response

similar handler based on impl futures::stream::Stream with poll_next has identical behaviour.