differential-dataflow 0.7.0

An incremental data-parallel dataflow platform
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

extern crate timely;
extern crate differential_dataflow;

use std::sync::{Arc, Mutex};
use std::net::TcpListener;
use std::time::Duration;

use timely::dataflow::operators::{Map, Inspect};
use timely::progress::nested::product::Product;
use timely::progress::timestamp::RootTimestamp;
use timely::logging::TimelyEvent;
use timely::communication::logging::{CommunicationEvent, CommunicationSetup};
use timely::dataflow::operators::Filter;
use timely::dataflow::operators::capture::{EventReader, Replay};

use differential_dataflow::AsCollection;
use differential_dataflow::operators::{Consolidate, Join};
// use differential_dataflow::logging::DifferentialEvent;

fn main() {

    let mut args = ::std::env::args();
    args.next().unwrap();

    let work_peers: usize = args.next().expect("Must provide number of source peers").parse().expect("Source peers must be an unsigned integer");
    let comm_peers: usize = args.next().expect("Must provide number of source peers").parse().expect("Comms peers must be an unsigned integer");
    let granularity: u64  = args.next().expect("Must provide report granularity in seconds").parse().expect("Granularity must be an unsigned integer");

	println!("starting with work peers: {}, comm peers: {}, granularity: {}", work_peers, comm_peers, granularity);

    let t_listener = TcpListener::bind("127.0.0.1:8000").unwrap();
    let d_listener = TcpListener::bind("127.0.0.1:9000").unwrap();
    let t_sockets =
    Arc::new(Mutex::new((0..work_peers).map(|_| {
            let socket = t_listener.incoming().next().unwrap().unwrap();
            socket.set_nonblocking(true).expect("failed to set nonblocking");
            Some(socket)
        }).collect::<Vec<_>>()));
    let d_sockets =
    Arc::new(Mutex::new((0..comm_peers).map(|_| {
            let socket = d_listener.incoming().next().unwrap().unwrap();
            socket.set_nonblocking(true).expect("failed to set nonblocking");
            Some(socket)
        }).collect::<Vec<_>>()));

    timely::execute_from_args(std::env::args(), move |worker| {

        let index = worker.index();
        let peers = worker.peers();

        let t_streams =
        t_sockets
            .lock()
            .unwrap()
            .iter_mut()
            .enumerate()
            .filter(|(i, _)| *i % peers == index)
            .map(move |(_, s)| s.take().unwrap())
            .map(|r| EventReader::<Product<RootTimestamp, Duration>, (Duration, usize, TimelyEvent),_>::new(r))
            .collect::<Vec<_>>();

        let d_streams =
        d_sockets
            .lock()
            .unwrap()
            .iter_mut()
            .enumerate()
            .filter(|(i, _)| *i % peers == index)
            .map(move |(_, s)| s.take().unwrap())
            .map(|r| EventReader::<Product<RootTimestamp, Duration>, (Duration, CommunicationSetup, CommunicationEvent),_>::new(r))
            .collect::<Vec<_>>();

        worker.dataflow::<_,_,_>(|scope| {

            let t_events = t_streams.replay_into(scope);
            let d_events = d_streams.replay_into(scope);

            // let operates =
            t_events
                .filter(|x| x.1 == 0)
                .flat_map(move |(ts, _worker, datum)| {
                    let ts = Duration::from_secs((ts.as_secs()/granularity + 1) * granularity);
                    if let TimelyEvent::Channels(event) = datum {
                        Some((event, RootTimestamp::new(ts), 1))
                    }
                    else { None }
                })
//                .inspect(|x| println!("???\t{:?}", x))
                .as_collection()
                // .consolidate()
                .inspect(|x| println!("CHANNEL\t{:?}", x));
                ;

            t_events
                .filter(|x| x.1 == 0)
                .flat_map(move |(ts, _worker, datum)| {
                    let ts = Duration::from_secs((ts.as_secs()/granularity + 1) * granularity);
                    if let TimelyEvent::CommChannels(event) = datum {
                        Some((event, RootTimestamp::new(ts), 1))
                    }
                    else { None }
                })
                .as_collection()
                // .consolidate()
                .inspect(|x| println!("COMM_CHANNEL\t{:?}", x));
                ;

            // let memory =
            d_events
                // .inspect(|x| println!("MESSAGE\t{:?}", x))
                .flat_map(move |(ts, _worker, datum)| {
                    let ts = Duration::from_secs((ts.as_secs()/granularity + 1) * granularity);
                    if let CommunicationEvent::Message(x) = datum {
                        Some((x.header.channel, RootTimestamp::new(ts), x.header.length as isize))
                    }
                    else { None }
                })
                .as_collection()
                .consolidate()
                .inspect(|x| println!("MESSAGE\t{:?}", x));
                ;

            // operates
            //     .inspect(|x| println!("OPERATES: {:?}", x))
            //     .semijoin(&memory)
            //     .inspect(|x| println!("{:?}", x));

        });

    }).unwrap(); // asserts error-free execution
}