Trait par_stream::prelude::ParStreamExt[−][src]

pub trait ParStreamExt {

Show 19 methods    fn tee<T>(self, buf_size: impl Into<Option<usize>>) -> Tee<T>
    where
        Self: 'static + Stream<Item = T> + Sized + Unpin + Send,
        T: 'static + Send + Clone,
    { ... }

    fn wrapping_enumerate<T>(self) -> WrappingEnumerate<T, Self>
    where
        Self: Stream<Item = T> + Sized + Unpin,
    { ... }

    fn reorder_enumerated<T>(self) -> ReorderEnumerated<T, Self>
    where
        Self: Stream<Item = (usize, T)> + Unpin + Sized,
    { ... }

    fn par_then<T, F, Fut>(
        self, 
        config: impl IntoParStreamParams, 
        f: F
    ) -> ParMap<T>
    where
        T: 'static + Send,
        F: 'static + FnMut(Self::Item) -> Fut + Send,
        Fut: 'static + Future<Output = T> + Send,
        Self: 'static + StreamExt + Sized + Unpin + Send,
        Self::Item: Send,
    { ... }

    fn par_then_init<T, B, InitF, MapF, Fut>(
        self, 
        config: impl IntoParStreamParams, 
        init_f: InitF, 
        f: MapF
    ) -> ParMap<T>
    where
        T: 'static + Send,
        B: 'static + Send + Clone,
        InitF: FnMut() -> B,
        MapF: 'static + FnMut(B, Self::Item) -> Fut + Send,
        Fut: 'static + Future<Output = T> + Send,
        Self: 'static + StreamExt + Sized + Unpin + Send,
        Self::Item: Send,
    { ... }

    fn par_then_unordered<T, F, Fut>(
        self, 
        config: impl IntoParStreamParams, 
        f: F
    ) -> ParMapUnordered<T>
    where
        T: 'static + Send,
        F: 'static + FnMut(Self::Item) -> Fut + Send,
        Fut: 'static + Future<Output = T> + Send,
        Self: 'static + StreamExt + Sized + Unpin + Send,
        Self::Item: Send,
    { ... }

    fn par_then_init_unordered<T, B, InitF, MapF, Fut>(
        self, 
        config: impl IntoParStreamParams, 
        init_f: InitF, 
        map_f: MapF
    ) -> ParMapUnordered<T>
    where
        T: 'static + Send,
        B: 'static + Send + Clone,
        InitF: FnMut() -> B,
        MapF: 'static + FnMut(B, Self::Item) -> Fut + Send,
        Fut: 'static + Future<Output = T> + Send,
        Self: 'static + StreamExt + Sized + Unpin + Send,
        Self::Item: Send,
    { ... }

    fn par_map<T, F, Func>(
        self, 
        config: impl IntoParStreamParams, 
        f: F
    ) -> ParMap<T>
    where
        T: 'static + Send,
        F: 'static + FnMut(Self::Item) -> Func + Send,
        Func: 'static + FnOnce() -> T + Send,
        Self: 'static + StreamExt + Sized + Unpin + Send,
        Self::Item: Send,
    { ... }

    fn par_map_init<T, B, InitF, MapF, Func>(
        self, 
        config: impl IntoParStreamParams, 
        init_f: InitF, 
        f: MapF
    ) -> ParMap<T>
    where
        T: 'static + Send,
        B: 'static + Send + Clone,
        InitF: FnMut() -> B,
        MapF: 'static + FnMut(B, Self::Item) -> Func + Send,
        Func: 'static + FnOnce() -> T + Send,
        Self: 'static + StreamExt + Sized + Unpin + Send,
        Self::Item: Send,
    { ... }

