How to make several requests at once

Short answer: futures::future::join_all polls every future in the list at once and hands back the results in input order — but it starts all of them, so five requests mean five connections and five slots on the server. buffer_unordered(n), from the same crate, is the same idea with a cap on how many run at the same time.

The dependencies

Three crates. reqwest sends the requests, tokio drives them, and futures supplies the two combinators — reqwest doesn’t re-export them:

[dependencies]
reqwest = "0.13.5"
tokio = { version = "1.53.3", features = ["macros", "rt-multi-thread"] }
futures = "0.3.34"

The test server

Each block below runs against the same small server, written with nothing but std. /<i> sleeps for a fixed per-index delay, then answers 200 with the index as the body. /stats reports three numbers: how many requests were handled, the peak number in flight at once, and the order the responses were written. The delays are what make the overlap visible — the last URL answers first.

One at a time

The baseline. Five requests, one .await at a time:

use std::io::{BufRead, BufReader, Write};
use std::net::TcpListener;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

const N: usize = 5;
const DELAYS: [u64; N] = [1000, 700, 500, 400, 100];

/// A one-file test server. `/<i>` sleeps for DELAYS[i], then answers 200 with the
/// index as the body. `/stats` reports how many requests it handled, the peak
/// number in flight at once, and the order the responses were written.
fn spawn_server() -> String {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let base = format!("http://{}", listener.local_addr().unwrap());
    let in_flight = Arc::new(AtomicUsize::new(0));
    let peak = Arc::new(AtomicUsize::new(0));
    let handled = Arc::new(AtomicUsize::new(0));
    let done_order = Arc::new(Mutex::new(Vec::<usize>::new()));
    thread::spawn(move || {
        for stream in listener.incoming() {
            let stream = stream.unwrap();
            let in_flight = Arc::clone(&in_flight);
            let peak = Arc::clone(&peak);
            let handled = Arc::clone(&handled);
            let done_order = Arc::clone(&done_order);
            thread::spawn(move || {
                let mut reader = BufReader::new(stream.try_clone().unwrap());
                let mut writer = stream;
                let mut request_line = String::new();
                if reader.read_line(&mut request_line).unwrap() == 0 {
                    return;
                }
                let path = request_line.split_whitespace().nth(1).unwrap_or("/").to_owned();
                if path == "/stats" {
                    let body = format!(
                        "handled={} peak={} done={}",
                        handled.load(Ordering::SeqCst),
                        peak.load(Ordering::SeqCst),
                        done_order.lock().unwrap().iter().map(|i| i.to_string()).collect::<Vec<_>>().join(",")
                    );
                    respond(&mut writer, &body);
                    return;
                }
                let index: usize = path.trim_start_matches('/').parse().unwrap();
                let now = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
                peak.fetch_max(now, Ordering::SeqCst);
                thread::sleep(Duration::from_millis(DELAYS[index]));
                in_flight.fetch_sub(1, Ordering::SeqCst);
                handled.fetch_add(1, Ordering::SeqCst);
                done_order.lock().unwrap().push(index);
                respond(&mut writer, &index.to_string());
            });
        }
    });
    base
}

fn respond(writer: &mut impl Write, body: &str) {
    write!(
        writer,
        "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
        body.len()
    )
    .unwrap();
    writer.flush().unwrap();
}

#[tokio::main]
async fn main() {
    let base = spawn_server();
    let client = reqwest::Client::new();
    for i in 0..N {
        let response = client.get(format!("{base}/{i}")).send().await.unwrap();
        println!("/{i} -> {}", response.status());
    }
    let stats = client
        .get(format!("{base}/stats"))
        .send()
        .await
        .unwrap()
        .text()
        .await
        .unwrap();
    println!("{stats}");
}

Run this in the Rust Playground

Run it with cargo run (cargo’s own Compiling and Finished lines are not shown here):

/0 -> 200 OK
/1 -> 200 OK
/2 -> 200 OK
/3 -> 200 OK
/4 -> 200 OK
handled=5 peak=1 done=0,1,2,3,4

