Trait par_stream::ParStreamExt[−][src]

pub trait ParStreamExt where
    Self: 'static + Send + Stream + IndexedStreamExt,
    Self::Item: 'static + Send, {

Show 22 methods    fn then_spawned<T, F, Fut>(
        self, 
        buf_size: impl Into<Option<usize>>, 
        f: F
    ) -> ThenSpawned<T>
    where
        T: 'static + Send,
        F: 'static + FnMut(Self::Item) -> Fut + Send,
        Fut: Future<Output = T> + Send;

    fn map_spawned<T, F>(
        self, 
        buf_size: impl Into<Option<usize>>, 
        f: F
    ) -> MapSpawned<T>
    where
        T: 'static + Send,
        F: 'static + FnMut(Self::Item) -> T + Send;

    fn batching<T, F, Fut>(self, f: F) -> Batching<T>
    where
        F: FnOnce(BatchingReceiver<Self::Item>, BatchingSender<T>) -> Fut,
        Fut: 'static + Future<Output = ()> + Send,
        T: 'static + Send;

    fn par_batching_unordered<P, T, F, Fut>(
        self, 
        config: P, 
        f: F
    ) -> ParBatchingUnordered<T>
    where
        F: FnMut(usize, Receiver<Self::Item>, Sender<T>) -> Fut,
        Fut: 'static + Future<Output = ()> + Send,
        T: 'static + Send,
        P: IntoParStreamParams;

    fn tee(self, buf_size: usize) -> Tee<Self::Item>
    where
        Self::Item: Clone;

    fn broadcast(self, buf_size: usize) -> BroadcastGuard<Self::Item>
    where
        Self::Item: Clone;

    fn par_then<P, T, F, Fut>(self, config: P, f: F) -> ParThen<T>
    where
        T: 'static + Send,
        F: 'static + FnMut(Self::Item) -> Fut + Send + Clone,
        Fut: 'static + Future<Output = T> + Send,
        P: IntoParStreamParams;

    fn par_then_init<P, T, B, InitF, MapF, Fut>(
        self, 
        config: P, 
        init_f: InitF, 
        map_f: MapF
    ) -> ParThen<T>
    where
        P: IntoParStreamParams,
        T: 'static + Send,
        B: 'static + Send + Clone,
        InitF: FnOnce() -> B,
        MapF: 'static + FnMut(B, Self::Item) -> Fut + Send,
        Fut: 'static + Future<Output = T> + Send;

    fn par_then_unordered<P, T, F, Fut>(
        self, 
        config: P, 
        f: F
    ) -> ParThenUnordered<T>
    where
        T: 'static + Send,
        F: 'static + FnMut(Self::Item) -> Fut + Send,
        Fut: 'static + Future<Output = T> + Send,
        P: IntoParStreamParams;

    fn par_then_init_unordered<P, T, B, InitF, MapF, Fut>(
        self, 
        config: P, 
        init_f: InitF, 
        map_f: MapF
    ) -> ParThenUnordered<T>
    where
        T: 'static + Send,
        B: 'static + Send + Clone,
        InitF: FnOnce() -> B,
        MapF: 'static + FnMut(B, Self::Item) -> Fut + Send,
        Fut: 'static + Future<Output = T> + Send,
        P: IntoParStreamParams;

    fn par_map<P, T, F, Func>(self, config: P, f: F) -> ParMap<T>
    where
        T: 'static + Send,
        F: 'static + FnMut(Self::Item) -> Func + Send,
        Func: 'static + FnOnce() -> T + Send,
        P: IntoParStreamParams;

    fn par_map_init<P, T, B, InitF, MapF, Func>(
        self, 
        config: P, 
        init_f: InitF, 
        f: MapF
    ) -> ParMap<T>
    where
        T: 'static + Send,
        B: 'static + Send + Clone,
        InitF: FnOnce() -> B,
        MapF: 'static + FnMut(B, Self::Item) -> Func + Send,
        Func: 'static + FnOnce() -> T + Send,
        P: IntoParStreamParams;

    fn par_map_unordered<P, T, F, Func>(
        self, 
        config: P, 
        f: F
    ) -> ParMapUnordered<T>
    where
        T: 'static + Send,
        F: 'static + FnMut(Self::Item) -> Func + Send,
        Func: 'static + FnOnce() -> T + Send,
        P: IntoParStreamParams;

    fn par_map_init_unordered<P, T, B, InitF, MapF, Func>(
        self, 
        config: P, 
        init_f: InitF, 
        f: MapF
    ) -> ParMapUnordered<T>
    where
        T: 'static + Send,
        B: 'static + Send + Clone,
        InitF: FnOnce() -> B,
        MapF: 'static + FnMut(B, Self::Item) -> Func + Send,
        Func: 'static + FnOnce() -> T + Send,
        P: IntoParStreamParams;

    fn par_reduce<P, F, Fut>(
        self, 
        config: P, 
        reduce_fn: F
    ) -> ParReduce<Self::Item>ⓘ