    fn par_map_unordered<T, F, Func>(
        self, 
        config: impl IntoParStreamParams, 
        f: F
    ) -> ParMapUnordered<T>
    where
        T: 'static + Send,
        F: 'static + FnMut(Self::Item) -> Func + Send,
        Func: 'static + FnOnce() -> T + Send,
        Self: 'static + StreamExt + Sized + Unpin + Send,
        Self::Item: Send,
    { ... }

    fn par_map_init_unordered<T, B, InitF, MapF, Func>(
        self, 
        config: impl IntoParStreamParams, 
        init_f: InitF, 
        f: MapF
    ) -> ParMapUnordered<T>
    where
        T: 'static + Send,
        B: 'static + Send + Clone,
        InitF: FnMut() -> B,
        MapF: 'static + FnMut(B, Self::Item) -> Func + Send,
        Func: 'static + FnOnce() -> T + Send,
        Self: 'static + StreamExt + Sized + Unpin + Send,
        Self::Item: Send,
    { ... }

    fn par_reduce<F, Fut>(
        self, 
        limit: impl Into<Option<usize>>, 
        buf_size: impl Into<Option<usize>>, 
        f: F
    ) -> ParReduce<Self::Item>ⓘ
Notable traits for ParReduce<T>
impl<T> Future for ParReduce<T>    type Output = T;

    where
        F: 'static + FnMut(Self::Item, Self::Item) -> Fut + Send,
        Fut: 'static + Future<Output = Self::Item> + Send,
        Self: 'static + StreamExt + Sized + Unpin + Send,
        Self::Item: Send,
    { ... }

    fn par_routing<F1, F2, Fut, T>(
        self, 
        buf_size: impl Into<Option<usize>>, 
        routing_fn: F1, 
        map_fns: Vec<F2>
    ) -> ParRouting<T>
    where
        Self: 'static + StreamExt + Sized + Unpin + Send,
        Self::Item: Send,
        F1: 'static + FnMut(&Self::Item) -> usize + Send,
        F2: 'static + FnMut(Self::Item) -> Fut + Send,
        Fut: 'static + Future<Output = T> + Send,
        T: 'static + Send,
    { ... }

    fn par_routing_unordered<F1, F2, Fut, T>(
        self, 
        buf_size: impl Into<Option<usize>>, 
        routing_fn: F1, 
        map_fns: Vec<F2>
    ) -> ParRoutingUnordered<T>
    where
        F1: 'static + FnMut(&Self::Item) -> usize + Send,
        F2: 'static + FnMut(Self::Item) -> Fut + Send,
        Fut: 'static + Future<Output = T> + Send,
        T: 'static + Send,
        Self: 'static + StreamExt + Sized + Unpin + Send,
        Self::Item: Send,
    { ... }

    fn scatter(self, buf_size: impl Into<Option<usize>>) -> Scatter<Self::Item>
    where
        Self: 'static + StreamExt + Sized + Unpin + Send,
        Self::Item: Send,
    { ... }

    fn par_for_each<F, Fut>(
        self, 
        config: impl IntoParStreamParams, 
        f: F
    ) -> ParForEachⓘ
Notable traits for ParForEach
impl Future for ParForEach    type Output = ();

    where
        Self: 'static + Stream + Unpin + Sized + Send,
        Self::Item: Send,
        F: 'static + FnMut(Self::Item) -> Fut + Send,
        Fut: 'static + Future<Output = ()> + Send,
    { ... }

    fn par_for_each_init<B, InitF, MapF, Fut>(
        self, 
        config: impl IntoParStreamParams, 
        init_f: InitF, 
        map_f: MapF
    ) -> ParForEachⓘ
Notable traits for ParForEach
impl Future for ParForEach    type Output = ();

    where
        Self: 'static + Stream + Unpin + Sized + Send,
        Self::Item: Send,
        B: 'static + Send + Clone,
        InitF: FnMut() -> B,
        MapF: 'static + FnMut(B, Self::Item) -> Fut + Send,
        Fut: 'static + Future<Output = ()> + Send,
    { ... }

