public-collect/books-futures-explained
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

fixed minor differences between 'compiled' generators and the example used. Added bonus section to prove it works

Browse Source
This commit is contained in:
Carl Fredrik Samson
2020-02-05 22:59:50 +01:00
parent 53529fa769
commit f9d3530949
9 changed files with 253 additions and 280 deletions
Download Patch File Download Diff File Expand all files Collapse all files

241
book/print.html
View File

@@ -463,7 +463,7 @@ async/await as keywords (it can even be done using a macro).</li>
<li>No need for context switching and saving/restoring CPU state</li>
<li>No need to handle dynamic stack allocation</li>
<li>Very memory efficient</li>
<li>Allowed for borrows across suspension points</li>
<li>Allows us to borrow across suspension points</li>
</ol>
<p>The last point is in contrast to <code>Futures 1.0</code>. With async/await we can do this:</p>
<pre><code class="language-rust ignore">async fn myfn() {
@@ -479,22 +479,27 @@ step require. That means that adding steps to a chain of computations might not
require any increased memory at all.</p>
<h2><a class="header" href="#how-generators-work" id="how-generators-work">How generators work</a></h2>
<p>In Nightly Rust today you can use the <code>yield</code> keyword. Basically using this
keyword in a closure, converts it to a generator. A closure looking like this
(I'm going to use the terminology that's currently in Rust):</p>
<pre><code class="language-rust noplaypen ignore">let a = 4;
let b = move || {
keyword in a closure, converts it to a generator. A closure could look like this
before we had a concept of <code>Pin</code>:</p>
<pre><code class="language-rust noplaypen ignore">#![feature(generators, generator_trait)]
use std::ops::{Generator, GeneratorState};
fn main() {
let a: i32 = 4;
let mut gen = move || {
println!(&quot;Hello&quot;);
yield a * 2;
println!(&quot;world!&quot;);
};
if let GeneratorState::Yielded(n) = gen.resume() {
if let GeneratorState::Yielded(n) = gen.resume() {
println!(&quot;Got value {}&quot;, n);
}
if let GeneratorState::Complete(()) = gen.resume() {
if let GeneratorState::Complete(()) = gen.resume() {
()
};
};
}
</code></pre>
<p>Early on, before there was a consensus about the design of <code>Pin</code>, this
compiled to something looking similar to this:</p>
@@ -576,17 +581,15 @@ you'll also know the basics of how <code>await</code> works. It's very similar.<
<p>We could forbid that, but <strong>one of the major design goals for the async/await syntax has been
to allow this</strong>. These kinds of borrows were not possible using <code>Futures 1.0</code> so we can't let this
limitation just slip and call it a day yet.</p>
<p>Instead of discussing it in theory, let's look at some code. </p>
<p>Instead of discussing it in theory, let's look at some code.</p>
<blockquote>
<p>We'll use the optimized version of the state machines which is used in Rust today. For a more
in deapth explanation see <a href="https://tmandry.gitlab.io/blog/posts/optimizing-await-1/">Tyler Mandry's excellent article: How Rust optimizes async/await</a></p>
in depth explanation see <a href="https://tmandry.gitlab.io/blog/posts/optimizing-await-1/">Tyler Mandry's excellent article: How Rust optimizes async/await</a></p>
</blockquote>
<pre><code class="language-rust noplaypen ignore">let a = 4;
let b = move || {
let to_borrow = String::new(&quot;Hello&quot;);
<pre><code class="language-rust noplaypen ignore">let mut gen = move || {
let to_borrow = String::from(&quot;Hello&quot;);
let borrowed = &amp;to_borrow;
println!(&quot;{}&quot;, borrowed);
yield a * 2;
yield borrowed.len();
println!(&quot;{} world!&quot;, borrowed);
};
</code></pre>
@@ -632,8 +635,10 @@ impl Generator for GeneratorA {
GeneratorA::Enter =&gt; {
let to_borrow = String::from(&quot;Hello&quot;);
let borrowed = &amp;to_borrow;
let res = borrowed.len();
*self = GeneratorA::Yield1 {to_borrow, borrowed};
GeneratorState::Yielded(borrowed.len())
GeneratorState::Yielded(res)
}
GeneratorA::Yield1 {to_borrow, borrowed} =&gt; {
@@ -742,7 +747,7 @@ Rust. This is a big problem!</p>
<p>But now, let's prevent this problem using <code>Pin</code>. We'll discuss
<code>Pin</code> more in the next chapter, but you'll get an introduction here by just
reading the comments.</p>
<pre><pre class="playpen"><code class="language-rust editable">#![feature(optin_builtin_traits)]
<pre><pre class="playpen"><code class="language-rust editable">#![feature(optin_builtin_traits)] // needed to implement `!Unpin`
use std::pin::Pin;
pub fn main() {
@@ -766,7 +771,7 @@ pub fn main() {
//let mut pinned2 = unsafe { Pin::new_unchecked(&amp;mut gen2) };
if let GeneratorState::Yielded(n) = pinned1.as_mut().resume() {
println!(&quot;Got value {}&quot;, n);
println!(&quot;Gen1 got value {}&quot;, n);
}
if let GeneratorState::Yielded(n) = pinned2.as_mut().resume() {
@@ -863,6 +868,43 @@ they did their unsafe implementation.</li>
<p>Hopefully, after this you'll have an idea of what happens when you use the
<code>yield</code> or <code>await</code> keywords inside an async function, and why we need <code>Pin</code> if
we want to be able to safely borrow across <code>yield/await</code> points.</p>
<h2><a class="header" href="#bonus" id="bonus">Bonus</a></h2>
<p>Thanks to <a href="https://github.com/rust-lang/rust/pull/45337/files">PR#45337</a> you can actually run code like the one we display here in Rust
today using the <code>static</code> keyword on nightly. Try it for yourself:</p>
<pre><pre class="playpen"><code class="language-rust">#![feature(generators, generator_trait)]
use std::ops::{Generator, GeneratorState};
pub fn main() {
let gen1 = static || {
let to_borrow = String::from(&quot;Hello&quot;);
let borrowed = &amp;to_borrow;
yield borrowed.len();
println!(&quot;{} world!&quot;, borrowed);
};
let gen2 = static || {
let to_borrow = String::from(&quot;Hello&quot;);
let borrowed = &amp;to_borrow;
yield borrowed.len();
println!(&quot;{} world!&quot;, borrowed);
};
let mut pinned1 = Box::pin(gen1);
let mut pinned2 = Box::pin(gen2);
if let GeneratorState::Yielded(n) = pinned1.as_mut().resume() {
println!(&quot;Gen1 got value {}&quot;, n);
}
if let GeneratorState::Yielded(n) = pinned2.as_mut().resume() {
println!(&quot;Gen2 got value {}&quot;, n);
};
let _ = pinned1.as_mut().resume();
let _ = pinned2.as_mut().resume();
}
</code></pre></pre>
<h1><a class="header" href="#pin" id="pin">Pin</a></h1>
<blockquote>
<p><strong>Relevant for</strong></p>
@@ -889,7 +931,7 @@ to govern the rules that need to apply for types which implement <code>!Unpin</c
<code>!Unpin</code> it's a good sign that it's time to lay down the work and start over
tomorrow with a fresh mind.</p>
<blockquote>
<p>That was of course a joke. There are very valid reasons for the names
<p>I hope you didn't mind the joke. There are valid reasons for the names
that were chosen. If you want you can read a bit of the discussion from the
<a href="https://internals.rust-lang.org/t/naming-pin-anchor-move/6864/12">internals thread</a>. The best takeaway from there in my eyes
is this quote from <code>tmandry</code>:</p>
@@ -1561,65 +1603,36 @@ fn main() {
reactor.lock().map(|mut r| r.close()).unwrap();
}
# // ============================ EXECUTOR ====================================
#
# // Our executor takes any object which implements the `Future` trait
# // ============================= EXECUTOR ====================================
# fn block_on&lt;F: Future&gt;(mut future: F) -&gt; F::Output {
# // the first thing we do is to construct a `Waker` which we'll pass on to
# // the `reactor` so it can wake us up when an event is ready.
# let mywaker = Arc::new(MyWaker{ thread: thread::current() });
# let waker = waker_into_waker(Arc::into_raw(mywaker));
# // The context struct is just a wrapper for a `Waker` object. Maybe in the
# // future this will do more, but right now it's just a wrapper.
# let mut cx = Context::from_waker(&amp;waker);
#
# // We poll in a loop, but it's not a busy loop. It will only run when
# // an event occurs, or a thread has a &quot;spurious wakeup&quot; (an unexpected wakeup
# // that can happen for no good reason).
# let val = loop {
# // So, since we run this on one thread and run one future to completion
# // we can pin the `Future` to the stack. This is unsafe, but saves an
# // allocation. We could `Box::pin` it too if we wanted. This is however
# // safe since we don't move the `Future` here.
# let pinned = unsafe { Pin::new_unchecked(&amp;mut future) };
# match Future::poll(pinned, &amp;mut cx) {
# // when the Future is ready we're finished
# Poll::Ready(val) =&gt; break val,
# // If we get a `pending` future we just go to sleep...
# Poll::Pending =&gt; thread::park(),
# };
# };
# val
# let mywaker = Arc::new(MyWaker{ thread: thread::current() });
# let waker = waker_into_waker(Arc::into_raw(mywaker));
# let mut cx = Context::from_waker(&amp;waker);
# let val = loop {
# let pinned = unsafe { Pin::new_unchecked(&amp;mut future) };
# match Future::poll(pinned, &amp;mut cx) {
# Poll::Ready(val) =&gt; break val,
# Poll::Pending =&gt; thread::park(),
# };
# };
# val
# }
#
# fn spawn&lt;F: Future&gt;(future: F) -&gt; Pin&lt;Box&lt;F&gt;&gt; {
# let mywaker = Arc::new(MyWaker{ thread: thread::current() });
# let waker = waker_into_waker(Arc::into_raw(mywaker));
# let mut cx = Context::from_waker(&amp;waker);
# let mut boxed = Box::pin(future);
# let _ = Future::poll(boxed.as_mut(), &amp;mut cx);
# boxed
# }
#
# // ====================== FUTURE IMPLEMENTATION ==============================
#
# // This is the definition of our `Waker`. We use a regular thread-handle here.
# // It works but it's not a good solution. If one of our `Futures` holds a handle
# // to our thread and takes it with it to a different thread the followinc could
# // happen:
# // 1. Our future calls `unpark` from a different thread
# // 2. Our `executor` thinks that data is ready and wakes up and polls the future
# // 3. The future is not ready yet but one nanosecond later the `Reactor` gets
# // an event and calles `wake()` which also unparks our thread.
# // 4. This could all happen before we go to sleep again since these processes
# // run in parallel.
# // 5. Our reactor has called `wake` but our thread is still sleeping since it was
# // awake alredy at that point.
# // 6. We're deadlocked and our program stops working
# // There are many better soloutions, here are some:
# // - Use `std::sync::CondVar`
# // - Use [crossbeam::sync::Parker](https://docs.rs/crossbeam/0.7.3/crossbeam/sync/# struct.Parker.html)
# #[derive(Clone)]
# struct MyWaker {
# thread: thread::Thread,
# }
#
# // This is the definition of our `Future`. It keeps all the information we
# // need. This one holds a reference to our `reactor`, that's just to make
# // this example as easy as possible. It doesn't need to hold a reference to
# // the whole reactor, but it needs to be able to register itself with the
# // reactor.
# #[derive(Clone)]
# pub struct Task {
# id: usize,
@@ -1628,26 +1641,18 @@ fn main() {
# is_registered: bool,
# }
#
# // These are function definitions we'll use for our waker. Remember the
# // &quot;Trait Objects&quot; chapter from the book.
# fn mywaker_wake(s: &amp;MyWaker) {
# let waker_ptr: *const MyWaker = s;
# let waker_arc = unsafe {Arc::from_raw(waker_ptr)};
# waker_arc.thread.unpark();
# }
#
# // Since we use an `Arc` cloning is just increasing the refcount on the smart
# // pointer.
# fn mywaker_clone(s: &amp;MyWaker) -&gt; RawWaker {
# let arc = unsafe { Arc::from_raw(s).clone() };
# std::mem::forget(arc.clone()); // increase ref count
# RawWaker::new(Arc::into_raw(arc) as *const (), &amp;VTABLE)
# }
#
# // This is actually a &quot;helper funtcion&quot; to create a `Waker` vtable. In contrast
# // to when we created a `Trait Object` from scratch we don't need to concern
# // ourselves with the actual layout of the `vtable` and only provide a fixed
# // set of functions
# const VTABLE: RawWakerVTable = unsafe {
# RawWakerVTable::new(
# |s| mywaker_clone(&amp;*(s as *const MyWaker)), // clone
@@ -1657,8 +1662,6 @@ fn main() {
# )
# };
#
# // Instead of implementing this on the `MyWaker` oject in `impl Mywaker...` we
# // just use this pattern instead since it saves us some lines of code.
# fn waker_into_waker(s: *const MyWaker) -&gt; Waker {
# let raw_waker = RawWaker::new(s as *const (), &amp;VTABLE);
# unsafe { Waker::from_raw(raw_waker) }
@@ -1675,27 +1678,16 @@ fn main() {
# }
# }
#
# // This is our `Future` implementation
# impl Future for Task {
# // The output for this kind of `leaf future` is just an `usize`. For other
# // futures this could be something more interesting like a byte stream.
# type Output = usize;
# fn poll(mut self: Pin&lt;&amp;mut Self&gt;, cx: &amp;mut Context&lt;'_&gt;) -&gt; Poll&lt;Self::Output&gt; {
# let mut r = self.reactor.lock().unwrap();
# // we check with the `Reactor` if this future is in its &quot;readylist&quot;
# if r.is_ready(self.id) {
# // if it is, we return the data. In this case it's just the ID of
# // the task.
# Poll::Ready(self.id)
# } else if self.is_registered {
# // If the future is registered alredy, we just return `Pending`
# Poll::Pending
# } else {
# // If we get here, it must be the first time this `Future` is polled
# // so we register a task with our `reactor`
# r.register(self.data, cx.waker().clone(), self.id);
# // oh, we have to drop the lock on our `Mutex` here because we can't
# // have a shared and exclusive borrow at the same time
# drop(r);
# self.is_registered = true;
# Poll::Pending
@@ -1704,20 +1696,11 @@ fn main() {
# }
#
# // =============================== REACTOR ===================================
# // This is a &quot;fake&quot; reactor. It does no real I/O, but that also makes our
# // code possible to run in the book and in the playground
# struct Reactor {
# // we need some way of registering a Task with the reactor. Normally this
# // would be an &quot;interest&quot; in an I/O event
# dispatcher: Sender&lt;Event&gt;,
# handle: Option&lt;JoinHandle&lt;()&gt;&gt;,
# // This is a list of tasks that are ready, which means they should be polled
# // for data.
# readylist: Arc&lt;Mutex&lt;Vec&lt;usize&gt;&gt;&gt;,
# }
#
# // We just have two kind of events. A timeout event, a &quot;timeout&quot; event called
# // `Timeout` and a `Close` event to close down our reactor.
# #[derive(Debug)]
# enum Event {
# Close,
@@ -1726,35 +1709,21 @@ fn main() {
#
# impl Reactor {
# fn new() -&gt; Self {
# // The way we register new events with our reactor is using a regular
# // channel
# let (tx, rx) = channel::&lt;Event&gt;();
# let readylist = Arc::new(Mutex::new(vec![]));
# let rl_clone = readylist.clone();
#
# // This `Vec` will hold handles to all threads we spawn so we can
# // join them later on and finish our programm in a good manner
# let mut handles = vec![];
# // This will be the &quot;Reactor thread&quot;
# let handle = thread::spawn(move || {
# // This simulates some I/O resource
# for event in rx {
# println!(&quot;REACTOR: {:?}&quot;, event);
# let rl_clone = rl_clone.clone();
# match event {
# // If we get a close event we break out of the loop we're in
# Event::Close =&gt; break,
# Event::Timeout(waker, duration, id) =&gt; {
#
# // When we get an event we simply spawn a new thread...
# let event_handle = thread::spawn(move || {
# //... which will just sleep for the number of seconds
# // we provided when creating the `Task`.
# thread::sleep(Duration::from_secs(duration));
# // When it's done sleeping we put the ID of this task
# // on the &quot;readylist&quot;
# rl_clone.lock().map(|mut rl| rl.push(id)).unwrap();
# // Then we call `wake` which will wake up our
# // executor and start polling the futures
# waker.wake();
# });
#
@@ -1763,9 +1732,6 @@ fn main() {
# }
# }
#
# // When we exit the Reactor we first join all the handles on
# // the child threads we've spawned so we catch any panics and
# // release all resources.
# for handle in handles {
# handle.join().unwrap();
# }
@@ -1779,8 +1745,6 @@ fn main() {
# }
#
# fn register(&amp;mut self, duration: u64, waker: Waker, data: usize) {
# // registering an event is as simple as sending an `Event` through
# // the channel.
# self.dispatcher
# .send(Event::Timeout(waker, duration, data))
# .unwrap();
@@ -1790,9 +1754,6 @@ fn main() {
# self.dispatcher.send(Event::Close).unwrap();
# }
#
# // We need a way to check if any event's are ready. This will simply
# // look through the &quot;readylist&quot; for an event macthing the ID we want to
# // check for.
# fn is_ready(&amp;self, id_to_check: usize) -&gt; bool {
# self.readylist
# .lock()
@@ -1801,16 +1762,19 @@ fn main() {
# }
# }
#
# // When our `Reactor` is dropped we join the reactor thread with the thread
# // owning our `Reactor` so we catch any panics and release all resources.
# // It's not needed for this to work, but it really is a best practice to join
# // all threads you spawn.
# impl Drop for Reactor {
# fn drop(&amp;mut self) {
# self.handle.take().map(|h| h.join().unwrap()).unwrap();
# }
# }
</code></pre></pre>
<p>I added a debug printout of the events the reactor registered interest for so we can observe
two things:</p>
<ol>
<li>How the <code>Waker</code> object looks just like the <em>trait object</em> we talked about in an earlier chapter</li>
<li>In what order the events register interest with the reactor</li>
</ol>
<p>The last point is relevant when we move on the the last paragraph.</p>
<h2><a class="header" href="#asyncawait-and-concurrent-futures" id="asyncawait-and-concurrent-futures">Async/Await and concurrent Futures</a></h2>
<p>This is the first time we actually see the <code>async/await</code> syntax so let's
finish this book by explaining them briefly.</p>
@@ -1860,7 +1824,7 @@ come up with:</p>
}
</code></pre>
<p>Now if we change our code in <code>main</code> to look like this instead.</p>
<pre><pre class="playpen"><code class="language-rust"># use std::{
<pre><pre class="playpen"><code class="language-rust edition2018"># use std::{
# future::Future, pin::Pin, sync::{mpsc::{channel, Sender}, Arc, Mutex},
# task::{Context, Poll, RawWaker, RawWakerVTable, Waker},
# thread::{self, JoinHandle}, time::{Duration, Instant}
@@ -1901,7 +1865,7 @@ fn main() {
block_on(mainfut);
reactor.lock().map(|mut r| r.close()).unwrap();
}
# //// ===== EXECUTOR =====
# // ============================= EXECUTOR ====================================
# fn block_on&lt;F: Future&gt;(mut future: F) -&gt; F::Output {
# let mywaker = Arc::new(MyWaker{ thread: thread::current() });
# let waker = waker_into_waker(Arc::into_raw(mywaker));
@@ -1925,7 +1889,7 @@ fn main() {
# boxed
# }
#
# // ===== FUTURE IMPLEMENTATION =====
# // ====================== FUTURE IMPLEMENTATION ==============================
# #[derive(Clone)]
# struct MyWaker {
# thread: thread::Thread,
@@ -1993,7 +1957,7 @@ fn main() {
# }
# }
#
# // ===== REACTOR =====
# // =============================== REACTOR ===================================
# struct Reactor {
# dispatcher: Sender&lt;Event&gt;,
# handle: Option&lt;JoinHandle&lt;()&gt;&gt;,
@@ -2002,7 +1966,7 @@ fn main() {
# #[derive(Debug)]
# enum Event {
# Close,
# Simple(Waker, u64, usize),
# Timeout(Waker, u64, usize),
# }
#
# impl Reactor {
@@ -2014,11 +1978,11 @@ fn main() {
# let handle = thread::spawn(move || {
# // This simulates some I/O resource
# for event in rx {
# println!(&quot;GOT EVENT: {:?}&quot;, event);
# println!(&quot;REACTOR: {:?}&quot;, event);
# let rl_clone = rl_clone.clone();
# match event {
# Event::Close =&gt; break,
# Event::Simple(waker, duration, id) =&gt; {
# Event::Timeout(waker, duration, id) =&gt; {
# let event_handle = thread::spawn(move || {
# thread::sleep(Duration::from_secs(duration));
# rl_clone.lock().map(|mut rl| rl.push(id)).unwrap();
@@ -2044,7 +2008,7 @@ fn main() {
#
# fn register(&amp;mut self, duration: u64, waker: Waker, data: usize) {
# self.dispatcher
# .send(Event::Simple(waker, duration, data))
# .send(Event::Timeout(waker, duration, data))
# .unwrap();
# }
#
@@ -2118,7 +2082,7 @@ fn main() {
reactor.lock().map(|mut r| r.close()).unwrap();
}
//// ===== EXECUTOR =====
// ============================= EXECUTOR ====================================
fn block_on&lt;F: Future&gt;(mut future: F) -&gt; F::Output {
let mywaker = Arc::new(MyWaker{ thread: thread::current() });
let waker = waker_into_waker(Arc::into_raw(mywaker));
@@ -2142,7 +2106,7 @@ fn spawn&lt;F: Future&gt;(future: F) -&gt; Pin&lt;Box&lt;F&gt;&gt; {
boxed
}
// ===== FUTURE IMPLEMENTATION =====
// ====================== FUTURE IMPLEMENTATION ==============================
#[derive(Clone)]
struct MyWaker {
thread: thread::Thread,
@@ -2210,7 +2174,7 @@ impl Future for Task {
}
}
// ===== REACTOR =====
// =============================== REACTOR ===================================
struct Reactor {
dispatcher: Sender&lt;Event&gt;,
handle: Option&lt;JoinHandle&lt;()&gt;&gt;,
@@ -2231,7 +2195,7 @@ impl Reactor {
let handle = thread::spawn(move || {
// This simulates some I/O resource
for event in rx {
println!(&quot;GOT EVENT: {:?}&quot;, event);
println!(&quot;REACTOR: {:?}&quot;, event);
let rl_clone = rl_clone.clone();
match event {
Event::Close =&gt; break,
@@ -2356,6 +2320,7 @@ articles I've already linked to in the book, here are some of my suggestions:</p
<p><a href="https://tokio.rs/blog/2019-10-scheduler/">The Tokio Blog</a></p>
<p><a href="https://stjepang.github.io/">Stjepan's blog with a series where he implements an Executor</a></p>
<p><a href="https://youtu.be/DkMwYxfSYNQ">Jon Gjengset's video on The Why, What and How of Pinning in Rust</a></p>
<p><a href="https://boats.gitlab.io/blog/post/2018-01-25-async-i-self-referential-structs/">Withoutboats blog series about async/await</a></p>
</main>