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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
use std::net::SocketAddr;
use hydroflow::util::ipv4_resolve;
use crate::flows::bp_flow::bp_flow;
use crate::flows::client_state_flow::client_state_flow;
use crate::flows::listener_flow::listener_flow;
use crate::flows::orig_flow::orig_flow;
use crate::flows::push_group_flow::push_group_flow;
use crate::flows::rep_server_flow::rep_server_flow;
use crate::flows::server_state_flow::server_state_flow;
use crate::flows::ssiv_flow::ssiv_flow;
use crate::test_data::{client100_vec, client1_vec, client2_vec};
use crate::wrappers::{bp_wrap, ssiv_wrap, tuple_wrap};
use crate::Opts;
// spawn a listener to get the output of a flow and print it on the console
async fn spawn_listener(
tuple_listener_addr: SocketAddr,
bp_listener_addr: SocketAddr,
ssiv_listener_addr: SocketAddr,
) {
let (_, tuple_listener_in, _) = hydroflow::util::bind_udp_bytes(tuple_listener_addr).await;
let (_, bp_listener_in, _) = hydroflow::util::bind_udp_bytes(bp_listener_addr).await;
let (_, ssiv_listener_in, _) = hydroflow::util::bind_udp_bytes(ssiv_listener_addr).await;
// spawn a listener thread to print out what each flow sends over the network
std::thread::spawn(move || {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let local = tokio::task::LocalSet::new();
local.block_on(&runtime, async {
let mut hf = listener_flow(tuple_listener_in, bp_listener_in, ssiv_listener_in).await;
hf.run_async().await;
});
});
}
// driver program to demonstrate the various shopping cart implementations
pub(crate) async fn run_driver(opts: Opts) {
// the address for the output; the listener will listen to this address
let out_addr = ipv4_resolve("localhost:0").unwrap();
let (out, _, _) = hydroflow::util::bind_udp_bytes(out_addr).await;
// define the shopping workload from the test data
let client1 = tuple_wrap(client1_vec().into_iter());
let client2 = tuple_wrap(client2_vec().into_iter());
let client100 = tuple_wrap(client100_vec().into_iter());
let shopping = client1.chain(client2).chain(client100);
// define a BoundedPrefix version of the shopping workload from the test data.
// each client gets a separate BoundedPrefix
let client1_bp = bp_wrap(client1_vec().into_iter()).map(|r| (1usize, r));
let client2_bp = bp_wrap(client2_vec().into_iter()).map(|r| (2usize, r));
let client100_bp = bp_wrap(client100_vec().into_iter()).map(|r| (100usize, r));
let shopping_bp = client1_bp.chain(client2_bp).chain(client100_bp);
// define an SSIV version of the shopping workload from the test data.
// each client gets a separate SSIV
let client1_ssiv = ssiv_wrap(client1_vec().into_iter()).map(|r| (1usize, r));
let client2_ssiv = ssiv_wrap(client2_vec().into_iter()).map(|r| (2usize, r));
let client100_ssiv = ssiv_wrap(client100_vec().into_iter()).map(|r| (100usize, r));
let shopping_ssiv = client1_ssiv.chain(client2_ssiv).chain(client100_ssiv);
// set up a listener to get the output of the flows and print to stdout
let tuple_listener_addr = ipv4_resolve("localhost:23470").unwrap();
let bp_listener_addr = ipv4_resolve("localhost:23471").unwrap();
let ssiv_listener_addr = ipv4_resolve("localhost:23472").unwrap();
spawn_listener(tuple_listener_addr, bp_listener_addr, ssiv_listener_addr).await;
// run the chosen dataflow
let mut hf = match opts.opt {
1 => orig_flow(shopping, tuple_listener_addr, out).await,
2 => bp_flow(shopping_bp, bp_listener_addr, out).await,
3 => ssiv_flow(shopping_ssiv, ssiv_listener_addr, out).await,
4 => push_group_flow(shopping_ssiv, ssiv_listener_addr, out).await,
opt @ (5 | 6) => {
// address for a server thread
let server_addr = ipv4_resolve("localhost:23456").unwrap();
// addresses for a client proxy thread
let client_addr = ipv4_resolve("localhost:23457").unwrap();
let client_out_addr = ipv4_resolve("localhost:23460").unwrap();
let (client_out, _, _) = hydroflow::util::bind_udp_bytes(client_out_addr).await;
// shopping input is handled by the client proxy transducer
// so the server transducer should get an empty iterator as its first argument
let empty_ssiv = std::iter::empty();
// Spawn server
std::thread::spawn(move || {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let local = tokio::task::LocalSet::new();
local.block_on(&runtime, async {
let mut hf = match opt {
5 => {
server_state_flow(
empty_ssiv,
ssiv_listener_addr,
out,
server_addr,
client_addr,
)
.await
}
6 => {
client_state_flow(
empty_ssiv,
ssiv_listener_addr,
out,
server_addr,
client_addr,
)
.await
}
_ => unreachable!(),
};
hf.run_async().await;
});
});
// The above thread spawn sets up some sockets, but we do not wait for it to do so.
// So there is a race condition where, while the thread is still spawning, we can proceed and send messages to a socket that is not bound.
// This means that the spawned thread would be waiting for a message that it has already missed and which won't get re-sent.
// There's a timeout at the bottom of this function that was presumably added to fix the issue of this test hanging.
// This sleep makes it much more likely that the thread has set up everything needed by the tiem we finish this sleep.
std::thread::sleep(std::time::Duration::from_secs(1));
// Run client proxy in this thread
match opt {
5 => {
server_state_flow(
shopping_ssiv,
client_out_addr,
client_out,
client_addr,
server_addr,
)
.await
}
6 => {
client_state_flow(
shopping_ssiv,
client_out_addr,
client_out,
client_addr,
server_addr,
)
.await
}
_ => unreachable!(),
}
}
7 => {
// define the server addresses
let addr1 = ipv4_resolve("localhost:23430").unwrap();
let addr2 = ipv4_resolve("localhost:23431").unwrap();
let addr3 = ipv4_resolve("localhost:23432").unwrap();
let server_addrs = [addr1, addr2, addr3];
// define the server addresses for gossip
let gossip_addr1 = ipv4_resolve("localhost:23440").unwrap();
let gossip_addr2 = ipv4_resolve("localhost:23441").unwrap();
let gossip_addr3 = ipv4_resolve("localhost:23442").unwrap();
let gossip_addrs = vec![gossip_addr1, gossip_addr2, gossip_addr3];
// address for a client proxy thread
let client_addr = ipv4_resolve("localhost:23457").unwrap();
let client_out_addr = ipv4_resolve("localhost:23460").unwrap();
let (client_out, _, _) = hydroflow::util::bind_udp_bytes(client_out_addr).await;
// Spawn 3 server replicas asynchronously
for pair in server_addrs.iter().zip(gossip_addrs.iter()) {
let (&addr, &gossip_addr) = pair;
let out_addr = ipv4_resolve("localhost:0").unwrap();
let (out, _, _) = hydroflow::util::bind_udp_bytes(out_addr).await;
let gossip_addrs = gossip_addrs.clone();
// shopping input is handled by the client proxy transducer
// so the server transducers should get an empty iterator as first argument
let empty_ssiv = std::iter::empty();
// Spawn server
std::thread::spawn(move || {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let local = tokio::task::LocalSet::new();
local.block_on(&runtime, async {
let mut hf = rep_server_flow(
empty_ssiv,
ssiv_listener_addr,
out,
addr,
ssiv_listener_addr,
gossip_addr,
gossip_addrs.into_iter(),
)
.await;
hf.run_async().await;
});
});
}
// The above thread spawn sets up some sockets, but we do not wait for it to do so.
// So there is a race condition where, while the thread is still spawning, we can proceed and send messages to a socket that is not bound.
// This means that the spawned thread would be waiting for a message that it has already missed and which won't get re-sent.
// There's a timeout at the bottom of this function that was presumably added to fix the issue of this test hanging.
// This sleep makes it much more likely that the thread has set up everything needed by the tiem we finish this sleep.
std::thread::sleep(std::time::Duration::from_secs(1));
// Run client proxy in this thread
rep_server_flow(
shopping_ssiv,
client_out_addr,
client_out,
client_addr,
server_addrs[0],
ipv4_resolve("localhost:23443").unwrap(),
gossip_addrs.into_iter(),
)
.await
}
_ => panic!("Invalid opt number"),
};
// optionally print the dataflow graph
#[cfg(feature = "debugging")]
if let Some(graph) = opts.graph {
let serde_graph = hf
.meta_graph()
.expect("No graph found, maybe failed to parse.");
serde_graph.open_graph(graph, opts.write_config).unwrap();
}
// Run the client for 1 second; should be long enough to get all the results
let _timeout = tokio::time::timeout(std::time::Duration::from_secs(1), hf.run_async())
.await
.unwrap_err();
}