    fn par_for_each_blocking<F, Func>(
        self, 
        config: impl IntoParStreamParams, 
        f: F
    ) -> ParForEachⓘ
Notable traits for ParForEach
impl Future for ParForEach    type Output = ();

    where
        Self: 'static + Stream + Unpin + Sized + Send,
        Self::Item: Send,
        F: 'static + FnMut(Self::Item) -> Func + Send,
        Func: 'static + FnOnce() + Send,
    { ... }

    fn par_for_each_blocking_init<B, InitF, MapF, Func>(
        self, 
        config: impl IntoParStreamParams, 
        init_f: InitF, 
        f: MapF
    ) -> ParForEachⓘ
Notable traits for ParForEach
impl Future for ParForEach    type Output = ();

    where
        Self: 'static + Stream + Unpin + Sized + Send,
        Self::Item: Send,
        B: 'static + Send + Clone,
        InitF: FnMut() -> B,
        MapF: 'static + FnMut(B, Self::Item) -> Func + Send,
        Func: 'static + FnOnce() + Send,
    { ... }

}
Expand description

An extension trait for streams providing combinators for parallel processing.

Provided methods

Converts the stream to a cloneable receiver that receiving items in fan-out pattern.

When a receiver is cloned, it creates a separate internal buffer, so that a background worker clones and passes each stream item to available receiver buffers. It can be used to fork a stream into copies and pass them to concurrent workers.

The internal buffer size is determined by the buf_size. If buf_size is None, the buffer size will be unbounded. A background worker is started and keeps copying item to available receivers. The background worker halts when all receivers are dropped.

use futures::stream::StreamExt;
use par_stream::ParStreamExt;

async fn main_async() {
    let orig: Vec<_> = (0..1000).collect();

    let rx1 = futures::stream::iter(orig.clone()).tee(1);
    let rx2 = rx1.clone();
    let rx3 = rx1.clone();

    let fut1 = rx1.map(|val| val).collect();
    let fut2 = rx2.map(|val| val * 2).collect();
    let fut3 = rx3.map(|val| val * 3).collect();

    let (vec1, vec2, vec3): (Vec<_>, Vec<_>, Vec<_>) = futures::join!(fut1, fut2, fut3);
}

Gives the current iteration count that may overflow to zero as well as the next value.

Reorder the input items paired with a iteration count.

The combinator asserts the input item has tuple type (usize, T). It reorders the items according to the first value of input tuple.

It is usually combined with ParStreamExt::wrapping_enumerate, then applies a series of unordered parallel mapping, and finally reorders the values back by this method. It avoids reordering the values after each parallel mapping step.

use futures::stream::StreamExt;
use par_stream::ParStreamExt;

async fn main_async() {
    let doubled = futures::stream::iter(0..1000)
        // add enumerated index that does not panic on overflow
        .wrapping_enumerate()
        // double the values in parallel
        .par_then_unordered(None, move |(index, value)| {
            // the closure is sent to parallel worker
            async move { (index, value * 2) }
        })
        // add values by one in parallel
        .par_then_unordered(None, move |(index, value)| {
            // the closure is sent to parallel worker
            async move { (index, value + 1) }
        })
        // reorder the values by enumerated index
        .reorder_enumerated()
        .collect::<Vec<_>>()
        .await;
    let expect = (0..1000).map(|value| value * 2 + 1).collect::<Vec<_>>();
    assert_eq!(doubled, expect);
}

Computes new items from the stream asynchronously in parallel with respect to the input order.

The limit is the number of parallel workers. If it is 0 or None, it defaults the number of cores on system. The method guarantees the order of output items obeys that of input items.

Each parallel task runs in two-stage manner. The f closure is invoked in the main thread and lets you clone over outer varaibles. Then, f returns a future and the future will be sent to a parallel worker.

use futures::stream::StreamExt;
use par_stream::ParStreamExt;

