Struct FuturesUnorderedBounded

pub struct FuturesUnorderedBounded<F> { /* private fields */ }

A set of futures which may complete in any order.

Much like futures::stream::FuturesUnordered, this is a thread-safe, Pin friendly, lifetime friendly, concurrent processing stream.

The is different to FuturesUnordered in that FuturesUnorderedBounded has a fixed capacity for processing count. This means it’s less flexible, but produces better memory efficiency.

Benchmarks

Speed

Running 65536 100us timers with 256 concurrent jobs in a single threaded tokio runtime:

FuturesUnordered         time:   [420.47 ms 422.21 ms 423.99 ms]
FuturesUnorderedBounded  time:   [366.02 ms 367.54 ms 369.05 ms]

Memory usage

Running 512000 Ready<i32> futures with 256 concurrent jobs.

• count: the number of times alloc/dealloc was called
• alloc: the number of cumulative bytes allocated
• dealloc: the number of cumulative bytes deallocated

FuturesUnordered
    count:    1024002
    alloc:    40960144 B
    dealloc:  40960000 B

FuturesUnorderedBounded
    count:    2
    alloc:    8264 B
    dealloc:  0 B

Conclusion

As you can see, FuturesUnorderedBounded massively reduces you memory overhead while providing a significant performance gain. Perfect for if you want a fixed batch size

Example

Making 1024 total HTTP requests, with a max concurrency of 128

use futures::future::Future;
use futures::stream::StreamExt;
use futures_buffered::FuturesUnorderedBounded;
use hyper::client::conn::http1::{handshake, SendRequest};
use hyper::body::Incoming;
use hyper::{Request, Response};
use hyper_util::rt::TokioIo;
use tokio::net::TcpStream;

// create a tcp connection
let stream = TcpStream::connect("example.com:80").await?;

// perform the http handshakes
let (mut rs, conn) = handshake(TokioIo::new(stream)).await?;
tokio::spawn(conn);

/// make http request to example.com and read the response
fn make_req(rs: &mut SendRequest<String>) -> impl Future<Output = hyper::Result<Response<Incoming>>> {
    let req = Request::builder()
        .header("Host", "example.com")
        .method("GET")
        .body(String::new())
        .unwrap();
    rs.send_request(req)
}

// create a queue that can hold 128 concurrent requests
let mut queue = FuturesUnorderedBounded::new(128);

// start up 128 requests
for _ in 0..128 {
    queue.push(make_req(&mut rs));
}
// wait for a request to finish and start another to fill its place - up to 1024 total requests
for _ in 128..1024 {
    queue.next().await;
    queue.push(make_req(&mut rs));
}
// wait for the tail end to finish
for _ in 0..128 {
    queue.next().await;
}

Implementations

impl<F> FuturesUnorderedBounded<F>

pub fn new(cap: usize) -> Self

Constructs a new, empty FuturesUnorderedBounded with the given fixed capacity.

The returned FuturesUnorderedBounded does not contain any futures. In this state, FuturesUnorderedBounded::poll_next will return Poll::Ready(None).

pub fn push(&mut self, fut: F)

Push a future into the set.

This method adds the given future to the set. This method will not call poll on the submitted future. The caller must ensure that FuturesUnorderedBounded::poll_next is called in order to receive wake-up notifications for the given future.

Panics

This method will panic if the buffer is currently full. See FuturesUnorderedBounded::try_push to get a result instead

pub fn try_push(&mut self, fut: F) -> Result<(), F>

Push a future into the set.

This method adds the given future to the set. This method will not call poll on the submitted future. The caller must ensure that FuturesUnorderedBounded::poll_next is called in order to receive wake-up notifications for the given future.

Errors

This method will error if the buffer is currently full, returning the future back

pub fn is_empty(&self) -> bool

Returns true if the set contains no futures.

pub fn len(&self) -> usize

Returns the number of futures contained in the set.

This represents the total number of in-flight futures.

pub fn capacity(&self) -> usize

Returns the number of futures that can be contained in the set.

Trait Implementations

impl<Fut> Debug for FuturesUnorderedBounded<Fut>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<F> FromIterator<F> for FuturesUnorderedBounded<F>

fn from_iter<T: IntoIterator<Item = F>>(iter: T) -> Self

Constructs a new, empty FuturesUnorderedBounded with a fixed capacity that is the length of the iterator.

Example

Making 1024 total HTTP requests, with a max concurrency of 128

use futures::future::Future;
use futures::stream::StreamExt;
use futures_buffered::FuturesUnorderedBounded;
use hyper::client::conn::http1::{handshake, SendRequest};
use hyper::body::Incoming;
use hyper::{Request, Response};
use hyper_util::rt::TokioIo;
use tokio::net::TcpStream;

// create a tcp connection
let stream = TcpStream::connect("example.com:80").await?;

// perform the http handshakes
let (mut rs, conn) = handshake(TokioIo::new(stream)).await?;
tokio::spawn(conn);

/// make http request to example.com and read the response
fn make_req(rs: &mut SendRequest<String>) -> impl Future<Output = hyper::Result<Response<Incoming>>> {
    let req = Request::builder()
        .header("Host", "example.com")
        .method("GET")
        .body(String::new())
        .unwrap();
    rs.send_request(req)
}

// create a queue with an initial 128 concurrent requests
let mut queue: FuturesUnorderedBounded<_> = (0..128).map(|_| make_req(&mut rs)).collect();

// wait for a request to finish and start another to fill its place - up to 1024 total requests
for _ in 128..1024 {
    queue.next().await;
    queue.push(make_req(&mut rs));
}
// wait for the tail end to finish
for _ in 0..128 {
    queue.next().await;
}

impl<Fut: Future> FusedStream for FuturesUnorderedBounded<Fut>

fn is_terminated(&self) -> bool

Returns true if the stream should no longer be polled.

impl<F: Future> Stream for FuturesUnorderedBounded<F>

type Item = <F as Future>::Output

Values yielded by the stream.

fn poll_next( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.

impl<F> Unpin for FuturesUnorderedBounded<F>