Skip to content

Commit 79b5e53

Browse files
frankmcsherryfrankmcsherry
committed
Deploying to gh-pages from @ 729f75b 🚀
1 parent 4402132 commit 79b5e53

File tree

6 files changed

+64
-64
lines changed

6 files changed

+64
-64
lines changed

‎chapter_2/chapter_2_4.html‎

Lines changed: 22 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -190,12 +190,12 @@ <h1 id="creating-operators"><a class="header" href="#creating-operators">Creatin
190190
.unary(Pipeline, "increment", |capability, info| {
191191

192192
move |input, output| {
193-
while let Some((time, data)) = input.next() {
193+
input.for_each_time(|time, data| {
194194
let mut session = output.session(&amp;time);
195-
for datum in data.drain(..) {
195+
for datum in data.flat_map(|d| d.drain(..)) {
196196
session.give(datum + 1);
197197
}
198-
}
198+
});
199199
}
200200
})
201201
.container::&lt;Vec&lt;_&gt;&gt;();
@@ -277,15 +277,15 @@ <h3 id="stateful-operators"><a class="header" href="#stateful-operators">Statefu
277277
let mut maximum = 0; // define this here; use in the closure
278278

279279
move |input, output| {
280-
while let Some((time, data)) = input.next() {
280+
input.for_each_time(|time, data| {
281281
let mut session = output.session(&amp;time);
282-
for datum in data.drain(..) {
282+
for datum in data.flat_map(|d| d.drain(..)) {
283283
if datum &gt; maximum {
284284
session.give(datum + 1);
285285
maximum = datum;
286286
}
287287
}
288-
}
288+
});
289289
}
290290
})
291291
.container::&lt;Vec&lt;_&gt;&gt;();
@@ -317,21 +317,21 @@ <h3 id="frontiered-operators"><a class="header" href="#frontiered-operators">Fro
317317
let mut notificator = FrontierNotificator::default();
318318
let mut stash = HashMap::new();
319319

