recursion 0.5.4

cache-aware stack safe recursion
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
131
132
133

use std::sync::Arc;

use crate::frame::MappableFrame;

use futures::{future::BoxFuture, stream::FuturesUnordered, FutureExt, StreamExt};
use tokio::sync::{mpsc, oneshot};

pub mod compose;

pub(crate) type Frame<F, X> = <F as MappableFrame>::Frame<X>;

// mostly just used for Compact (defined over frame, needs to collapse_ref via ref frame)
pub trait MappableFrameRef: MappableFrame {
    type RefFrameToken<'a>: MappableFrame;

    fn as_ref<X>(input: &Self::Frame<X>) -> <Self::RefFrameToken<'_> as MappableFrame>::Frame<&X>;
}

pub trait AsyncMappableFrame: MappableFrame {
    // NOTE: what does having 'a here mean/imply? should 'a bound be on A/B/E?
    fn map_frame_async<'a, A, B, E>(
        input: Self::Frame<A>,
        f: impl Fn(A) -> BoxFuture<'a, Result<B, E>> + Send + Sync + 'a,
    ) -> BoxFuture<'a, Result<Self::Frame<B>, E>>
    where
        E: Send + 'a,
        A: Send + 'a,
        B: Send + 'a;
}

pub async fn expand_and_collapse_async<'a, Seed, Out, E, F>(
    seed: Seed,
    expand_frame: impl Fn(Seed) -> BoxFuture<'a, Result<Frame<F, Seed>, E>> + Send + Sync + 'a,
    collapse_frame: impl Fn(Frame<F, Out>) -> BoxFuture<'a, Result<Out, E>> + Send + Sync + 'a,
) -> Result<Out, E>
where
    E: Send + Sync + 'a,
    F: AsyncMappableFrame + 'a,
    Seed: Send + Sync + 'a,
    Out: Send + Sync + 'a,
    Frame<F, Seed>: Send + Sync + 'a,
    Frame<F, Out>: Send + Sync + 'a,
{
    let expand_frame = Arc::new(expand_frame);
    let collapse_frame = Arc::new(collapse_frame);

    let mut work_pool: FuturesUnordered<BoxFuture<'a, Result<(), E>>> = FuturesUnordered::new();
    let (work_sender, mut work_receiver) = mpsc::channel(1024); // idk what size is right here

    let (root_step, root_receiver) = Step::new(seed, work_sender);

    work_pool.push(
        root_step
            .run::<'a, F, E>(expand_frame.clone(), collapse_frame.clone())
            .boxed(),
    );

    loop {
        tokio::select! {
            // enqueue more work if we have it (bias should be here)
            Some(work) = work_receiver.recv() => work_pool.push(work.run::<F, E>(expand_frame.clone(), collapse_frame.clone()).boxed()),

            // push existing work to completion and short circuit if err
            Some(completion) = work_pool.next() => match completion{
                                Ok(_) => continue,
                                Err(e) => return Err(e),
                            },
            else => break,
        }
    }

    Ok(root_receiver.await.unwrap()) // will always terminate
}

struct Step<Seed, Out> {
    seed: Seed,
    completion_sender: oneshot::Sender<Out>,
    work_queue: mpsc::Sender<Self>,
}

// NOTE[future work]: I could build in visualization via an mpsc channel of events just like in Viz (woah, nice, etc)
impl<Seed: Sync + Send, Out: Sync + Send> Step<Seed, Out> {
    fn new(seed: Seed, work_queue: mpsc::Sender<Self>) -> (Self, oneshot::Receiver<Out>) {
        let (sender, receiver) = oneshot::channel();

        let item = Step {
            seed,
            completion_sender: sender,
            work_queue,
        };

        (item, receiver)
    }

    async fn run<'a, F: AsyncMappableFrame, E: Send + Sync + 'a>(
        self,
        expand_frame: Arc<
            dyn Fn(Seed) -> BoxFuture<'a, Result<Frame<F, Seed>, E>> + Send + Sync + 'a,
        >,
        collapse_frame: Arc<
            dyn Fn(Frame<F, Out>) -> BoxFuture<'a, Result<Out, E>> + Send + Sync + 'a,
        >,
    ) -> Result<(), E> {
        // first we expand the seed to a frame of seeds
        let frame = expand_frame(self.seed).await?;

        // enqueue work to expand each seed in the frame
        let node = F::map_frame_async(frame, |seed| {
            async {
                // for each seed in the frame, enqueue a 'Step' while hanging on to that step's receiver
                let (step, receiver) = Step::new(seed, self.work_queue.clone());
                self.work_queue.send(step).await.ok().expect("mpsc error");

                // wait on that step's receiver, at which point we have an 'Out' value to complete with
                let recvd: Out = receiver.await.expect("oneshot recv error");
                Ok(recvd)
            }
            .boxed()
        })
        .await?;

        // collapse the resulting 'Frame<F, Out>' into an 'Out'
        let collapsed = collapse_frame(node).await?;

        // pass the resulting 'Out' to this step's parent
        self.completion_sender
            .send(collapsed)
            .ok()
            .expect("oneshot send failure");

        Ok(())
    }
}