async fn main_async() {
    let outer = Box::new(2);

    let doubled = futures::stream::iter(0..1000)
        // doubles the values in parallel up to maximum number of cores
        .par_then(None, move |value| {
            // cloned needed variables in the main thread
            let cloned_outer = outer.clone();

            // the future is sent to a parallel worker
            async move { value * (*cloned_outer) }
        })
        // the collected values will be ordered
        .collect::<Vec<_>>()
        .await;
    let expect = (0..1000).map(|value| value * 2).collect::<Vec<_>>();
    assert_eq!(doubled, expect);
}

Creates a parallel stream with in-local thread initializer.

Computes new items from the stream asynchronously in parallel without respecting the input order.

The limit is the number of parallel workers. If it is 0 or None, it defaults the number of cores on system. The order of output items is not guaranteed to respect the order of input items.

Each parallel task runs in two-stage manner. The f closure is invoked in the main thread and lets you clone over outer varaibles. Then, f returns a future and the future will be sent to a parallel worker.

use futures::stream::StreamExt;
use par_stream::ParStreamExt;
use std::collections::HashSet;

async fn main_async() {
    let outer = Box::new(2);

    let doubled = futures::stream::iter(0..1000)
        // doubles the values in parallel up to maximum number of cores
        .par_then_unordered(None, move |value| {
            // clone needed variables in the main thread
            let cloned_outer = outer.clone();

            // the future is sent to a parallel worker
            async move { value * (*cloned_outer) }
        })
        // the collected values may NOT be ordered
        .collect::<HashSet<_>>()
        .await;
    let expect = (0..1000).map(|value| value * 2).collect::<HashSet<_>>();
    assert_eq!(doubled, expect);
}

Creates a stream analogous to par_then_unordered with in-local thread initializer.

Computes new items in a function in parallel with respect to the input order.

The limit is the number of parallel workers. If it is 0 or None, it defaults the number of cores on system. The method guarantees the order of output items obeys that of input items.

Each parallel task runs in two-stage manner. The f closure is invoked in the main thread and lets you clone over outer varaibles. Then, f returns a closure and the closure will be sent to a parallel worker.

use futures::stream::StreamExt;
use par_stream::ParStreamExt;

async fn main_async() {
    // the variable will be shared by parallel workers
    let outer = Box::new(2);

    let doubled = futures::stream::iter(0..1000)
        // doubles the values in parallel up to maximum number of cores
        .par_map(None, move |value| {
            // clone needed variables in the main thread
            let cloned_outer = outer.clone();

            // the closure is sent to parallel worker
            move || value * (*cloned_outer)
        })
        // the collected values may NOT be ordered
        .collect::<Vec<_>>()
        .await;
    let expect = (0..1000).map(|value| value * 2).collect::<Vec<_>>();
    assert_eq!(doubled, expect);
}

Creates a parallel stream analogous to par_map with in-local thread initializer.

Computes new items in a function in parallel without respecting the input order.

The limit is the number of parallel workers. If it is 0 or None, it defaults the number of cores on system. The method guarantees the order of output items obeys that of input items.

Each parallel task runs in two-stage manner. The f closure is invoked in the main thread and lets you clone over outer varaibles. Then, f returns a future and the future will be sent to a parallel worker.

use futures::stream::StreamExt;
use par_stream::ParStreamExt;
use std::collections::HashSet;

async fn main_async() {
    // the variable will be shared by parallel workers
    let outer = Box::new(2);

    let doubled = futures::stream::iter(0..1000)
        // doubles the values in parallel up to maximum number of cores
        .par_map_unordered(None, move |value| {
            // clone needed variables in the main thread
            let cloned_outer = outer.clone();

            // the closure is sent to parallel worker
            move || value * (*cloned_outer)
        })
        // the collected values may NOT be ordered
        .collect::<HashSet<_>>()
        .await;
    let expect = (0..1000).map(|value| value * 2).collect::<HashSet<_>>();
    assert_eq!(doubled, expect);
}

