stream-future 0.5.0

Implement an async iterator with return value.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115

#![feature(async_closure)]
#![feature(coroutines)]
#![feature(iter_from_coroutine)]
#![feature(test)]

extern crate test;
use std::future::Future;
use stream_future::*;
use test::{black_box, Bencher};
use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver};
use tokio_stream::{wrappers::UnboundedReceiverStream, StreamExt};

fn async_bench<F: Future>(b: &mut Bencher, mut f: impl FnMut() -> F) {
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    b.iter(|| runtime.block_on(f()))
}

const ITER_MAX: i32 = 10000;

#[bench]
fn this(b: &mut Bencher) {
    #[stream(i32)]
    async fn foo() {
        for i in 0..ITER_MAX {
            yield i;
        }
    }

    async_bench(b, async move || {
        let s = foo();
        tokio::pin!(s);
        while let Some(v) = s.next().await {
            black_box(v);
        }
    })
}

#[bench]
fn async_stream(b: &mut Bencher) {
    fn foo() -> impl Stream<Item = i32> {
        async_stream::stream! {
            for i in 0..ITER_MAX {
                yield i;
            }
        }
    }

    async_bench(b, async move || {
        let s = foo();
        tokio::pin!(s);
        while let Some(v) = s.next().await {
            black_box(v);
        }
    })
}

#[bench]
fn mpsc_channel(b: &mut Bencher) {
    fn foo() -> UnboundedReceiver<i32> {
        let (tx, rx) = unbounded_channel();
        tokio::spawn(async move {
            for i in 0..ITER_MAX {
                tx.send(i).unwrap();
            }
        });
        rx
    }

    async_bench(b, async move || {
        let s = UnboundedReceiverStream::new(foo());
        tokio::pin!(s);
        while let Some(v) = s.next().await {
            black_box(v);
        }
    })
}

#[bench]
fn iter_stream(b: &mut Bencher) {
    async_bench(b, async move || {
        let mut s = tokio_stream::iter(0..ITER_MAX);
        while let Some(v) = s.next().await {
            black_box(v);
        }
    })
}

#[bench]
fn iter(b: &mut Bencher) {
    async_bench(b, async move || {
        for i in 0..ITER_MAX {
            black_box(i);
        }
    })
}

#[bench]
fn coroutine(b: &mut Bencher) {
    async_bench(b, async move || {
        let s = std::iter::from_coroutine(
            #[coroutine]
            || {
                for i in 0..ITER_MAX {
                    yield i;
                }
            },
        );
        for v in s {
            black_box(v);
        }
    })
}