320-
move |input1, input2, output| {
321-
while let Some((time, data)) = input1.next() {
320+
move |(input1, frontier1), (input2, frontier2), output| {
321+
input1.for_each_time(|time, data| {
322322
stash.entry(time.time().clone())
323323
.or_insert(Vec::new())
324-
.push(std::mem::take(data));
324+
.extend(data.map(std::mem::take));
325325
notificator.notify_at(time.retain());
326-
}
327-
while let Some((time, data)) = input2.next() {
326+
});
327+
input2.for_each_time(|time, data| {
328328
stash.entry(time.time().clone())
329329
.or_insert(Vec::new())
330-
.push(std::mem::take(data));
330+
.extend(data.map(std::mem::take));
331331
notificator.notify_at(time.retain());
332-
}
332+
});
333333

334-
notificator.for_each(&amp;[input1.frontier(), input2.frontier()], |time, notificator| {
334+
notificator.for_each(&amp;[frontier1, frontier2], |time, notificator| {
335335
let mut session = output.session(&amp;time);
336336
if let Some(list) = stash.remove(time.time()) {
337337
for mut vector in list.into_iter() {
@@ -362,21 +362,21 @@ <h3 id="frontiered-operators"><a class="header" href="#frontiered-operators">Fro
362362

363363
let mut stash = HashMap::new();
364364

365-
move |input1, input2, output| {
365+
move |(input1, frontier1), (input2, frontier2), output| {
366366

367-
while let Some((time, data)) = input1.next() {
367+
input1.for_each_time(|time, data| {
368368
stash.entry(time.retain())
369369
.or_insert(Vec::new())
370-
.push(std::mem::take(data));
371-
}
372-
while let Some((time, data)) = input2.next() {
370+
.extend(data.map(std::mem::take));
371+
});
372+
input2.for_each_time(|time, data| {
373373
stash.entry(time.retain())
374374
.or_insert(Vec::new())
375-
.push(std::mem::take(data));
376-
}
375+
.extend(data.map(std::mem::take));
376+
});
377377

378378
// consider sending everything in `stash`.
379-
let frontiers = &amp;[input1.frontier(), input2.frontier()];
379+
let frontiers = &amp;[frontier1, frontier2];
380380
for (time, list) in stash.iter_mut() {
381381
// if neither input can produce data at `time`, ship `list`.
382382
if frontiers.iter().all(|f| !f.less_equal(time.time())) {

‎chapter_2/chapter_2_5.html‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -337,18 +337,18 @@ <h2 id="maintaining-word-counts"><a class="header" href="#maintaining-word-count
337337
let mut counts = HashMap::new();
338338
let mut buffer = Vec::new();
339339

340-
move |input, output| {
340+
move |(input, frontier), output| {
341341

342342
// for each input batch, stash it at `time`.
343-
while let Some((time, data)) = input.next() {
343+
input.for_each_time(|time, data| {
344344
queues.entry(time.retain())
345345
.or_insert(Vec::new())
346-
.extend(std::mem::take(data));
347-
}
346+
.extend(data.flat_map(|d| d.drain(..)));
347+
});
348348

349349
// enable each stashed time if ready.
350350
for (time, vals) in queues.iter_mut() {
351-
if !input.frontier().less_equal(time.time()) {
351+
if !frontier.less_equal(time.time()) {
352352
let vals = std::mem::replace(vals, Vec::new());
353353
buffer.push((time.clone(), vals));
354354
}

‎chapter_4/chapter_4_3.html‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -234,20 +234,20 @@ <h1 id="flow-control"><a class="header" href="#flow-control">Flow Control</a></h
234234
// Buffer records until all prior timestamps have completed.
235235
.binary_frontier(&amp;cycle, Pipeline, Pipeline, "Buffer", move |capability, info| {
236236

237-
move |input1, input2, output| {
237+
move |(input1, frontier1), (input2, frontier2), output| {
238238

239239
// Stash received data.
240-
input1.for_each(|time, data| {
240+
input1.for_each_time(|time, data| {
241241
stash.entry(time.retain())
242242
.or_insert(Vec::new())
243-
.extend(data.drain(..));
243+
.extend(data.flat_map(|d| d.drain(..)));
244244
});
245245

246246
// Consider sending stashed data.
247247
for (time, data) in stash.iter_mut() {
248248
// Only send data once the probe is not less than the time.
249249
// That is, once we have finished all strictly prior work.
250-
if !input2.frontier().less_than(time.time()) {
250+
if !frontier2.less_than(time.time()) {
251251
output.session(&amp;time).give_iterator(data.drain(..));
252252
}
253253
}

‎print.html‎

Lines changed: 31 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -831,12 +831,12 @@ <h2 id="other-operators"><a class="header" href="#other-operators">Other operato
831831
.unary(Pipeline, "increment", |capability, info| {
832832

833833
move |input, output| {
834-
while let Some((time, data)) = input.next() {
834+
input.for_each_time(|time, data| {
835835
let mut session = output.session(&amp;time);
836-
for datum in data.drain(..) {
836+
for datum in data.flat_map(|d| d.drain(..)) {
837837
session.give(datum + 1);
838838
}
839-
}
839+
});
840840
}
841841
})
842842
.container::&lt;Vec&lt;_&gt;&gt;();
@@ -918,15 +918,15 @@ <h3 id="stateful-operators"><a class="header" href="#stateful-operators">Statefu
918918
let mut maximum = 0; // define this here; use in the closure
919919

920920
move |input, output| {
921-
while let Some((time, data)) = input.next() {
921+
input.for_each_time(|time, data| {
922922
let mut session = output.session(&amp;time);
923-
for datum in data.drain(..) {
923+
for datum in data.flat_map(|d| d.drain(..)) {
924924
if datum &gt; maximum {
925925
session.give(datum + 1);
926926
maximum = datum;
927927
}
928928
}
929-
}
929+
});
930930
}
931931
})
932932
.container::&lt;Vec&lt;_&gt;&gt;();
@@ -958,21 +958,21 @@ <h3 id="frontiered-operators"><a class="header" href="#frontiered-operators">Fro
958958
let mut notificator = FrontierNotificator::default();
959959
let mut stash = HashMap::new();
960960

961-
move |input1, input2, output| {
962-
while let Some((time, data)) = input1.next() {
961+
move |(input1, frontier1), (input2, frontier2), output| {
962+
input1.for_each_time(|time, data| {
963963
stash.entry(time.time().clone())
964964
.or_insert(Vec::new())
965-
.push(std::mem::take(data));
965+
.extend(data.map(std::mem::take));
966966
notificator.notify_at(time.retain());
967-
}
968-
while let Some((time, data)) = input2.next() {
967+
});
968+
input2.for_each_time(|time, data| {
969969
stash.entry(time.time().clone())
970970
.or_insert(Vec::new())
971-
.push(std::mem::take(data));
971+
.extend(data.map(std::mem::take));
972972
notificator.notify_at(time.retain());
973-
}
973+
});
974974

975-
notificator.for_each(&amp;[input1.frontier(), input2.frontier()], |time, notificator| {
975+
notificator.for_each(&amp;[frontier1, frontier2], |time, notificator| {
976976
let mut session = output.session(&amp;time);
977977
if let Some(list) = stash.remove(time.time()) {
978978
for mut vector in list.into_iter() {
@@ -1003,21 +1003,21 @@ <h3 id="frontiered-operators"><a class="header" href="#frontiered-operators">Fro
10031003

10041004
let mut stash = HashMap::new();
10051005

1006-
move |input1, input2, output| {
1006+
move |(input1, frontier1), (input2, frontier2), output| {
10071007

1008-
while let Some((time, data)) = input1.next() {
1008+
input1.for_each_time(|time, data| {
10091009
stash.entry(time.retain())
10101010
.or_insert(Vec::new())
1011-
.push(std::mem::take(data));
1012-
}
1013-
while let Some((time, data)) = input2.next() {
1011+
.extend(data.map(std::mem::take));
1012+
});
1013+
input2.for_each_time(|time, data| {
10141014
stash.entry(time.retain())
10151015
.or_insert(Vec::new())
1016-
.push(std::mem::take(data));
1017-
}
1016+
.extend(data.map(std::mem::take));
1017+
});
10181018

10191019
// consider sending everything in `stash`.
1020-
let frontiers = &amp;[input1.frontier(), input2.frontier()];
1020+
let frontiers = &amp;[frontier1, frontier2];
10211021
for (time, list) in stash.iter_mut() {
10221022
// if neither input can produce data at `time`, ship `list`.
10231023
if frontiers.iter().all(|f| !f.less_equal(time.time())) {
@@ -1199,18 +1199,18 @@ <h2 id="maintaining-word-counts"><a class="header" href="#maintaining-word-count
11991199
let mut counts = HashMap::new();
12001200
let mut buffer = Vec::new();
12011201

1202-
move |input, output| {
1202+
move |(input, frontier), output| {
12031203

12041204
// for each input batch, stash it at `time`.
1205-
while let Some((time, data)) = input.next() {
1205+
input.for_each_time(|time, data| {
12061206
queues.entry(time.retain())
12071207
.or_insert(Vec::new())
1208-
.extend(std::mem::take(data));
1209-
}
1208+
.extend(data.flat_map(|d| d.drain(..)));
1209+
});
12101210

12111211
// enable each stashed time if ready.
12121212
for (time, vals) in queues.iter_mut() {
1213-
if !input.frontier().less_equal(time.time()) {
1213+
if !frontier.less_equal(time.time()) {
12141214
let vals = std::mem::replace(vals, Vec::new());
12151215
buffer.push((time.clone(), vals));
12161216
}
@@ -1641,20 +1641,20 @@ <h2 id="scopes-1"><a class="header" href="#scopes-1">Scopes</a></h2>
16411641
// Buffer records until all prior timestamps have completed.
16421642
.binary_frontier(&amp;cycle, Pipeline, Pipeline, "Buffer", move |capability, info| {
16431643

1644-
move |input1, input2, output| {
1644+
move |(input1, frontier1), (input2, frontier2), output| {
16451645

16461646
// Stash received data.
1647-
input1.for_each(|time, data| {
1647+
input1.for_each_time(|time, data| {
16481648
stash.entry(time.retain())
16491649
.or_insert(Vec::new())
1650-
.extend(data.drain(..));
1650+
.extend(data.flat_map(|d| d.drain(..)));
16511651
});
16521652

16531653
// Consider sending stashed data.
16541654
for (time, data) in stash.iter_mut() {
16551655
// Only send data once the probe is not less than the time.
16561656
// That is, once we have finished all strictly prior work.
1657-
if !input2.frontier().less_than(time.time()) {
1657+
if !frontier2.less_than(time.time()) {
16581658
output.session(&amp;time).give_iterator(data.drain(..));
16591659
}
16601660
}

‎searchindex.js‎

Lines changed: 1 addition & 1 deletion
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎searchindex.json‎

Lines changed: 1 addition & 1 deletion
Large diffs are not rendered by default.

0 commit comments

Comments
 (0)