peak=1 is the whole story: each response arrives before the next request is sent. The five delays add up to 2.7 seconds of waiting for 2.7 seconds of server time.

All at once

join_all takes a Vec of futures and polls them all on one task:

use std::io::{BufRead, BufReader, Write};
use std::net::TcpListener;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

const N: usize = 5;
const DELAYS: [u64; N] = [1000, 700, 500, 400, 100];

/// A one-file test server. `/<i>` sleeps for DELAYS[i], then answers 200 with the
/// index as the body. `/stats` reports how many requests it handled, the peak
/// number in flight at once, and the order the responses were written.
fn spawn_server() -> String {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let base = format!("http://{}", listener.local_addr().unwrap());
    let in_flight = Arc::new(AtomicUsize::new(0));
    let peak = Arc::new(AtomicUsize::new(0));
    let handled = Arc::new(AtomicUsize::new(0));
    let done_order = Arc::new(Mutex::new(Vec::<usize>::new()));
    thread::spawn(move || {
        for stream in listener.incoming() {
            let stream = stream.unwrap();
            let in_flight = Arc::clone(&in_flight);
            let peak = Arc::clone(&peak);
            let handled = Arc::clone(&handled);
            let done_order = Arc::clone(&done_order);
            thread::spawn(move || {
                let mut reader = BufReader::new(stream.try_clone().unwrap());
                let mut writer = stream;
                let mut request_line = String::new();
                if reader.read_line(&mut request_line).unwrap() == 0 {
                    return;
                }
                let path = request_line.split_whitespace().nth(1).unwrap_or("/").to_owned();
                if path == "/stats" {
                    let body = format!(
                        "handled={} peak={} done={}",
                        handled.load(Ordering::SeqCst),
                        peak.load(Ordering::SeqCst),
                        done_order.lock().unwrap().iter().map(|i| i.to_string()).collect::<Vec<_>>().join(",")
                    );
                    respond(&mut writer, &body);
                    return;
                }
                let index: usize = path.trim_start_matches('/').parse().unwrap();
                let now = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
                peak.fetch_max(now, Ordering::SeqCst);
                thread::sleep(Duration::from_millis(DELAYS[index]));
                in_flight.fetch_sub(1, Ordering::SeqCst);
                handled.fetch_add(1, Ordering::SeqCst);
                done_order.lock().unwrap().push(index);
                respond(&mut writer, &index.to_string());
            });
        }
    });
    base
}

fn respond(writer: &mut impl Write, body: &str) {
    write!(
        writer,
        "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
        body.len()
    )
    .unwrap();
    writer.flush().unwrap();
}

#[tokio::main]
async fn main() {
    let base = spawn_server();
    let client = reqwest::Client::new();
    let requests: Vec<_> = (0..N).map(|i| client.get(format!("{base}/{i}")).send()).collect();
    let responses = futures::future::join_all(requests).await;
    for (i, response) in responses.iter().enumerate() {
        println!("/{i} -> {}", response.as_ref().unwrap().status());
    }
    let stats = client
        .get(format!("{base}/stats"))
        .send()
        .await
        .unwrap()
        .text()
        .await
        .unwrap();
    println!("{stats}");
}

Run this in the Rust Playground

Run it with cargo run:

/0 -> 200 OK
/1 -> 200 OK
/2 -> 200 OK
/3 -> 200 OK
/4 -> 200 OK
handled=5 peak=5 done=4,3,2,1,0

peak=5: five connections were open at the same time, and the batch now takes as long as the slowest request — one second — instead of the sum. The done= line is the server’s view of that overlap: /4 finished first, because it has the shortest delay. join_all still returned the results in input order, /0 first. Futures are lazy, so collecting them into the Vec sends nothing; join_all is the thing that polls all five.

The limit you forgot

join_all has no cap. Give it a thousand URLs and it opens a thousand connections, and the server meets all of them at once. buffer_unordered(n) starts the first n and adds another as each one finishes:

