timely 0.8.0

A low-latency data-parallel dataflow system in Rust
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

//! Barrier synchronization.

use ::communication::Allocate;
use dataflow::{InputHandle, ProbeHandle};
use worker::Worker;

/// A re-usable barrier synchronization mechanism.
pub struct Barrier<A: Allocate> {
    round: usize,
    input: InputHandle<usize, ()>,
    probe: ProbeHandle<usize>,
    worker: Worker<A>,
}

impl<A: Allocate> Barrier<A> {

    /// Allocates a new barrier.
    pub fn new(worker: &mut Worker<A>) -> Self {
        use dataflow::operators::{Input, Probe};
        let (input, probe) = worker.dataflow(|scope| {
            let (handle, stream) = scope.new_input::<()>();
            (handle, stream.probe())
        });
        Barrier { round: 0, input, probe, worker: worker.clone() }
    }

    /// Blocks until all other workers have reached this barrier.
    ///
    /// This method does *not* block dataflow execution, which continues to execute while
    /// we await the arrival of the other workers.
    pub fn wait(&mut self) {
        self.round += 1;
        self.input.advance_to(self.round);
        while self.probe.less_than(self.input.time()) {
            self.worker.step();
        }
    }
}