Notable traits for ParReduce<T>
impl<T> Future for ParReduce<T>    type Output = Option<T>;

    where
        P: IntoParStreamParams,
        F: 'static + FnMut(Self::Item, Self::Item) -> Fut + Send + Clone,
        Fut: 'static + Future<Output = 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
        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;

    fn scatter(self, buf_size: impl Into<Option<usize>>) -> Scatter<Self::Item>;

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

    where
        F: 'static + FnMut(Self::Item) -> Fut + Send,
        Fut: 'static + Future<Output = ()> + Send,
        P: IntoParStreamParams;

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

    where
        B: 'static + Send + Clone,
        InitF: FnOnce() -> B,
        MapF: 'static + FnMut(B, Self::Item) -> Fut + Send,
        Fut: 'static + Future<Output = ()> + Send,
        P: IntoParStreamParams;

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

    where
        F: 'static + FnMut(Self::Item) -> Func + Send,
        Func: 'static + FnOnce() + Send,
        P: IntoParStreamParams;

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

    where
        B: 'static + Send + Clone,
        InitF: FnOnce() -> B,
        MapF: 'static + FnMut(B, Self::Item) -> Func + Send,
        Func: 'static + FnOnce() + Send,
        P: IntoParStreamParams;

}
Expand description

An extension trait that provides parallel processing combinators on streams.

Required methods

Maps the stream element to a different type on a spawned worker.

Maps the stream element to a different type on a parallel thread.

A combinator that consumes as many elements as it likes, and produces the next stream element.

The function f(receiver, sender) takes one or more elements by calling receiver.recv().await, It returns Some(item) if an input element is available, otherwise it returns None. Calling sender.send(item).await will produce an output element. It returns Ok(()) when success, or returns Err(item) if the output stream is closed.

use futures::prelude::*;
use par_stream::prelude::*;
use std::mem;

async fn main_async() {
    let data = vec![1, 2, -3, 4, 5, -6, 7, 8];
    let mut stream = futures::stream::iter(data).batching(|mut rx, mut tx| async move {
        let mut buffer = vec![];
        while let Some(value) = rx.recv().await {
            buffer.push(value);
            if value < 0 {
                let result = tx.send(mem::take(&mut buffer)).await;
                if result.is_err() {
                    return;
                }
            }
        }

        let _ = tx.send(mem::take(&mut buffer)).await;
    });

    assert_eq!(stream.next().await, Some(vec![1, 2, -3]));
    assert_eq!(stream.next().await, Some(vec![4, 5, -6]));
    assert_eq!(stream.next().await, Some(vec![7, 8]));
    assert!(stream.next().await.is_none());
}

The combinator maintains a collection of concurrent workers, each consuming as many elements as it likes, and produces the next stream element.

use futures::prelude::*;
use par_stream::prelude::*;
use std::mem;