use std::io::{BufRead, BufReader, Write};
use std::net::TcpListener;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

use futures::stream::{self, StreamExt};

const N: usize = 5;
const DELAYS: [u64; N] = [1000, 700, 500, 400, 100];

/// A one-file test server. `/<i>` sleeps for DELAYS[i], then answers 200 with the
/// index as the body. `/stats` reports how many requests it handled, the peak
/// number in flight at once, and the order the responses were written.
fn spawn_server() -> String {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let base = format!("http://{}", listener.local_addr().unwrap());
    let in_flight = Arc::new(AtomicUsize::new(0));
    let peak = Arc::new(AtomicUsize::new(0));
    let handled = Arc::new(AtomicUsize::new(0));
    let done_order = Arc::new(Mutex::new(Vec::<usize>::new()));
    thread::spawn(move || {
        for stream in listener.incoming() {
            let stream = stream.unwrap();
            let in_flight = Arc::clone(&in_flight);
            let peak = Arc::clone(&peak);
            let handled = Arc::clone(&handled);
            let done_order = Arc::clone(&done_order);
            thread::spawn(move || {
                let mut reader = BufReader::new(stream.try_clone().unwrap());
                let mut writer = stream;
                let mut request_line = String::new();
                if reader.read_line(&mut request_line).unwrap() == 0 {
                    return;
                }
                let path = request_line.split_whitespace().nth(1).unwrap_or("/").to_owned();
                if path == "/stats" {
                    let body = format!(
                        "handled={} peak={} done={}",
                        handled.load(Ordering::SeqCst),
                        peak.load(Ordering::SeqCst),
                        done_order.lock().unwrap().iter().map(|i| i.to_string()).collect::<Vec<_>>().join(",")
                    );
                    respond(&mut writer, &body);
                    return;
                }
                let index: usize = path.trim_start_matches('/').parse().unwrap();
                let now = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
                peak.fetch_max(now, Ordering::SeqCst);
                thread::sleep(Duration::from_millis(DELAYS[index]));
                in_flight.fetch_sub(1, Ordering::SeqCst);
                handled.fetch_add(1, Ordering::SeqCst);
                done_order.lock().unwrap().push(index);
                respond(&mut writer, &index.to_string());
            });
        }
    });
    base
}

fn respond(writer: &mut impl Write, body: &str) {
    write!(
        writer,
        "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
        body.len()
    )
    .unwrap();
    writer.flush().unwrap();
}

#[tokio::main]
async fn main() {
    let base = spawn_server();
    let client = reqwest::Client::new();
    let bodies: Vec<String> = stream::iter(0..N)
        .map(|i| {
            let client = client.clone();
            let url = format!("{base}/{i}");
            async move { client.get(url).send().await.unwrap().text().await.unwrap() }
        })
        .buffer_unordered(2)
        .collect()
        .await;
    println!("bodies: {bodies:?}");
    let stats = client
        .get(format!("{base}/stats"))
        .send()
        .await
        .unwrap()
        .text()
        .await
        .unwrap();
    println!("{stats}");
}

Run this in the Rust Playground

Run it with cargo run:

bodies: ["1", "0", "2", "4", "3"]
handled=5 peak=2 done=1,0,2,4,3

peak=2 — never more than two requests in flight, and the batch takes 1.4 seconds rather than one. Note the order: buffer_unordered yields in completion order, so /1 comes back before /0, and the server’s done= line says the same. That order is a property of the delays, not a contract — under load two responses can swap places — so sort the results when input order matters, or stay with join_all when the list is small and bounded. join_all keeps the whole batch on one task; requests that need to be cancelled one by one belong in tokio::task::JoinSet instead.

How to make a GET request — the single request this article multiplies.
How to retry a failed request — retries multiply the request count, so the two limits meet.
How to set a timeout on a request — the timeout is per request, and it is what a slow batch feels.

The next article in this chapter reads the bodies as JSON.

How to make several requests at once

The dependencies

The test server

One at a time

All at once

The limit you forgot

Next