Deploying to gh-pages from @ 5323fd6 🚀

Author: frankmcsherry

chapter_2/chapter_2_4.html

@@ -189,12 +189,10 @@ <h1 id="creating-operators"><a class="header" href="#creating-operators">Creatin
             .to_stream(scope)
             .unary(Pipeline, "increment", |capability, info| {
 
-                let mut vector = Vec::new();
                 move |input, output| {
                     while let Some((time, data)) = input.next() {
-                        data.swap(&amp;mut vector);
                         let mut session = output.session(&amp;time);
-                        for datum in vector.drain(..) {
+                        for datum in data.drain(..) {
                             session.give(datum + 1);
                         }
                     }
@@ -277,13 +275,11 @@ <h3 id="stateful-operators"><a class="header" href="#stateful-operators">Statefu
             .unary(Pipeline, "increment", |capability, info| {
 
                 let mut maximum = 0;    // define this here; use in the closure
-                let mut vector = Vec::new();
 
                 move |input, output| {
                     while let Some((time, data)) = input.next() {
-                        data.swap(&amp;mut vector);
                         let mut session = output.session(&amp;time);
-                        for datum in vector.drain(..) {
+                        for datum in data.drain(..) {
                             if datum &gt; maximum {
                                 session.give(datum + 1);
                                 maximum = datum;
@@ -325,13 +321,13 @@ <h3 id="frontiered-operators"><a class="header" href="#frontiered-operators">Fro
                 while let Some((time, data)) = input1.next() {
                     stash.entry(time.time().clone())
                          .or_insert(Vec::new())
-                         .push(data.replace(Vec::new()));
+                         .push(std::mem::take(data));
                     notificator.notify_at(time.retain());
                 }
                 while let Some((time, data)) = input2.next() {
                     stash.entry(time.time().clone())
                          .or_insert(Vec::new())
-                         .push(data.replace(Vec::new()));
+                         .push(std::mem::take(data));
                     notificator.notify_at(time.retain());
                 }
 
@@ -371,12 +367,12 @@ <h3 id="frontiered-operators"><a class="header" href="#frontiered-operators">Fro
                 while let Some((time, data)) = input1.next() {
                     stash.entry(time.retain())
                          .or_insert(Vec::new())
-                         .push(data.replace(Vec::new()));
+                         .push(std::mem::take(data));
                 }
                 while let Some((time, data)) = input2.next() {
                     stash.entry(time.retain())
                          .or_insert(Vec::new())
-                         .push(data.replace(Vec::new()));
+                         .push(std::mem::take(data));
                 }
 
                 // consider sending everything in `stash`.

chapter_2/chapter_2_5.html

@@ -343,7 +343,7 @@ <h2 id="maintaining-word-counts"><a class="header" href="#maintaining-word-count
                         while let Some((time, data)) = input.next() {
                             queues.entry(time.retain())
                                   .or_insert(Vec::new())
-                                  .extend(data.replace(Vec::new()));
+                                  .extend(std::mem::take(data));
                         }
 
                         // enable each stashed time if ready.
@@ -412,7 +412,7 @@ <h2 id="maintaining-word-counts"><a class="header" href="#maintaining-word-count
         while let Some((time, data)) = input.next() {
             queues.entry(time.retain())
                   .or_insert(Vec::new())
-                  .extend(data.replace(Vec::new()));
+                  .extend(std::mem::take(data));
         }</code></pre>
 <p>The <code>input</code> handle has a <code>next</code> method, and it optionally returns a pair of <code>time</code> and <code>data</code>, representing a timely dataflow timestamp and a hunk of data bearing that timestamp, respectively. Our plan is to iterate through all available input (the <code>next()</code> method doesn't block, it just returns <code>None</code> when it runs out of data), accepting it from the timely dataflow system and moving it into our <code>queue</code> hash map.</p>
 <p>Why do we do this? Because this is a streaming system, we could be getting data out of order. Our goal is to update the counts in time order, and to do this we'll need to enqueue what we get until we also get word that the associated <code>time</code> is complete. That happens in the next few hunks of code</p>

chapter_4/chapter_4_3.html

@@ -234,16 +234,13 @@ <h1 id="flow-control"><a class="header" href="#flow-control">Flow Control</a></h
             // Buffer records until all prior timestamps have completed.
             .binary_frontier(&amp;cycle, Pipeline, Pipeline, "Buffer", move |capability, info| {
 
-                let mut vector = Vec::new();
-
                 move |input1, input2, output| {
 
                     // Stash received data.
                     input1.for_each(|time, data| {
-                        data.swap(&amp;mut vector);
                         stash.entry(time.retain())
                              .or_insert(Vec::new())
-                             .extend(vector.drain(..));
+                             .extend(data.drain(..));
                     });
 
                     // Consider sending stashed data.

print.html

@@ -827,12 +827,10 @@ <h2 id="other-operators"><a class="header" href="#other-operators">Other operato
             .to_stream(scope)
             .unary(Pipeline, "increment", |capability, info| {
 
-                let mut vector = Vec::new();
                 move |input, output| {
                     while let Some((time, data)) = input.next() {
-                        data.swap(&amp;mut vector);
                         let mut session = output.session(&amp;time);
-                        for datum in vector.drain(..) {
+                        for datum in data.drain(..) {
                             session.give(datum + 1);
                         }
                     }
@@ -915,13 +913,11 @@ <h3 id="stateful-operators"><a class="header" href="#stateful-operators">Statefu
             .unary(Pipeline, "increment", |capability, info| {
 
                 let mut maximum = 0;    // define this here; use in the closure
-                let mut vector = Vec::new();
 
                 move |input, output| {
                     while let Some((time, data)) = input.next() {
-                        data.swap(&amp;mut vector);
                         let mut session = output.session(&amp;time);
-                        for datum in vector.drain(..) {
+                        for datum in data.drain(..) {
                             if datum &gt; maximum {
                                 session.give(datum + 1);
                                 maximum = datum;
@@ -963,13 +959,13 @@ <h3 id="frontiered-operators"><a class="header" href="#frontiered-operators">Fro
                 while let Some((time, data)) = input1.next() {
                     stash.entry(time.time().clone())
                          .or_insert(Vec::new())
-                         .push(data.replace(Vec::new()));
+                         .push(std::mem::take(data));
                     notificator.notify_at(time.retain());
                 }
                 while let Some((time, data)) = input2.next() {
                     stash.entry(time.time().clone())
                          .or_insert(Vec::new())
-                         .push(data.replace(Vec::new()));
+                         .push(std::mem::take(data));
                     notificator.notify_at(time.retain());
                 }
 
@@ -1009,12 +1005,12 @@ <h3 id="frontiered-operators"><a class="header" href="#frontiered-operators">Fro
                 while let Some((time, data)) = input1.next() {
                     stash.entry(time.retain())
                          .or_insert(Vec::new())
-                         .push(data.replace(Vec::new()));
+                         .push(std::mem::take(data));
                 }
                 while let Some((time, data)) = input2.next() {
                     stash.entry(time.retain())
                          .or_insert(Vec::new())
-                         .push(data.replace(Vec::new()));
+                         .push(std::mem::take(data));
                 }
 
                 // consider sending everything in `stash`.
@@ -1206,7 +1202,7 @@ <h2 id="maintaining-word-counts"><a class="header" href="#maintaining-word-count
                         while let Some((time, data)) = input.next() {
                             queues.entry(time.retain())
                                   .or_insert(Vec::new())
-                                  .extend(data.replace(Vec::new()));
+                                  .extend(std::mem::take(data));
                         }
 
                         // enable each stashed time if ready.
@@ -1275,7 +1271,7 @@ <h2 id="maintaining-word-counts"><a class="header" href="#maintaining-word-count
         while let Some((time, data)) = input.next() {
             queues.entry(time.retain())
                   .or_insert(Vec::new())
-                  .extend(data.replace(Vec::new()));
+                  .extend(std::mem::take(data));
         }</code></pre>
 <p>The <code>input</code> handle has a <code>next</code> method, and it optionally returns a pair of <code>time</code> and <code>data</code>, representing a timely dataflow timestamp and a hunk of data bearing that timestamp, respectively. Our plan is to iterate through all available input (the <code>next()</code> method doesn't block, it just returns <code>None</code> when it runs out of data), accepting it from the timely dataflow system and moving it into our <code>queue</code> hash map.</p>
 <p>Why do we do this? Because this is a streaming system, we could be getting data out of order. Our goal is to update the counts in time order, and to do this we'll need to enqueue what we get until we also get word that the associated <code>time</code> is complete. That happens in the next few hunks of code</p>
@@ -1642,16 +1638,13 @@ <h2 id="scopes-1"><a class="header" href="#scopes-1">Scopes</a></h2>
             // Buffer records until all prior timestamps have completed.
             .binary_frontier(&amp;cycle, Pipeline, Pipeline, "Buffer", move |capability, info| {
 
-                let mut vector = Vec::new();
-
                 move |input1, input2, output| {
 
                     // Stash received data.
                     input1.for_each(|time, data| {
-                        data.swap(&amp;mut vector);
                         stash.entry(time.retain())
                              .or_insert(Vec::new())
-                             .extend(vector.drain(..));
+                             .extend(data.drain(..));
                     });
 
                     // Consider sending stashed data.