async fn main_async() {
    let data = vec![1, 2, -3, 4, 5, -6, 7, 8];
    futures::stream::iter(data).batching(|mut rx, mut tx| async move {
        while let Some(value) = rx.recv().await {
            if value > 0 {
                let result = tx.send(value).await;
                if result.is_err() {
                    return;
                }
            }
        }
    });
}

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.

Each receiver maintains an internal buffer with buf_size. If one of the receiver buffer is full, the stream will halt until the blocking buffer spare the space.

use futures::prelude::*;
use par_stream::prelude::*;

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);
}

Converts to a guard that can create receivers, each receiving cloned elements from this stream.

The generated receivers can produce elements only after the guard is dropped. It ensures the receivers start receiving elements at the mean time.

Each receiver maintains an internal buffer with buf_size. If one of the receiver buffer is full, the stream will halt until the blocking buffer spare the space.

use futures::prelude::*;
use par_stream::prelude::*;

async fn main_async() {
    let mut guard = futures::stream::iter(0..).broadcast(2);
    let rx1 = guard.register();
    let rx2 = guard.register();
    guard.finish(); // drop the guard

    let (ret1, ret2): (Vec<_>, Vec<_>) =
        futures::join!(rx1.take(100).collect(), rx2.take(100).collect());
    let expect: Vec<_> = (0..100).collect();

    assert_eq!(ret1, expect);
    assert_eq!(ret2, 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::prelude::*;
use par_stream::prelude::*;

async fn main_async() {
    let doubled: Vec<_> = futures::stream::iter(0..1000)
        // doubles the values in parallel
        .par_then(None, move |value| async move { value * 2 })
        // the collected values will be ordered
        .collect()
        .await;
    let expect: Vec<_> = (0..1000).map(|value| value * 2).collect();
    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::prelude::*;
use par_stream::prelude::*;
use std::collections::HashSet;

async fn main_async() {
    let doubled: HashSet<_> = futures::stream::iter(0..1000)
        // doubles the values in parallel
        .par_then_unordered(None, move |value| {
            // the future is sent to a parallel worker
            async move { value * 2 }
        })
        // the collected values may NOT be ordered
        .collect()
        .await;
    let expect: HashSet<_> = (0..1000).map(|value| value * 2).collect();
    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::prelude::*;
use par_stream::prelude::*;

async fn main_async() {
    // the variable will be shared by parallel workers
    let doubled: Vec<_> = futures::stream::iter(0..1000)
        // doubles the values in parallel
        .par_map(None, move |value| {
            // the closure is sent to parallel worker
            move || value * 2
        })
        // the collected values may NOT be ordered
        .collect()
        .await;
    let expect: Vec<_> = (0..1000).map(|value| value * 2).collect();
    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::prelude::*;
use par_stream::prelude::*;
use std::collections::HashSet;

async fn main_async() {
    // the variable will be shared by parallel workers

    let doubled: HashSet<_> = futures::stream::iter(0..1000)
        // doubles the values in parallel
        .par_map_unordered(None, move |value| {
            // the closure is sent to parallel worker
            move || value * 2
        })
        // the collected values may NOT be ordered
        .collect()
        .await;
    let expect: HashSet<_> = (0..1000).map(|value| value * 2).collect();
    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 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::prelude::*;
use par_stream::prelude::*;

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, move |lhs, rhs| {
            // the closure is sent to parallel worker
            async move { lhs + rhs }
        })
        .await;
    assert_eq!(sum, Some((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::prelude::*;
use par_stream::prelude::*;
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: Vec<_> = futures::stream::iter(0..1000)
        // doubles the values in parallel
        .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()
        .await;
    let expect: Vec<_> = (0..1000)
        .map(|value| {
            if value % 2 == 0 {
                value / 2
            } else {
                value * 2 + 1
            }
        })
        .collect();
    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::prelude::*;
use par_stream::prelude::*;

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