moirai-iter 0.6.1

Parallel and async iterator combinators for Moirai concurrency library
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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

use super::super::{Consumer, ParallelIterator};
use std::ops::ControlFlow;

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
struct ExponentialBlockPolicy;

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
struct UniformBlockPolicy;

#[repr(transparent)]
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct BlockSize(usize);

impl BlockSize {
    fn new(value: usize) -> Self {
        assert!(value != 0, "block size must be non-zero");
        Self(value)
    }
}

/// Indexed source adapter for exponential logical block scheduling.
pub struct ExponentialBlocks<I> {
    base: I,
    policy: ExponentialBlockPolicy,
}

impl<I> ExponentialBlocks<I> {
    pub(in crate::parallel) fn new(base: I) -> Self {
        Self {
            base,
            policy: ExponentialBlockPolicy,
        }
    }
}

impl<I> ParallelIterator for ExponentialBlocks<I>
where
    I: ParallelIterator,
    I::Item: Sync + 'static,
{
    type Item = I::Item;

    fn seq_items(self) -> Vec<Self::Item> {
        let _policy = self.policy;
        self.base.seq_items()
    }

    fn seq_try_fold<Acc, B, FoldFn>(self, init: Acc, fold_fn: FoldFn) -> ControlFlow<B, Acc>
    where
        FoldFn: FnMut(Acc, Self::Item) -> ControlFlow<B, Acc>,
    {
        let _policy = self.policy;
        self.base.seq_try_fold(init, fold_fn)
    }

    fn drive<C, R>(self, consumer: C) -> R
    where
        C: Consumer<Self::Item, Result = R> + Send + Sync,
        R: Send,
    {
        // The block policy selects a logical scheduling shape and leaves the
        // item stream identical to the base's, so this adapter drives the base
        // directly rather than collecting it. Collecting cost the source's
        // shards to express an identity.
        let _policy = self.policy;
        self.base.drive(consumer)
    }
}

/// Indexed source adapter for uniform logical block scheduling.
pub struct UniformBlocks<I> {
    base: I,
    block_size: BlockSize,
    policy: UniformBlockPolicy,
}

impl<I> UniformBlocks<I> {
    pub(in crate::parallel) fn new(base: I, block_size: usize) -> Self {
        Self {
            base,
            block_size: BlockSize::new(block_size),
            policy: UniformBlockPolicy,
        }
    }
}

impl<I> ParallelIterator for UniformBlocks<I>
where
    I: ParallelIterator,
    I::Item: Sync + 'static,
{
    type Item = I::Item;

    fn seq_items(self) -> Vec<Self::Item> {
        let _block_size = self.block_size;
        let _policy = self.policy;
        self.base.seq_items()
    }

    fn seq_try_fold<Acc, B, FoldFn>(self, init: Acc, fold_fn: FoldFn) -> ControlFlow<B, Acc>
    where
        FoldFn: FnMut(Acc, Self::Item) -> ControlFlow<B, Acc>,
    {
        let _block_size = self.block_size;
        let _policy = self.policy;
        self.base.seq_try_fold(init, fold_fn)
    }

    fn drive<C, R>(self, consumer: C) -> R
    where
        C: Consumer<Self::Item, Result = R> + Send + Sync,
        R: Send,
    {
        // Identity item stream, as on [`ExponentialBlocks`].
        let _block_size = self.block_size;
        let _policy = self.policy;
        self.base.drive(consumer)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn block_policy_markers_are_zero_sized() {
        assert_eq!(std::mem::size_of::<ExponentialBlockPolicy>(), 0);
        assert_eq!(std::mem::size_of::<UniformBlockPolicy>(), 0);
    }
}