Creates a parallel stream analogous to par_map_unordered with in-local thread initializer.

Reduces the input items into single value in parallel.

The limit is the number of parallel workers. If it is 0 or None, it defaults the number of cores on system.

The buf_size is the size of buffer that stores the temporary reduced values. If it is 0 or None, it defaults the number of cores on system.

Unlike [StreamExt::fold], the method does not combine the values sequentially. Instead, the parallel workers greedly take two values from the buffer, reduce to one value, and push back to the buffer.

use futures::stream::StreamExt;
use par_stream::ParStreamExt;

async fn main_async() {
    // the variable will be shared by parallel workers
    let sum = futures::stream::iter(1..=1000)
        // sum up the values in parallel
        .par_reduce(None, None, move |lhs, rhs| {
            // the closure is sent to parallel worker
            async move { lhs + rhs }
        })
        .await;
    assert_eq!(sum, (1 + 1000) * 1000 / 2);
}

Distributes input items to specific workers and compute new items with respect to the input order.

The buf_size is the size of input buffer before each mapping function. If it is 0 or None, it defaults the number of cores on system.

routing_fn assigns input items to specific indexes of mapping functions. routing_fn is executed on the calling thread.

map_fns is a vector of mapping functions, each of which produces an asynchronous closure.

use futures::stream::StreamExt;
use par_stream::ParStreamExt;
use std::{future::Future, pin::Pin};

async fn main_async() {
    let map_fns: Vec<
        Box<dyn FnMut(usize) -> Pin<Box<dyn Future<Output = usize> + Send>> + Send>,
    > = vec![
        // even number processor
        Box::new(|even_value| Box::pin(async move { even_value / 2 })),
        // odd number processor
        Box::new(|odd_value| Box::pin(async move { odd_value * 2 + 1 })),
    ];

    let transformed = futures::stream::iter(0..1000)
        // doubles the values in parallel up to maximum number of cores
        .par_routing(
            None,
            move |value| {
                // distribute the value according to its parity
                if value % 2 == 0 {
                    0
                } else {
                    1
                }
            },
            map_fns,
        )
        // the collected values may NOT be ordered
        .collect::<Vec<_>>()
        .await;
    let expect = (0..1000)
        .map(|value| {
            if value % 2 == 0 {
                value / 2
            } else {
                value * 2 + 1
            }
        })
        .collect::<Vec<_>>();
    assert_eq!(transformed, expect);
}

Distributes input items to specific workers and compute new items without respecting the input order.

The buf_size is the size of input buffer before each mapping function. If it is 0 or None, it defaults the number of cores on system.

routing_fn assigns input items to specific indexes of mapping functions. routing_fn is executed on the calling thread.

map_fns is a vector of mapping functions, each of which produces an asynchronous closure.

Splits the stream into a receiver and a future.

The returned future scatters input items into the receiver and its clones, and should be manually awaited by user.

The returned receiver can be cloned and distributed to resepctive workers.

It lets user to write custom workers that receive items from the same stream.

use futures::stream::StreamExt;
use par_stream::ParStreamExt;

async fn main_async() {
    let orig = futures::stream::iter(1isize..=1000);

    // scatter the items
    let rx1 = orig.scatter(None);
    let rx2 = rx1.clone();

    // collect the values concurrently
    let (values1, values2): (Vec<_>, Vec<_>) = futures::join!(rx1.collect(), rx2.collect());

    // the total item count is equal to the original set
    assert_eq!(values1.len() + values2.len(), 1000);
}

Runs an asynchronous task on each element of an stream in parallel.

Creates a parallel stream analogous to par_for_each with a in-local thread initializer.

Runs an blocking task on each element of an stream in parallel.

Creates a parallel stream analogous to par_for_each_blocking with a in-local thread initializer.

Implementors