$ cargo new --bin usetokio </code></pre> If you want to make sure you're using the same compiler version as me, set up your rust-toolchain</code> correctly:$ echo 1.39.0 > rust-toolchain </code></pre> And then set up Tokio as a dependency. For simplicity, we'll install all the bells and whistles. In your Cargo.toml</code>: [dependencies] tokio = { version = "0.2", features = ["full"] } </code></pre> PROTIP You can run cargo build</code> now to kick off the download and build of crates while you keep reading... And now we're going to write an asynchronous hello world application. Type this into your src/main.rs</code>: use tokio::io; #[tokio::main] async fn main() -> Result<(), std::io::Error> { let mut stdout = io::stdout(); let mut hello: &[u8] = b"Hello, world!\n"; io::copy(&mut hello, &mut stdout).await?; Ok(()) } </code></pre> NOTE I specifically said "type this in" instead of "copy and paste." For getting comfortable with this stuff, I recommend manually typing in the code. A lot of this should look familiar from our previous lesson. To recap: Since we'll be await</code>ing something and generating a Future</code>, our main</code> function is async</code>.</li> Since main</code> is async</code>, we need to use an executor to run it. That's why we use the #[tokio::main]</code> attribute.</li> Since performing I/O can fail, we return a Result</code>.</li> </ul> The first really new thing since last lesson is this little bit of syntax: .await? </code></pre> I mentioned it last time, but now we're seeing it in real life. This is just the combination of our two pieces of prior art: .await</code> for chaining together Future</code>s, and ?</code> for error handling. The fact that these work together so nicely is really awesome. I'll probably mention this a few more times, because I love it that much. The next thing to note is that we use tokio::io::stdout()</code> to get access to some value that lets us interact with standard output. If you're familiar with it, this looks really similar to std::io::stdout()</code>. That's by design: a large part of the tokio</code> API is simply async-ifying things from std</code>. And finally, we can look at the actual tokio::io::copy</code> call. As you may have guessed, and as stated in the API docs</a>: This is an asynchronous version of std::io::copy</code></a>. </blockquote> However, instead of working with the Read</code> and Write</code> traits, this works with their async cousins: AsyncRead</code> and AsyncWrite</code>. A byte slice (&[u8]</code>) is a valid AsyncRead</code>, so we're able to store our input there. And as you may have guessed, Stdout</code> is an AsyncWrite</code>. EXERCISE 1 Modify this application so that instead of printing "Hello, world!", it copies the entire contents of standard input to standard output. NOTE You can simplify this code using stdout.write_all</code> after use</code>ing tokio::io::AsyncWriteExt</code>, but we'll stick to tokio::io::copy</code>, since we'll be using it throughout. But if you're curious: use tokio::io::{self, AsyncWriteExt}; #[tokio::main] async fn main() -> Result<(), std::io::Error> { let mut stdout = io::stdout(); stdout.write_all(b"Hello, world!\n").await?; Ok(()) } </code></pre>Spawning processes</h2> Tokio provides a tokio::process</code> module which resembles the std::process</code> module. We can use this to implement Hello World once again: use tokio::process::Command; #[tokio::main] async fn main() -> Result<(), std::io::Error> { Command::new("echo").arg("Hello, world!").spawn()?.await?; Ok(()) } </code></pre> Notice how the ?</code> and .await</code> bits can go in whatever order they are needed. You can read this line as: Create a new Command</code> to run echo</code></li> Give it the argument "Hello, world!"</code></li> Spawn this, which may fail</li> Using the first ?</code>: if it fails, return the error. Otherwise, return a Future</code></li> Using the .await</code>: wait until that Future</code> completes, and capture its Result</code></li> Using the second ?</code>: if that Result</code> is Err</code>, return that error.</li> </ul> Pretty nice for a single line! One of the great advantages of async</code>/.await</code> versus the previous way of doing async with callbacks is how easily it works with looping. EXERCISE 2 Extend this example so that it prints Hello, world!</code> 10 times. Take a break</h2> So far we've only really done a single bit of .await</code>ing. But it's easy enough to .await</code> on multiple things. Let's use delay_for</code></a> to pause for a bit. use tokio::time; use tokio::process::Command; use std::time::Duration; #[tokio::main] async fn main() -> Result<(), std::io::Error> { Command::new("date").spawn()?.await?; time::delay_for(Duration::from_secs(1)).await; Command::new("date").spawn()?.await?; time::delay_for(Duration::from_secs(1)).await; Command::new("date").spawn()?.await?; Ok(()) } </code></pre> We can also use the tokio::time::interval</code> function to create a stream of "ticks" for each time a certain amount of time has passed. For example, this program will keep calling date</code> once per second until it is killed: use tokio::time; use tokio::process::Command; use std::time::Duration; #[tokio::main] async fn main() -> Result<(), std::io::Error> { let mut interval = time::interval(Duration::from_secs(1)); loop { interval.tick().await; Command::new("date").spawn()?.await?; } } </code></pre> EXERCISE 3 Why isn't there a Ok(())</code> after the loop</code>? Time to spawn</h2> This is all well and good, but we're not really taking advantage of asynchronous programming at all. Let's fix that! We've seen two different interesting programs: Infinitely pausing 1 seconds and calling date</code></li> Copying all input from stdin</code> to stdout</code></li> </ol> It's time to introduce spawn</code> so that we can combine these two into one program. First, let's demonstrate a trivial usage of spawn</code>: use std::time::Duration; use tokio::process::Command; use tokio::task; use tokio::time; #[tokio::main] async fn main() -> Result<(), std::io::Error> { task::spawn(dating()).await??; Ok(()) } async fn dating() -> Result<(), std::io::Error> { let mut interval = time::interval(Duration::from_secs(1)); loop { interval.tick().await; Command::new("date").spawn()?.await?; } } </code></pre> You may be wondering: what's up with that ??</code> operator? Is that some special super-error handler? No, it's just the normal error handling ?</code> applied twice. Let's look at some type signatures to help us out here: pub fn spawn<T>(task: T) -> JoinHandle<T::Output>; impl<T> Future for JoinHandle<T> { type Output = Result<T, JoinError>; } </code></pre> Calling spawn</code> gives us back a JoinHandle<T::Output></code>. In our case, the Future</code> we provide as input is dating()</code>, which has an output of type Result<(), std::io::Error></code>. So that means the type of task::spawn(dating())</code> is JoinHandle<Result<(), std::io::Error>></code>. We also see that JoinHandle</code> implements Future</code>. So when we apply .await</code> to this value, we end up with whatever that type Output = Result<T, JoinError></code> thing is. Since we know that T</code> is Result<(), std::io::Error></code>, this means we end up with Result<Result<(), std::io::Error>, JoinError></code>. The first ?</code> deals with the outer Result</code>, exiting with the JoinError</code> on an Err</code>, and giving us a Result<(), std::io::Error></code> value on Ok</code>. The second ?</code> deals with the std::io::Error</code>, giving us a ()</code> on Ok</code>. Whew! EXERCISE 4 Now that we've seen spawn</code>, you should modify the program so that it calls both date</code> in a loop, and copies stdin</code> to stdout</code>. Synchronous code</h2> You may not have the luxury of interacting exclusively with async</code>-friendly code. Maybe you have some really nice library you want to leverage, but it performs blocking calls internally. Fortunately, Tokio's got you covered with the spawn_blocking</code></a> function. Since the docs are so perfect, let me quote them: The task::spawn_blocking</code> function is similar to the task::spawn</code> function discussed in the previous section, but rather than spawning an non-blocking</code> future on the Tokio runtime, it instead spawns a blocking</code> function on a dedicated thread pool for blocking tasks. </blockquote> EXERCISE 5 Rewrite the dating()</code> function to use spawn_blocking</code> and std::thread::sleep</code> so that it calls date</code> approximately once per second. Let's network!</h2> I could keep stepping through the other cools functions in the Tokio library. I encourage you to poke around at them yourself. But I promised some networking, and by golly, I'm gonna deliver! I'm going to slightly extend the example from the TcpListener</code> docs</a> to (1) make it compile and (2) implement an echo server. This program has a pretty major flaw in it though, I recommend trying to find it. use tokio::io; use tokio::net::{TcpListener, TcpStream}; #[tokio::main] async fn main() -> io::Result<()> { let mut listener = TcpListener::bind("127.0.0.1:8080").await?; loop { let (socket, _) = listener.accept().await?; echo(socket).await?; } } async fn echo(socket: TcpStream) -> io::Result<()> { let (mut recv, mut send) = io::split(socket); io::copy(&mut recv, &mut send).await?; Ok(()) } </code></pre> We use TcpListener</code> to bind a socket. The binding itself is asynchronous, so we use .await</code> to wait for the listening socket to be available. And we use ?</code> to deal with any errors while binding the listening socket. Next, we loop forever. Inside the loop, we accept new connections, using .await?</code> like before. We capture the socket</code> (ignoring the address as the second part of the tuple). Then we call our echo</code> function and .await</code> it. Within echo</code>, we use tokio::io::split</code> to split up our TcpStream</code> into its constituent read and write halves, and then pass those into tokio::io::copy</code>, as we've done before. Awesome! Where's the bug? Let me ask you a question: what should the behavior be if a second connection comes in while the first connection is still active? Ideally, it would be handled. However, our program has just one task. And that task .await</code>s on each call to echo</code>. So our second connection won't be serviced until the first one closes. EXERCISE 6 Modify the program above so that it handles concurrent connections correctly. TCP client and ownership</h2> Let's write a poor man's HTTP client. It will establish a connection to a hard-coded server, copy all of stdin</code> to the server, and then copy all data from the server to stdout</code>. To use this, you'll manually type in the HTTP request and then hit Ctrl-D</code> for end-of-file. use tokio::io; use tokio::net::TcpStream; #[tokio::main] async fn main() -> io::Result<()> { let stream = TcpStream::connect("127.0.0.1:8080").await?; let (mut recv, mut send) = io::split(stream); let mut stdin = io::stdin(); let mut stdout = io::stdout(); io::copy(&mut stdin, &mut send).await?; io::copy(&mut recv, &mut stdout).await?; Ok(()) } </code></pre> That's all well and good, but it's limited. It only handles half-duplex protocols like HTTP, and doesn't actually support keep-alive in any way. We'd like to use spawn</code> to run the two copy</code>s in different tasks. Seems easy enough: let send = spawn(io::copy(&mut stdin, &mut send)); let recv = spawn(io::copy(&mut recv, &mut stdout)); send.await??; recv.await??; </code></pre> Unfortunately, this doesn't compile. We get four nearly-identical error messages. Let's look at the first: error[E0597]: `stdin` does not live long enough --> src/main.rs:12:31 | 12 | let send = spawn(io::copy(&mut stdin, &mut send)); | ---------^^^^^^^^^^------------ | | | | | borrowed value does not live long enough | argument requires that `stdin` is borrowed for `'static` ... 19 | } | - `stdin` dropped here while still borrowed </code></pre> Here's the issue: our copy</code> Future</code> does not own the stdin</code> value (or the send</code> value, for that matter). Instead, it has a (mutable) reference to it. That value remains in the main</code> function's Future</code>. Ignoring error cases, we know that the main</code> function will wait for send</code> to complete (thanks to send.await</code>), and therefore the lifetimes appear to be correct. However, Rust doesn't recognize this lifetime information. (Also, and I haven't thought this through completely, I'm fairly certain that send</code> may be dropped earlier than the Future</code> using it in the case of panic</code>s.) In order to fix this, we need to convince the compiler to make a Future</code> that owns stdin</code>. And the easiest way to do that here is to use an async move</code> block. Exercise 7 Make the code above compile using two async move</code> blocks. Playing with lines</code></h2> This section will have a series of modifications to a program. I recommend you solve each challenge before looking at the solution. However, unlike the other exercises, I'm going to show the solutions inline since they build on each other. Let's build an async program that counts the number of lines on standard input. You'll want to use the lines</code></a> method for this. Read the docs and try to figure out what use</code>s and wrappers will be necessary to make the types line up. use tokio::prelude::*; use tokio::io::AsyncBufReadExt; #[tokio::main] async fn main() -> Result<(), std::io::Error> { let stdin = io::stdin(); let stdin = io::BufReader::new(stdin); let mut count = 0u32; let mut lines = stdin.lines(); while let Some(_) = lines.next_line().await? { count += 1; } println!("Lines on stdin: {}", count); Ok(()) } </code></pre> OK, bumping this up one more level. Instead of standard input, let's take a list of file names as command line arguments, and count up the total number of lines in all the files. Initially, it's OK to read the files one at a time. In other words: don't bother calling spawn</code>. Give it a shot, and then come back here: use tokio::prelude::*; use tokio::io::AsyncBufReadExt; #[tokio::main] async fn main() -> Result<(), std::io::Error> { let mut args = std::env::args(); let _me = args.next(); // ignore command name let mut count = 0u32; for filename in args { let file = tokio::fs::File::open(filename).await?; let file = io::BufReader::new(file); let mut lines = file.lines(); while let Some(_) = lines.next_line().await? { count += 1; } } println!("Total lines: {}", count); Ok(()) } </code></pre> But now it's time to make this properly asynchronous, and process the files in separate spawn</code>ed tasks. In order to make this work, we need to spawn all of the tasks, and then .await</code> each of them. I used a Vec</code> of Future<Output=Result<u32, std::io::Error>></code>s for this. Give it a shot! use tokio::prelude::*; use tokio::io::AsyncBufReadExt; #[tokio::main] async fn main() -> Result<(), std::io::Error> { let mut args = std::env::args(); let _me = args.next(); // ignore command name let mut tasks = vec![]; for filename in args { tasks.push(tokio::spawn(async { let file = tokio::fs::File::open(filename).await?; let file = io::BufReader::new(file); let mut lines = file.lines(); let mut count = 0u32; while let Some(_) = lines.next_line().await? { count += 1; } Ok(count) as Result<u32, std::io::Error> })); } let mut count = 0; for task in tasks { count += task.await??; } println!("Total lines: {}", count); Ok(()) } </code></pre> And finally in this progression: let's change how we handle the count</code>. Instead of .await</code>ing the count in the second for</code> loop, let's have each individual task update a shared mutable variable. You should use an Arc<Mutex<u32>></code> for that. You'll still need to keep a Vec</code> of the tasks though to ensure you wait for all files to be read. use tokio::prelude::*; use tokio::io::AsyncBufReadExt; use std::sync::Arc; // avoid thread blocking by using Tokio's mutex use tokio::sync::Mutex; #[tokio::main] async fn main() -> Result<(), std::io::Error> { let mut args = std::env::args(); let _me = args.next(); // ignore command name let mut tasks = vec![]; let count = Arc::new(Mutex::new(0u32)); for filename in args { let count = count.clone(); tasks.push(tokio::spawn(async move { let file = tokio::fs::File::open(filename).await?; let file = io::BufReader::new(file); let mut lines = file.lines(); let mut local_count = 0u32; while let Some(_) = lines.next_line().await? { local_count += 1; } let mut count = count.lock().await; *count += local_count; Ok(()) as Result<(), std::io::Error> })); } for task in tasks { task.await??; } let count = count.lock().await; println!("Total lines: {}", *count); Ok(()) } </code></pre>LocalSet and !Send</code></h2> Thanks to @xudehseng</a> for the inspiration on this section. OK, did that last exercise seem a bit contrived? It was! In my opinion, the previous approach of .await</code>ing the counts and summing in the main</code> function itself was superior. However, I wanted to teach you something else. What happens if you replace the Arc<Mutex<u32>></code> with a Rc<RefCell<u32>></code>? With this code: use tokio::prelude::*; use tokio::io::AsyncBufReadExt; use std::rc::Rc; use std::cell::RefCell; #[tokio::main] async fn main() -> Result<(), std::io::Error> { let mut args = std::env::args(); let _me = args.next(); // ignore command name let mut tasks = vec![]; let count = Rc::new(RefCell::new(0u32)); for filename in args { let count = count.clone(); tasks.push(tokio::spawn(async { let file = tokio::fs::File::open(filename).await?; let file = io::BufReader::new(file); let mut lines = file.lines(); let mut local_count = 0u32; while let Some(_) = lines.next_line().await? { local_count += 1; } *count.borrow_mut() += local_count; Ok(()) as Result<(), std::io::Error> })); } for task in tasks { task.await??; } println!("Total lines: {}", count.borrow()); Ok(()) } </code></pre> You get an error: error[E0277]: `std::rc::Rc<std::cell::RefCell<u32>>` cannot be shared between threads safely --> src/main.rs:15:20 | 15 | tasks.push(tokio::spawn(async { | ^^^^^^^^^^^^ `std::rc::Rc<std::cell::RefCell<u32>>` cannot be shared between threads safely | ::: /Users/michael/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.2/src/task/spawn.rs:49:17 | 49 | T: Future + Send + 'static, | ---- required by this bound in `tokio::task::spawn::spawn` </code></pre> Tasks can be scheduled to multiple different threads. Therefore, your Future</code> must be Send</code>. And Rc<RefCell<u32>></code> is definitely !Send</code>. However, in our use case, using multiple OS threads is unlikely to speed up our program; we're going to be doing lots of blocking I/O. It would be nice if we could insist on spawning all our tasks on the same OS thread and avoid the need for Send</code>. And sure enough, Tokio provides such a function: tokio::task::spawn_local</code>. Using it (and adding back in async move</code> instead of async</code>), our program compiles, but breaks at runtime: thread 'main' panicked at '`spawn_local` called from outside of a local::LocalSet!', src/libcore/option.rs:1190:5 </code></pre> Uh-oh! Now I'm personally not a big fan of this detect-it-at-runtime stuff, but the concept is simple enough: if you want to spawn onto the current thread, you need to set up your runtime to support that. And the way we do that is with LocalSet</code></a>. In order to use this, you'll need to ditch the #[tokio::main]</code> attribute. EXERCISE 8 Follow the documentation for LocalSet</code> to make the program above work with Rc<RefCell<u32>></code>. Conclusion</h2> That lesson felt short. Definitely compared to the previous Tokio lesson which seemed to go on forever. I think this is a testament to how easy to use the new async/</code>.await` syntax is. There's obviously a lot more that can be covered in asynchronous programming, but hopefully this establishes the largest foundations you need to understand to work with the async</code>/.await</code> syntax and the Tokio library itself. If we have future lessons, I believe they'll cover additional libraries like Hyper as they move over to Tokio 0.2, as well as specific use cases people raise. If you want something covered, mention it to me on Twitter or in the comments below. Solutions</h2> Solution 1</h3> use tokio::io; #[tokio::main] async fn main() -> Result<(), std::io::Error> { let mut stdin = io::stdin(); let mut stdout = io::stdout(); io::copy(&mut stdin, &mut stdout).await?; Ok(()) } </code></pre>Solution 2</h3> use tokio::process::Command; #[tokio::main] async fn main() -> Result<(), std::io::Error> { for _ in 1..=10 { Command::new("echo").arg("Hello, world!").spawn()?.await?; } Ok(()) } </code></pre>Solution 3</h3> Since the loop</code> will either run forever or be short circuited by an error, any code following loop</code> will never actually be called. Therefore, code placed there will generate a warning. Solution 4</h3> use std::time::Duration; use tokio::process::Command; use tokio::{io, task, time}; #[tokio::main] async fn main() -> Result<(), std::io::Error> { let dating = task::spawn(dating()); let copying = task::spawn(copying()); dating.await??; copying.await??; Ok(()) } async fn dating() -> Result<(), std::io::Error> { let mut interval = time::interval(Duration::from_secs(1)); loop { interval.tick().await; Command::new("date").spawn()?.await?; } } async fn copying() -> Result<(), std::io::Error> { let mut stdin = io::stdin(); let mut stdout = io::stdout(); io::copy(&mut stdin, &mut stdout).await?; Ok(()) } </code></pre>Solution 5</h3> async fn dating() -> Result<(), std::io::Error> { loop { task::spawn_blocking(|| { std::thread::sleep(Duration::from_secs(1)) }).await?; Command::new("date").spawn()?.await?; } } </code></pre>Solution 6</h3> The simplest tweak is to wrap the echo</code> call with tokio::spawn</code>: loop { let (socket, _) = listener.accept().await?; tokio::spawn(echo(socket)); } </code></pre> There is a downside to this worth noting, however: we're ignoring the errors produced by the spawned tasks. Likely the best behavior in this case is to handle the errors inside the spawned task: #[tokio::main] async fn main() -> io::Result<()> { let mut listener = TcpListener::bind("127.0.0.1:8080").await?; let mut counter = 1u32; loop { let (socket, _) = listener.accept().await?; println!("Accepted connection #{}", counter); tokio::spawn(async move { match echo(socket).await { Ok(()) => println!("Connection #{} completed successfully", counter), Err(e) => println!("Connection #{} errored: {:?}", counter, e), } }); counter += 1; } } </code></pre>Exericse 7</h3> use tokio::io; use tokio::spawn; use tokio::net::TcpStream; #[tokio::main] async fn main() -> io::Result<()> { let stream = TcpStream::connect("127.0.0.1:8080").await?; let (mut recv, mut send) = io::split(stream); let mut stdin = io::stdin(); let mut stdout = io::stdout(); let send = spawn(async move { io::copy(&mut stdin, &mut send).await }); let recv = spawn(async move { io::copy(&mut recv, &mut stdout).await }); send.await??; recv.await??; Ok(()) } </code></pre>Solution 8</h2> use tokio::prelude::*; use tokio::io::AsyncBufReadExt; use std::rc::Rc; use std::cell::RefCell; fn main() -> Result<(), std::io::Error> { let mut rt = tokio::runtime::Runtime::new()?; let local = tokio::task::LocalSet::new(); local.block_on(&mut rt, main_inner()) } async fn main_inner() -> Result<(), std::io::Error> { let mut args = std::env::args(); let _me = args.next(); // ignore command name let mut tasks = vec![]; let count = Rc::new(RefCell::new(0u32)); for filename in args { let count = count.clone(); tasks.push(tokio::task::spawn_local(async move { let file = tokio::fs::File::open(filename).await?; let file = io::BufReader::new(file); let mut lines = file.lines(); let mut local_count = 0u32; while let Some(_) = lines.next_line().await? { local_count += 1; } *count.borrow_mut() += local_count; Ok(()) as Result<(), std::io::Error> })); } for task in tasks { task.await??; } println!("Total lines: {}", count.borrow()); Ok(()) } </code></pre>

Async, futures, and tokio - Rust Crash Course lesson 7

Mon, 03 Dec 2018 00:00:00 +0000

Heads up This blog post series has been updated and published as an eBook by FP Complete. I'd recommend reading that version instead of these posts. If you're interested, please check out the Rust Crash Course eBook</a>.

NOTE With the addition of async/await syntax in Rust 1.39 (November 2019), everything related to async code in Rust is getting an overhaul. As such, this lesson is now pretty deeply out of date. It's still useful for understanding the deeper principles, but I hope to write up an updated tutorial in the future covering the new approach.

Unlike languages like Haskell, Erlang, and Go, Rust does not have a runtime system providing green threads and asynchronous I/O. However, for many real world use cases, async I/O is strongly desired, if not a hard requirement. The de facto standard library for handling this in Rust is tokio.

This post is part of a series based on teaching Rust at FP Complete</a>. If you're reading this post outside of the blog, you can find links to all posts in the series at the top of the introduction post</a>. You can also subscribe to the RSS feed</a>.

This lesson in the crash course is going to be a bit different from others, since:

There's a lot of interconnected material to cover which can't as easily be separated out</li>
It's more important to understand the motivation behind the design of these libraries than in many other cases</li>
I believe the material may be useful for people who haven't been following the rest of the crash course</li> </ol>
Therefore, some different rules will apply:

I'm defining up front the knowledge you'll need of Rust to understand this lesson, namely:

All the basics of syntax</li>
Traits and associated types</li>
Iterators</li>
Closures</li> </ul> </li>
Instead of providing the exercise solutions in a later post, I'll be providing them immediately, since the material is so cumulative. I still strongly recommend spending significant time and effort trying to solve the exercises yourself before looking at the solutions. It's harder and more time consuming, but ultimately worthwhile.</li> </ul>
Also, this lesson is much longer and more involved than previous lessons. You should plan on it taking more time to complete than others. I considered breaking this up into multiple lessons, but decided to keep all of the content together. Instead, I'll be taking a break from weekly lessons after this one for a bit.
Consider that the intro to the intro. Now the real intro!
Why async?</h2> I'm going to assume that readers are already familiar with async I/O and its motivations in general. If you're not, it's worth reading up a bit on the C10k problem</a>, where many of us started thinking hard about async I/O. You may also be interested in reading a post I wrote about green threads</a>, a language runtime-based solution to the same problem.
At the end of the day, the goals of Rust's approach to async I/O are:

Minimize system resources for handling a large number of concurrent I/O tasks</li>
Provide a zero-cost abstraction on top of the async I/O mechanisms provided by operating systems</li>
Do it at a library level, instead of introducing a runtime to Rust</li> </ul>
A sample problem</h2>
It's difficult to dive in to the wonderful world of tokio. You need to learn about futures and streams, tasks and executors, async I/O system calls and the Async</code> type, etc. To try and decouple this learning experience, we're going to start with a simplified problem. This problem mostly, but not perfectly, models async I/O in the real world, and will demonstrate many of the design concerns. It will also let us play just a bit more with concurrent programming before diving into futures and tokio.
We're going to run a separate thread. This thread will have access to two atomic values:
An AtomicBool</code> to tell us whether we want this side thread to keep running</li>An AtomicUsize</code> counter</li> </ul> We haven't covered atomic types yet, but they're exactly what they sound like: variables that can be safely accessed from multiple threads. Since learning about them isn't the point of this lesson, I'll defer further questions about the usage of these types to their API documentation</a>. Our separate thread is going to run in a loop. As long as the AtomicBool</code> is true</code>, it will: Sleep for a given number of milliseconds</li> Print a message to the console</li> Increment the AtomicUsize</code> counter</li> </ul> The series of examples we'll be looking at will then try different approaches to observing the changes in the counter. Interval</h2> We're going to call this thing an Interval</code>, since it somewhat represents a setInterval</code> call in Javascript. Go ahead and start a new project: $ cargo new interval --bin $ cd interval </code></pre> We're going to put the code for our Interval</code> into a separate module. First, put the following code into src/interval.rs</code>: use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; use std::thread::{sleep, spawn}; use std::time::Duration; #[derive(Clone)] pub struct Interval { counter: Arc<AtomicUsize>, still_running: Arc<AtomicBool>, } impl Drop for Interval { fn drop(&mut self) { println!("Interval thread shutting down"); self.still_running.store(false, Ordering::SeqCst); } } impl Interval { pub fn from_millis(millis: u64) -> Interval { let duration = Duration::from_millis(millis); let counter = Arc::new(AtomicUsize::new(0)); let counter_clone = counter.clone(); let still_running = Arc::new(AtomicBool::new(true)); let still_running_clone = still_running.clone(); spawn(move || { println!("Interval thread launched"); while still_running_clone.load(Ordering::SeqCst) { sleep(duration); let old = counter_clone.fetch_add(1, Ordering::SeqCst); println!("Interval thread still alive, value was: {}", old); } }); Interval { counter, still_running, } } pub fn get_counter(&self) -> usize { self.counter.load(Ordering::SeqCst) } } </code></pre> Next, let's provide a minimal src/main.rs</code> which uses this Interval</code> data type: mod interval; use self::interval::Interval; fn main() { let interval = Interval::from_millis(500); // half a second let duration = std::time::Duration::from_millis(2000); // 2 seconds for i in 1..11 { println!("Iteration number {}, counter is {}", i, interval.get_counter()); std::thread::sleep(duration); } } </code></pre> You should see something like the following: Iteration number 1, counter is 0 Interval thread launched Interval thread still alive, value was: 0 Interval thread still alive, value was: 1 Interval thread still alive, value was: 2 Iteration number 2, counter is 3 Interval thread still alive, value was: 3 Interval thread still alive, value was: 4 ... Interval thread still alive, value was: 33 Interval thread still alive, value was: 34 Iteration number 10, counter is 35 Interval thread still alive, value was: 35 Interval thread still alive, value was: 36 Interval thread still alive, value was: 37 Interval thread still alive, value was: 38 Interval thread shutting down </code></pre> Hurrah, we have some concurrent communication. Problems with this approach</h2> The first thing that jumps out as a problem is that we're missing some updates in the main thread. Notice how the counter jumps from 0 to 3. This is obviously a problem with the interval set in the main thread: we're delaying for 2 seconds instead of half a second. Let's instead delay for a tenth of a second (100ms) in the main thread, and check if the value has changed since last time. NOTE It's still possible we'll miss some updates this way, since sleep</code> guarantees a thread will sleep at least a given amount of time, but may sleep longer. However, by having such a large difference, we're fairly certain we'll catch all of the updates. fn main() { let interval = Interval::from_millis(500); // half a second let duration = std::time::Duration::from_millis(100); // 0.1 seconds let mut last = interval.get_counter(); for i in 1..51 { let curr = interval.get_counter(); if curr != last { last = curr; println!("Iteration number {}, counter is {}", i, curr); } std::thread::sleep(duration); } } </code></pre> I had to increase the number of iterations to 50, because so many of our main thread iterations end up showing no change in the counter. Here's an example of running this on my machine: Interval thread launched Interval thread still alive, value was: 0 Iteration number 6, counter is 1 Interval thread still alive, value was: 1 Iteration number 11, counter is 2 Interval thread still alive, value was: 2 Iteration number 16, counter is 3 Interval thread still alive, value was: 3 Iteration number 21, counter is 4 Interval thread still alive, value was: 4 Iteration number 26, counter is 5 Interval thread still alive, value was: 5 Iteration number 31, counter is 6 Interval thread still alive, value was: 6 Iteration number 36, counter is 7 Interval thread still alive, value was: 7 Iteration number 41, counter is 8 Interval thread still alive, value was: 8 Iteration number 46, counter is 9 Interval thread still alive, value was: 9 Interval thread shutting down </code></pre> We didn't lose any counter updates here, but from the bumps in interval thread numbers, we can see that we're wasting a lot of time in the main thread checking numbers that aren't changing. Another problem that's less obvious is that we're dedicating an entire OS thread to this sleep-and-check iteration. In our simple program, that's not a big deal. But imagine we decided we wanted to have 50 different similar tasks going on. It would require 49 extra threads, most of which would sit around sleep</code>ing the majority of the time. That's highly wasteful. We should be able to do better. Finally, and perhaps least important for the moment, this is all rather ad hoc. It seems like a common need to be able to abstract over "this thing will produce a value in the future." Even though this seems like the least important problem, we'll start by solving it first. The Future trait</h2> Who'd have thought that our meandering would naturally lead to one of the topics mentioned in this lesson's title! You may have noticed a pattern that developed in the main</code> thread's loop: Check if a new value is available</li> Use it if it is available</li> Skip if it isn't available</li> </ul> That's exactly what the Future</code> trait allows us to do, with one addition: it also allows for error handling. We're not going to worry about that for now, since my code doesn't have any errors :). We'll start by adding the futures</code> crate as a dependency. In Cargo.toml</code>: [dependencies] futures = "0.1" </code></pre> Next, let's add a new module to provide a struct that will provide a Future</code> implementation. Behold, src/future.rs</code>: extern crate futures; use super::interval::Interval; use futures::prelude::*; pub struct IntervalFuture { interval: Interval, last: usize, } impl IntervalFuture { pub fn new(interval: Interval) -> IntervalFuture { let last = interval.get_counter(); IntervalFuture { interval, last } } } impl Future for IntervalFuture { type Item = usize; type Error = (); fn poll(&mut self) -> Poll<Self::Item, Self::Error> { let curr = self.interval.get_counter(); if curr == self.last { Ok(Async::NotReady) } else { self.last = curr; Ok(Async::Ready(curr)) } } } </code></pre> We're going to own an Interval</code> and the last value we provided, just like our main</code> loop used to. The new</code> method is fairly straightforward. For the impl Future</code>, we need to define three things: The thing which will be returned by this type when ready. In our case, it's the counter value, which is a usize</code>.</li> The type of errors that can occur. We don't have any errors, so we use ()</code>. (The Haskeller in me screams that we should use Void</code>. Soon enough, we'll be able to use never</code></a>.)</li> A function poll</code>, which returns a Result</code>. In the error case, this will be our Self::Error</code>. In the success case, this is an Async</code> enum type. As we can see in our method body, this is either a Ready</code> variant with the value, or NotReady</code>.</li> </ul> The logic in our function is the same as before, so I won't comment on implementation. Back in our src/main.rs</code>, instead of playing around with the curr</code>/last</code> logic, we can just pattern match on the result of poll()</code>ing: extern crate futures; mod future; mod interval; use self::interval::Interval; use self::future::IntervalFuture; use futures::prelude::*; fn main() { let interval = Interval::from_millis(500); // half a second let mut interval_future = IntervalFuture::new(interval); let duration = std::time::Duration::from_millis(100); // 0.1 seconds for i in 1..51 { match interval_future.poll() { Ok(Async::Ready(curr)) => { println!("Iteration number {}, counter is {}", i, curr); } Ok(Async::NotReady) => (), Err(()) => unreachable!(), } std::thread::sleep(duration); } } </code></pre> Arguably a minor improvement on the previous code, though nothing major. But congratulations, you're now officially using the futures</code> crate! The Poll type definition</h3> Just a minor helper to mention. The type Result<Async<Self::Item>, Self::Error></code> may look a bit unwieldy to you. If so, you'll be happy to learn about the Poll</code> type definition, which lets you replace the above with Poll<Self::Item, Self::Error></code>. Not a big deal, but important to recognize as you're reading other code. The tokio executor</h2> Right now, we're running our own executor in our main</code> function: we're manually looping, delaying, etc. Besides tedium, we've already mentioned some downsides to this above: We need a single thread per task we wish to perform</li> We need to implement some kind of guess-and-check thread sleeping</li> </ul> It's time to pull out the big guns, and tokio this thing. We'll end up losing some functionality first, and then we'll build it back. First, add tokio = "0.1"</code> to your Cargo.toml</code>. Now, let's try using the tokio executor by calling tokio::run</code> on a Future</code>: extern crate futures; extern crate tokio; mod future; mod interval; use self::interval::Interval; use self::future::IntervalFuture; use tokio::prelude::*; fn main() { let interval = Interval::from_millis(500); // half a second let mut interval_future = IntervalFuture::new(interval); tokio::run(interval_future) } </code></pre> This fails with a compilation error: error[E0271]: type mismatch resolving `<future::IntervalFuture as futures::Future>::Item == ()` --> src/main.rs:15:5 | 15 | tokio::run(interval_future) | ^^^^^^^^^^ expected usize, found () | = note: expected type `usize` found type `()` = note: required by `tokio::run` </code></pre> The tokio::run</code> function expects a Future</code> where the Item</code> is ()</code>, but ours is usize</code>. This kind of makes sense anyway: don't we want to write some code to actually do something when we get a value? We're going to fix this, first the overly painful way, and then the pleasant way. That will also help you appreciate why lesson 5 spent so much time on closures. Define an adapter Future</h3> Remember how you can define Iterator</code>s that consume other Iterator</code>s and compose more powerful streams? Well, you can do the same thing with Future</code>s. Let's define a new type that will: Wrap around an IntervalFuture</code></li> Print the new value whenever it's ready</li> </ul> We'll put this in src/main.rs</code> for now, it won't last long anyway. extern crate futures; extern crate tokio; mod future; mod interval; use self::interval::Interval; use self::future::IntervalFuture; use tokio::prelude::*; struct IntervalPrinter(IntervalFuture); impl Future for IntervalPrinter { type Item = (); type Error = (); fn poll(&mut self) -> Poll<Self::Item, Self::Error> { match self.0.poll() { Ok(Async::Ready(curr)) => { println!("Counter is: {}", curr); Ok(Async::Ready(())) } Ok(Async::NotReady) => Ok(Async::NotReady), Err(e) => Err(e), } } } fn main() { let interval = Interval::from_millis(500); // half a second let interval_future = IntervalFuture::new(interval); let interval_printer = IntervalPrinter(interval_future); tokio::run(interval_printer) } </code></pre> Compile the code, but don't run it yet. This is relatively straight-forward given all of the types and traits we've seen, but it's obviously tedious. Let's start off with a minor simplification. The try_ready macro</h3> The body of poll spends a lot of lines of code to: Pattern match</li> If it's NotReady</code>, return NotReady</code></li> If it's an Err</code>, return the Err</code></li> </ul> This is a repetitive pattern, and is pretty similar to the error handling we saw previously in lesson 3. The futures crate provides a macro, try_ready!</code>, to deal with this annoyance. Add the following above the extern crate futures;</code> in src/main.rs</code>: #[macro_use] </code></pre> And then your implementation of poll</code> can be simplified to: let curr = try_ready!(self.0.poll()); println!("Counter is: {}", curr); Ok(Async::Ready(())) </code></pre> Nice! Compile, and again, don't run. (Getting curious yet why I keep saying that? We'll find out soon, just one more pitstop first.) I need some closure</h3> It's amazing I've made it to lesson 7 in this crash course without making that pun. Obviously, defining an entire struct and Future</code> implementation is a bit overkill to just print a line. Fortunately, the authors of the futures</code> crate noticed this too. There are a number of combinators built into the Future</code> trait that make it easy to chain things together. We're going to use the and_then</code> method: extern crate futures; extern crate tokio; mod future; mod interval; use self::interval::Interval; use self::future::IntervalFuture; use tokio::prelude::*; fn main() { let interval = Interval::from_millis(500); // half a second let interval_future = IntervalFuture::new(interval); let interval_printer = interval_future.and_then(|curr| { println!("Counter is: {}", curr); futures::future::ok(()) }); tokio::run(interval_printer) } </code></pre> That's much nicer! If you're anything like me though, the futures::future::ok(())</code> is bothering you. What purpose does it serve there? This is a vital part of the design of futures, which we'll be taking advantage of quite a bit going forward. It allows us to create chains of actions to run as each bit of async I/O completes. For now, we don't want to do anything else after we print the first value from the counter, so we just return futures::future::ok(())</code>, which means "don't do anything, and return the item ()</code>". Exercise 1 There's another method, .map</code>, which is actually a better choice for us here than .and_then</code>. Try rewriting the code above to use .map</code>. Note: no solution provided to this one. Out of curiosity, what's the type of interval_printer</code>? Let's use the dumb trick from before of giving it the wrong type. Insert something silly like let interval_printer: bool = ...</code> and try compiling, you'll get some type like: futures::AndThen< future::IntervalFuture, futures::FutureResult<(), ()>, [closure@src/main.rs:14:59: 17:6] > </code></pre> If this is starting to look a bit like Iterator</code> types, that's by design. Just like Iterator</code>s, Future</code>s capture a large amount of information in the types themselves about what they'll do. This allows Future</code>s to compile down to highly efficient code, living up to the Rust mantra of zero-cost abstractions. Finally, run it!</h3> Is the suspense killing you? Alright, your moment has arrived. What happens when you run cargo run</code>? $ cargo run Interval thread launched Interval thread shutting down Interval thread still alive, value was: 0 ... </code></pre> And that's it. It hangs. It doesn't print out the message we painstakingly added with and_then</code>. I'm a complete failure, my life is in ruins. After an hour to myself to contemplate my life and a few shots of whiskey, things started to get clear. In our original implementation, we had a really wasteful loop in the main function. It kept checking if the counter had changed. We did that with our Future</code> implementation too at first, sleeping and then checking again for a NotReady</code>. But tokio probably isn't doing that, right? (The answer is yes, right.) Instead of doing the silly wasteful thing, the futures</code> crate is far smarter. It has a mechanism to: determine which task is trying to get access to the data provided by this future, and then</li> notify that task that new data is available</li> </ul> The futures::task::current()</code></a> function gives us the current task being run, as a Task</code></a> struct. That struct has a method, notify</code></a>, to let the task know that more data is available. In our program, we have the logic split between Interval</code> and IntervalFuture</code>. IntervalFuture</code> will need to be responsible for calling the current()</code> function (give some thought as to why that's the case). The changes needed to make this work are: Add a new field to Interval</code> to hold an Arc<Mutex<Option<Task>>></code> (yes, that's a mouthful), and initialize correctly.</li> Each time we call fetch_add</code> and update the counter, also call notify()</code> on that task, if it's there.</li> Provide a method set_task</code> to set the Task</code> on an Interval</code></li> When returning NotReady</code> in IntervalFuture</code>, call set_task</code></li> </ul> Exercise 2 Take a stab at implementing these four changes before looking at the solution below. A hint on some features of Rust we haven't covered yet: you'll end up wanting to pattern match on the Option</code> held inside the Mutex</code>. You'll want to pattern match by reference, which will require some code that looks like Some(ref task) =></code>. And the final output should look like: Interval thread launched Interval thread still alive, value was: 0 Counter is: 1 Interval thread shutting down </code></pre> If you want to be sure, you can see the initial version of the code on Github</a>. Solution 2</h3> You can check out the diff</a> and full solution</a> on Github. Here's the diff included inline: diff --git a/src/future.rs b/src/future.rs index 9aaee3c..e231e7b 100644 --- a/src/future.rs +++ b/src/future.rs @@ -22,6 +22,8 @@ impl Future for IntervalFuture { fn poll(&mut self) -> Result<Async<Self::Item>, Self::Error> { let curr = self.interval.get_counter(); if curr == self.last { + let task = futures::task::current(); + self.interval.set_task(task); Ok(Async::NotReady) } else { self.last = curr; diff --git a/src/interval.rs b/src/interval.rs index 044e2ca..8013ac6 100644 --- a/src/interval.rs +++ b/src/interval.rs @@ -1,12 +1,14 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use std::thread::{sleep, spawn}; use std::time::Duration; +use futures::task::Task; #[derive(Clone)] pub struct Interval { counter: Arc<AtomicUsize>, still_running: Arc<AtomicBool>, + task: Arc<Mutex<Option<Task>>>, } impl Drop for Interval { @@ -26,22 +28,37 @@ impl Interval { let still_running = Arc::new(AtomicBool::new(true)); let still_running_clone = still_running.clone(); + let task: Arc<Mutex<Option<Task>>> = Arc::new(Mutex::new(None)); + let task_clone = task.clone(); + spawn(move || { println!("Interval thread launched"); while still_running_clone.load(Ordering::SeqCst) { sleep(duration); let old = counter_clone.fetch_add(1, Ordering::SeqCst); println!("Interval thread still alive, value was: {}", old); + + let task = task_clone.lock().unwrap(); + match *task { + None => (), + Some(ref task) => task.notify(), + }; } }); Interval { counter, still_running, + task, } } pub fn get_counter(&self) -> usize { self.counter.load(Ordering::SeqCst) } + + pub fn set_task(&mut self, task: Task) { + let mut guard = self.task.lock().unwrap(); + *guard = Some(task); + } } </code></pre> Hurrah, we finally have a working tokio program! In case you're worried about how complex that was, don't be. Notification is a vital aspect of how tokio works internally. However, in most cases, you won't be creating your own primitive Future</code>s, but instead dealing with existing ones provided by tokio or other libraries. Those existing Future</code>s will provide the necessary notification logic. You'll simply need to obey this one rule: Only return a NotReady</code> from a poll</code> function if you received a NotReady</code> from an underlying Future</code>. </blockquote> Just one value?</h2> It's a bit disappointing that our wonderful long running counter only ends up printing a single value. Can we create some kind of a loop? A simple approach like the following doesn't work: let interval_printer = interval_future.and_then(|curr| { println!("Counter is: {}", curr); interval_printer }); </code></pre> This isn't Haskell, we can't recursively refer to interval_printer</code> we're in the middle of defining. Go ahead and take a few other stabs at doing something like that, and you'll eventually get frustrated and go back to the whiskey. Digging through the futures</code> docs, a helper function like loop_fn</code></a> looks promising, but I didn't see a simple way to make it work in this case. (Please let me know if I missed something!) I ended up with something wonky like this before stopping: fn main() { let interval = Interval::from_millis(500); // half a second let interval_future = Arc::new(Mutex::new(IntervalFuture::new(interval))); let interval_printer = loop_fn(interval_future, |interval_future| { let interval_future_clone = interval_future.clone(); interval_future.lock().unwrap().and_then(|curr| { println!("Counter: {}", curr); futures::future::ok(Continue(interval_future_clone)) }) }); tokio::run(interval_printer) } </code></pre>Another struct!</h3> Like before, we're going to define another helper type to implement this concept of looping. Then we'll see that this problem has already been solved better in the futures</code> crate itself, but we'll get there soon. We want to define a new struct, KeepPrinting</code>, which is a newtype around an IntervalFuture</code>. It's going to: Have a Future</code> implementation</li> Have Item = ()</code></li> Use a loop</code> in its implementation</li> Use the try_ready!</code> macro</li> </ul> Exercise 3 Try implementing KeepPrinting</code> and using it in the main</code> function. Solution follows immediately, but try not to cheat! #[macro_use] extern crate futures; extern crate tokio; mod future; mod interval; use self::future::IntervalFuture; use self::interval::Interval; use tokio::prelude::*; struct KeepPrinting(IntervalFuture); impl Future for KeepPrinting { type Item = (); type Error = (); fn poll(&mut self) -> Poll<(), ()> { loop { let curr = try_ready!(self.0.poll()); println!("Counter: {}", curr); } } } fn main() { let interval = Interval::from_millis(500); // half a second let interval_future = IntervalFuture::new(interval); let keep_printing = KeepPrinting(interval_future); tokio::run(keep_printing) } </code></pre> And, just like that, we get an infinitely looping program. This almost looks like something that we could have done with Iterator</code>s. Which makes me wonder... is there something like Iterator</code>s in futures</code>? Streams</h2> A Future</code> is an action with a delayed single result. A Stream</code> is a stream of results, like an Iterator</code>, with a delay between each value. In src/main.rs</code>, add mod stream;</code> and then edit src/stream.rs</code>. The file will end up looking remarkably similar to src/future.rs</code>, except: Call the struct IntervalStream</code> instead of IntervalFuture</code></li> Provide an impl Stream for IntervalStream</code> instead of impl Future</code></li> Follow the compiler errors to fix it</li> </ul> Within the main</code> function, instead of using KeepPrinting</code> or anything else, we'll want to create an IntervalStream</code> value. However, tokio::run</code> needs a Future</code>, not a Stream</code>, to run. Fortunately, there's a helper function, for_each</code>, that runs a given closure</code> on each value in the stream. Exercise 4 Try to implement src/stream.rs</code> and src/main.rs</code>. Solution to follow. The trickiest bit for me when first learning for_each</code> was to realize that, like and_then</code>, it needs to end with a Future</code>. I don't know if that was just my own shortcoming, or a common issue. In any event, if you struggled to realize you needed something like future::ok(())</code> at the end of your closure, you're in good company. In addition, the poll</code> function for a Stream</code> is slightly different, in returning an Option<Item></code>. This is similar to how Iterator</code> works. In our case, we have an infinite stream, so we never provide the None</code> case. Anyway, here's src/stream.rs</code>: extern crate futures; use super::interval::Interval; use futures::prelude::*; pub struct IntervalStream { interval: Interval, last: usize, } impl IntervalStream { pub fn new(interval: Interval) -> IntervalStream { let last = interval.get_counter(); IntervalStream { interval, last } } } impl Stream for IntervalStream { type Item = usize; type Error = (); fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> { let curr = self.interval.get_counter(); if curr == self.last { let task = futures::task::current(); self.interval.set_task(task); Ok(Async::NotReady) } else { self.last = curr; Ok(Async::Ready(Some(curr))) } } } </code></pre> And src/main.rs</code>: extern crate futures; extern crate tokio; mod interval; mod stream; use self::interval::Interval; use self::stream::IntervalStream; use tokio::prelude::*; fn main() { let interval = Interval::from_millis(500); // half a second let interval_stream = IntervalStream::new(interval); let future = interval_stream.for_each(|curr| { println!("Counter: {}", curr); futures::future::ok(()) }); tokio::run(future) } </code></pre> Exercise 5 Like Iterator</code>s, Stream</code>s have helper methods that you can use to build up more complex things. For example, try throwing in map</code> and take</code> to print only the first 10 counter values, but double them before printing. (No solution provided.) This is all beginning to fit together nicely! While there are still details to learn in the futures</code> crate, you've got most of the big ideas down. The next bit is to get familiar with the API in tokio, but relatively speaking this is less mind-bending. To hammer home what we've done so far, we'll hit a few exercises, and then continue with tokio. Exercise 6</h2> Define a new struct MyOk</code> such that this main</code> function works: fn main() { let name = String::from("Alice"); let future = MyOk::new(name).and_then(|name| { println!("Name: {}", name); MyOk::new(()) }); tokio::run(future) } </code></pre> Hint: before cheating and looking at the solution, here's one piece of help: you'll want an Option</code> inside the MyOk</code> newtype, and it's invalid to call poll</code> on it twice. Solution 6</h3> struct MyOk<T>(Option<T>); impl<T> MyOk<T> { fn new(t: T) -> MyOk<T> { MyOk(Some(t)) } } impl<T> Future for MyOk<T> { type Item = T; type Error = (); fn poll(&mut self) -> Poll<T, ()> { Ok(Async::Ready(self.0.take().unwrap())) } } </code></pre>Exercise 7</h2> Use iter_ok</code> to convert the range 1..11</code> to a Stream</code>, and then collect it as a Vec</code> and print it. Solution 7</h3> fn main() { tokio::run(stream::iter_ok(1..11).collect().and_then(|x| { println!("{:?}", x); future::ok(()) })) } </code></pre>Async I/O</h2> We've played around with the futures</code> crate by creating a fake async I/O source of data (the Interval</code>). We've built up Future</code>s and Stream</code>s in that world. And we've used tokio's executor to run these things. It's now time to take it to use some real async I/O. Most async I/O we care about will end up being network traffic. Filesystem operations don't always play nicely with async I/O at an operating system level. That said, to get our feet wet, let's play with a filesystem based example. You'll want to look at the docs quite a bit, you can find them on docs.rs</a>. List files in a directory</h2> If you look through the docs above, you may find the function read_dir</code></a>. It takes a path, and returns a ReadDirFuture</code>. This is a standard approach in tokio, like we had with Iterator</code>s: simple wrapper functions providing access to the structs that do the heavy lifting. One thing to get used to in tokio is how to read these docs. Click through on the ReadDirFuture</code> struct. It has a Future</code> implementation, where Item</code> is ReadDir</code>, and Error</code> is std::io::Error</code>. Before we deal with that ReadDir</code>, let's just get something that compiles. Since this is still a crash course, we'll bash our heads against a brick wall each step of the way. First, to call read_dir</code>, we need a directory. Let's use "."</code> (the current directory). We'll use command line arguments later. Here's a naive implementation: extern crate tokio; use tokio::prelude::*; use tokio::fs; fn main() { let future = fs::read_dir("."); tokio::run(future) } </code></pre> This gives us a somewhat intimidating error message: error[E0271]: type mismatch resolving `<tokio_fs::read_dir::ReadDirFuture<&str> as tokio::prelude::Future>::Item == ()` --> src/main.rs:8:5 | 8 | tokio::run(future) | ^^^^^^^^^^ expected struct `tokio_fs::read_dir::ReadDir`, found () | = note: expected type `tokio_fs::read_dir::ReadDir` found type `()` = note: required by `tokio::run` error[E0271]: type mismatch resolving `<tokio_fs::read_dir::ReadDirFuture<&str> as tokio::prelude::Future>::Error == ()` --> src/main.rs:8:5 | 8 | tokio::run(future) | ^^^^^^^^^^ expected struct `std::io::Error`, found () | = note: expected type `std::io::Error` found type `()` = note: required by `tokio::run` </code></pre> However, let me narrow that down for you: expected type `tokio_fs::read_dir::ReadDir`, found type `()` expected type `std::io::Error`, found type `()` </code></pre> Oh, right! tokio::run</code> requires that we have Item</code> and Error</code> as ()</code>. We can modify the Error</code> with map_err</code>. Let's just print out the error if one occurs: let future = fs::read_dir(".") .map_err(|e| eprintln!("Error reading directory: {}", e)) ; </code></pre> That knocked out the first compilation error. Let's also throw in a .and_then</code>: .and_then(|readdir| { println!("FIXME: use this: {:?}", readdir); }) </code></pre> Uh oh, we got this compilation error. Can you figure out how to solve it? error[E0277]: the trait bound `(): tokio::prelude::Future` is not satisfied </code></pre> In my experience, when you see that, it almost always means: "I forgot to add future::ok(())</code>. Remember, and_then</code> needs to end with the next Future</code> to run. Add that line, and your code should compile. Running produces the output: FIXME: use this: ReadDir(ReadDir(".")) </code></pre> Cool! Now it's time to look at the docs for ReadDir</code></a>. Instead of a Future</code> implementation, this has a Stream</code>. Let's shove a for_each</code> in there and see what happens. Challenge try to guess the errors in the code below before you compile it. .and_then(|readdir| { readdir .for_each(|entry| { println!("{:?}", entry.path()); }) }) </code></pre> There are two problems with this code: It leaves an error type of std::io::Error</code></li> It doesn't include future::ok(())</code> at the end of the closure provided to for_each</code></li> </ol> Go ahead and fix those problems. To be sure we're on the same page, here's my solution which compiles and runs successfully: extern crate tokio; use tokio::prelude::*; use tokio::fs; fn main() { let future = fs::read_dir(".") .map_err(|e| eprintln!("Error reading directory: {}", e)) .and_then(|readdir| { readdir .map_err(|e| eprintln!("Error reading directory: {}", e)) .for_each(|entry| { println!("{:?}", entry.path()); future::ok(()) }) }) ; tokio::run(future) } </code></pre>Duplicated error handling</h3> It's a bit irritating that we have two identical map_err</code> calls. We have two different sources of errors: the initial read_dir</code> Future</code>, and then streaming the individual DirEntry</code>s from it. However, the type of the errors in both cases is the same: std::io::Error</code>. Therefore, we can move our error handling to the end, and just do it once: fn main() { let future = fs::read_dir(".") .and_then(|readdir| { readdir .for_each(|entry| { println!("{:?}", entry.path()); future::ok(()) }) }) .map_err(|e| eprintln!("Error reading directory: {}", e)) ; tokio::run(future) } </code></pre>Flattening</h3> It turns out that it's common enough to have a Future</code> that generates another Future</code>, and then we want to run that second Future</code>, that there's a helper method for it flatten()</code>. There's also a flatten_stream()</code></a> that does the same thing when a Future</code> gives us a Stream</code>. Exercise 8 Rewrite the code above to use flatten_stream</code>. You should end up with no calls to and_then</code>. Solution follows immediately: extern crate tokio; use tokio::prelude::*; use tokio::fs; fn main() { let future = fs::read_dir(".") .flatten_stream() .for_each(|entry| { println!("{:?}", entry.path()); future::ok(()) }) .map_err(|e| eprintln!("Error reading directory: {}", e)) ; tokio::run(future) } </code></pre>Command line arguments</h3> It's somewhat boring to always print out what's in the current directory. Instead, let's take all of the command line arguments (skipping the first, which is the executable name), and list the directory contents. We'll use stream::iter_ok</code> to convert the Args</code> Iterator</code> into a Stream</code>: extern crate tokio; use tokio::prelude::*; use tokio::fs; use std::env::args; fn main() { let future = stream::iter_ok(args()) .skip(1) .for_each(|dir| { fs::read_dir(dir) .flatten_stream() .for_each(|entry| { println!("{:?}", entry.path()); future::ok(()) }) }) .map_err(|e| eprintln!("Error reading directory: {}", e)) ; tokio::run(future) } </code></pre> Unfortunately, this doesn't compile. The full error message is large (I encourage you to check it out yourself), but the first few lines are sufficient to find the problem: error[E0277]: `std::env::Args` cannot be sent between threads safely --> src/main.rs:20:5 | 20 | tokio::run(future) | ^^^^^^^^^^ `std::env::Args` cannot be sent between threads safely </code></pre> Oh, darn. Args</code> isn't thread safe, and so cannot be converted into a Stream</code>. Fair enough: vectors to the rescue! Exercise 9 Create a vector of arguments before the definition of future</code> and use that. Solution: extern crate tokio; use tokio::prelude::*; use tokio::fs; use std::env::args; fn main() { let args: Vec<String> = args().skip(1).collect(); let future = stream::iter_ok(args) .for_each(|dir| { fs::read_dir(dir) .flatten_stream() .for_each(|entry| { println!("{:?}", entry.path()); future::ok(()) }) }) .map_err(|e| eprintln!("Error reading directory: {}", e)) ; tokio::run(future) } </code></pre>Where's the concurrency?</h3> If you provide this program two different directories with a large number of files, you may notice that it processes these directories sequentially: it will print all of the files in the first directory, and then all of the files in the second directory. Given that async I/O and concurrency usually go hand-in-hand, that may be a bit surprising. So far, we've only ever had a single task at a time. Our program streams out the value of args</code>, and for each one provides a Future</code>. That Future</code> is run to completion, and then the next value from args</code> is processed. What if we want to process each directory concurrently? To do that, we need to spawn another task, the same way we would spawn a new thread. Like tokio::run</code>, tokio::spawn</code> takes a Future</code> where both Item</code> and Error</code> are ()</code>. Here's a more concurrent version of our program: let future = stream::iter_ok(args) .for_each(|dir| { let future = fs::read_dir(dir) .flatten_stream() .map_err(|e| eprintln!("Error reading directory: {}", e)) .for_each(|entry| { println!("{:?}", entry.path()); future::ok(()) }) ; tokio::spawn(future); future::ok(()) }) ; </code></pre> Notice how I've put future::ok(())</code> after the tokio::spawn(future);</code> call. It turns out that's not needed: spawn</code> returns a Spawn</code> value, which behaves like future::ok(())</code> (via its IntoFuture</code> implementation). So just remove future::ok</code> and the semicolon after spawn</code>, and your code will still work. NOTE You may not notice the concurrency unless you have a large number of files in each directory. Skipping the vector</h3> One final thing that annoyed me above is that Vec</code>. It really seems like we should be able to get away without it. We can't convert Args</code> into a Stream</code>, because that would require sending the value between threads. But now we've got a new trick up our sleeves: spawning. What if we never send the Args</code> anywhere, but just spawn a bunch of tasks. When I was first learning tokio, I'll admit that I spent way more time trying to figure out the trick I'm about to show you than I'm proud of. We need to create a Future</code>, then run a for</code> loop inside of it. How do we create a Future</code> that lets us run some code without waiting for anything else? We can use future::ok(())</code> to create a dummy Future</code>, and then chain the next action together with and_then</code>. let future = future::ok(()).and_then(|()| { for dir in args().skip(1) { let future = fs::read_dir(dir) .flatten_stream() .map_err(|e| eprintln!("Error reading directory: {}", e)) .for_each(|entry| { println!("{:?}", entry.path()); future::ok(()) }) ; tokio::spawn(future); } future::ok(()) }); </code></pre> Another approach, if you're so inclined, is to use the future::poll_fn</code></a> helper function. This takes a 0 argument function which returns a Result<Async<Item>, Error></code>, just like the poll</code> method of Future</code> does. Exercise 10 Rewrite our program above to use future::poll_fn</code>. Your program should not use and_then</code> at all. Solution: extern crate tokio; use tokio::prelude::*; use tokio::fs; use std::env::args; fn main() { let future = future::poll_fn(|| { for dir in args().skip(1) { let future = fs::read_dir(dir) .flatten_stream() .map_err(|e| eprintln!("Error reading directory: {}", e)) .for_each(|entry| { println!("{:?}", entry.path()); future::ok(()) }) ; tokio::spawn(future); } Ok(Async::Ready(())) }); tokio::run(future) } </code></pre> Is all of this ceremony worth it for file system operations? Probably not. Let's get into something more interesting: network communications! TCP client</h2> Rust's standard library already provides really nice support for TCP communications out of the box. For example, the following code will print the full response headers and body from an HTTP request to http://httpbin.org/json: use std::io::{Read, Write}; use std::net::TcpStream; fn main() -> Result<(), Box<std::error::Error>> { let mut stream = TcpStream::connect("httpbin.org:80")?; stream.write_all(b"GET /json HTTP/1.1\r\nHost: httpbin.org\r\nConnection: close\r\n\r\n")?; let mut buffer = vec![]; stream.read_to_end(&mut buffer)?; println!("{}", std::str::from_utf8(&buffer)?); Ok(()) } </code></pre> There are some simplifying assumptions here, like using connection: close</code> so that we can use read_to_end</code>, and assuming the response body is properly UTF-8 encoded. But that's not really an indictment of the TCP support in the standard library. The real problem is the same one we've been talking about throughout this lesson: it hogs an entire OS thread blocking on a successful write to and subsequent read from the network. Let's look at how tokio can help. It looks like there's a TcpStream</code> type in tokio::net::TcpStream</code></a>. That looks like a good place to start. It takes a SocketAddr</code></a>, which we can probably make easily enough. And it returns a ConnectFuture</code></a>. Let's start with some code that simple establishes a connection: extern crate tokio; use tokio::net::TcpStream; use tokio::prelude::*; use std::net::ToSocketAddrs; fn main() { let mut addr_iter = "httpbin.org:80".to_socket_addrs().unwrap(); let addr = match addr_iter.next() { None => panic!("DNS resolution failed"), Some(addr) => addr, }; let future = TcpStream::connect(&addr) .map_err(|e| eprintln!("Error connecting: {:?}", e)) .map(|stream| { println!("Got a stream: {:?}", stream); }); tokio::run(future) } </code></pre> The to_socket_addrs</code> business isn't our focus right now, so I'm going to ignore it. Feel free as an exercise to improve the error handling of that bit of code. We've got all of the familiar pieces here: define a Future</code>, handle errors, and use map</code> to chain together the action to take with the open connection. Let's look a bit more closely at that stream</code> value passed to the closure. It comes from a ConnectFuture</code>, so we need to look at the Item</code> associated type there. And sure enough, if you check the docs, you'll see that it's TcpStream</code>. Great. We used write_all</code> in our original, non-async, blocking code. If I search for write_all</code> in tokio</a>, I find that there's such a helper function</a> which returns a WriteAll</code></a> Future</code>. Something interesting: write_all</code> takes two parameters: an AsyncWrite</code> and an AsRef<[u8]></code>. This will work out to be our TcpStream</code> and the data to send.</li> The Item</code> for AsyncWrite</code> is a pair of the variables A</code> and T</code>, which turns out to be exactly the same as the parameters we passed in.</li> </ul> Giving us back the stream we originally provided is vital to continuing the connection. I didn't see any documentation clarifying the point of returning the byte buffer, but I believe it's returned so that, if desired, you can reuse a mutable buffer. Anyway, let's put this together and make our request: extern crate tokio; use tokio::net::TcpStream; use tokio::io::write_all; use tokio::prelude::*; use std::net::ToSocketAddrs; const REQ_BODY: &[u8] = b"GET /json HTTP/1.1\r\nHost: httpbin.org\r\nConnection: close\r\n\r\n"; fn main() { let mut addr_iter = "httpbin.org:80".to_socket_addrs().unwrap(); let addr = match addr_iter.next() { None => panic!("DNS resolution failed"), Some(addr) => addr, }; let future = TcpStream::connect(&addr) .and_then(|stream| { write_all(stream, REQ_BODY) .map(|(stream, _body)| println!("Write succeeded: {:?}", stream)) }) .map_err(|e| eprintln!("Error occured: {:?}", e)) ; tokio::run(future) } </code></pre> Notice how I replaced the previous map</code> with an and_then</code> call, so that I could provide another Future</code> to be performed after the connection was established. Would it be too much to ask to also get a read_to_end</code> function in tokio? Nope, not at all.</a> Exercise 11 Use read_to_end</code> to consume the entire response into a Vec<u8></code>, and then print that out, using std::str::from_utf8</code> and being as careless with error handling as you like. Alright, solution: extern crate tokio; use std::net::ToSocketAddrs; use tokio::io::{read_to_end, write_all}; use tokio::net::TcpStream; use tokio::prelude::*; const REQ_BODY: &[u8] = b"GET /json HTTP/1.1\r\nHost: httpbin.org\r\nConnection: close\r\n\r\n"; fn main() { let mut addr_iter = "httpbin.org:80".to_socket_addrs().unwrap(); let addr = match addr_iter.next() { None => panic!("DNS resolution failed"), Some(addr) => addr, }; let future = TcpStream::connect(&addr) .and_then(|stream| { write_all(stream, REQ_BODY) }).and_then(|(stream, _body)| { let buffer = vec![]; read_to_end(stream, buffer) }).map(|(_stream, buffer)| { let s = std::str::from_utf8(&buffer).unwrap(); println!("{}", s); }).map_err(|e| eprintln!("Error occured: {:?}", e)); tokio::run(future) } </code></pre>Streaming to a file</h2> I'll repeat for maximum annoyance: tokio is not intended for asynchronous file operations. That said, there is a tokio::fs::File</code> struct which we can use. Let's try to write the response contents to httpbin.json</code> instead: let future = TcpStream::connect(&addr) .and_then(|stream| { write_all(stream, REQ_BODY) }).and_then(|(stream, _body)| { let buffer = vec![]; read_to_end(stream, buffer) }).and_then(|(_stream, buffer)| { File::create("httpbin.json").and_then(|file| { write_all(file, &buffer).map(|_| ()) }) }).map_err(|e| eprintln!("Error occured: {:?}", e)); </code></pre> Unfortunately, the compiler doesn't like this too much: error[E0277]: the trait bound `std::fs::File: tokio::io::AsyncWrite` is not satisfied --> src/main.rs:25:17 | 25 | write_all(file, &buffer).map(|_| ()) | ^^^^^^^^^ the trait `tokio::io::AsyncWrite` is not implemented for `std::fs::File` | = note: required by `tokio::io::write_all </code></pre> Well, I guess that makes sense: you can't asynchronously write to a File</code>, so tokio::io::write_all</code> isn't going to work. Fortunately, File</code> does implement the Write</code> trait, which provides a blocking write_all</code>, which is sufficient for our purposes. Exercise 12 Rewrite the code above to successfully write httpbin.json</code>. First solution, ignoring anything close to proper error handling: let future = TcpStream::connect(&addr) .and_then(|stream| { write_all(stream, REQ_BODY) }).and_then(|(stream, _body)| { let buffer = vec![]; read_to_end(stream, buffer) }).and_then(|(_stream, buffer)| { File::create("httpbin.json").map(|mut file| { file.write_all(&buffer).unwrap() }) }).map_err(|e| eprintln!("Error occured: {:?}", e)); </code></pre> But ideally, we'd like to avoid that unwrap()</code> and instead promote an I/O error here to be handle by the map_err</code> below. It turns out that there's a surprisingly trivial change to make that happen: File::create("httpbin.json").and_then(|mut file| { file.write_all(&buffer) }) </code></pre> Instead of using map</code>, we use and_then</code>, which requires that we return some value that implements Future</code>. But fortunately, Result</code> itself implements Future</code>! The Ok</code> variant becomes the Item</code> for that Future</code>, and the Err</code> variant becomes its Error</code>. Problem solved! Exercise 13</h2> We haven't taken advantage of tokio here at all! Let's make this program concurrent. Write a program that takes command line arguments to determine HTTP requests to make and files to store them to. To simplify the implementation, we'll have it take input that looks like the following: $ cargo run httpbin.org:80 /json httpbin.json example.com:80 / homepage.html </code></pre> Feel free to handle invalid command line arguments however's easiest. Solution</h3> extern crate tokio; use std::net::ToSocketAddrs; use tokio::io::{read_to_end, write_all}; use tokio::net::TcpStream; use tokio::prelude::*; use std::fs::File; fn download(host: String, path: String, filename: String) -> impl Future<Item=(), Error=()> { let mut addr_iter = host.to_socket_addrs().unwrap(); let addr = match addr_iter.next() { None => panic!("DNS resolution failed"), Some(addr) => addr, }; let req_body = format!( "GET {} HTTP/1.1\r\nHost: {}:80\r\nConnection: close\r\n\r\n", path, host, ); TcpStream::connect(&addr) .and_then(|stream| { write_all(stream, req_body).and_then(|(stream, _body)| { let buffer = vec![]; read_to_end(stream, buffer).and_then(|(_stream, buffer)| { File::create(filename).and_then(|mut file| { file.write_all(&buffer) }) }) }) }).map_err(|e| eprintln!("Error occured: {:?}", e)) } fn main() { tokio::run(future::poll_fn(|| { let mut args = std::env::args().skip(1); loop { match (args.next(), args.next(), args.next()) { (Some(host), Some(path), Some(filename)) => { tokio::spawn(download(host, path, filename)); } _ => return Ok(Async::Ready(())), } } })) } </code></pre>Nicer error handling</h2> We're just panic!</code>ing when we have a bad address. Let's do a little bit better. First, I'll define a helper function to return a nice Result</code>. We'll use a String</code> for the Err</code> variant, but we could should ideally define an enum</code> instead: fn resolve_addr(host: &str) -> Result<SocketAddr, String> { let mut addr_iter = match host.to_socket_addrs() { Ok(addr_iter) => addr_iter, Err(e) => return Err(format!("Invalid host name {:?}: {:?}", host, e)), }; match addr_iter.next() { None => Err(format!("No addresses found for host: {:?}", host)), Some(addr) => Ok(addr), } } </code></pre> Inside download</code>, we could continue panic!</code>ing with: let addr = resolve_addr(&host).unwrap(); </code></pre> But let's do better. Using ?</code> won't work, since we aren't returning a Result</code>. One idea would be to use return</code> to return early: let addr = match resolve_addr(&host) { Ok(addr) => addr, Err(e) => { eprintln!("Error resolving address: {}", e); return future::err(()); } }; </code></pre> However, we get an interesting error message from the compiler: error[E0308]: mismatched types --> src/main.rs:34:5 | 34 | / TcpStream::connect(&addr) 35 | | .and_then(|stream| { 36 | | write_all(stream, req_body).and_then(|(stream, _body)| { 37 | | let buffer = vec![]; ... | 43 | | }) 44 | | }).map_err(|e| eprintln!("Error occured: {:?}", e)) | |___________________________________________________________^ expected struct `tokio::prelude::future::FutureResult`, found struct `tokio::prelude::future::MapErr` </code></pre> In order to make things work, we need to ensure that we always return the same type. We've so far used impl Future</code> to say "we'll return some type which is a Future</code>," but we haven't told the compiler what that type is. Instead, the compiler has inferred that. But now, we have two different types. One approach would be dynamic dispatch, such as using Box<Future></code>. But there's a better way: using the Either</code> helper type. This type is used in a case where we have two different types of Future</code>s which both have the same Item</code> and Error</code>. Let's see how we can rewrite our code above to use Either</code>: fn download(host: String, path: String, filename: String) -> impl Future<Item=(), Error=()> { let addr = match resolve_addr(&host) { Ok(addr) => addr, Err(e) => { eprintln!("Error resolving address: {}", e); return future::Either::A(future::err(())); } }; let req_body = format!( "GET {} HTTP/1.1\r\nHost: {}:80\r\nConnection: close\r\n\r\n", path, host, ); future::Either::B(TcpStream::connect(&addr) .and_then(|stream| { write_all(stream, req_body).and_then(|(stream, _body)| { let buffer = vec![]; read_to_end(stream, buffer).and_then(|(_stream, buffer)| { File::create(filename).and_then(|mut file| { file.write_all(&buffer) }) }) }) }).map_err(|e| eprintln!("Error occured: {:?}", e))) } </code></pre> Exercise 14 Implement your own Either</code> data type and use it in the code above. Solution: enum Either<A, B> { A(A), B(B), } impl<A, B> Future for Either<A, B> where A: Future<Item=B::Item, Error=B::Error>, B: Future, { type Item = A::Item; type Error = A::Error; fn poll(&mut self) -> Poll<Self::Item, Self::Error> { match self { Either::A(a) => a.poll(), Either::B(b) => b.poll(), } } } </code></pre>TCP server</h2> Having been so successful with our TCP client, let's move over to the server side. Conceptually, we want to: Bind a listening socket</li> Accept connections from that socket</li> Copy all data from the input side of the socket to the output side of the socket</li> </ol> Binding a listening socket is going to be a blocking call</a> to bind</code>, taking our old friend SocketAddr</code>. Since we're not playing around with DNS resolution anymore, we can be a bit lazier about how we do this: extern crate tokio; use tokio::prelude::*; use tokio::net::TcpListener; fn main() { let addr = "127.0.0.1:3000".parse().expect("couldn't parse address"); let listener = TcpListener::bind(&addr).expect("couldn't bind address"); println!("It worked! {:?}", listener); } </code></pre> We now have have a TcpListener</code>. Unlike other types we've seen, this doesn't have an implementation of either Future</code> or Stream</code>. However, it does have a method called incoming()</code>, which returns an Incoming</code>, which has a Stream</code> implementation, where Item</code> is TcpStream</code>. That looks promising! let future = listener .incoming() .for_each(|socket| { println!("Accepted a connection! {:?}", socket); future::ok(()) }) .map_err(|e| eprintln!("An error occurred: {:?}", e)) ; tokio::run(future) </code></pre> And just like that, we've implemented points (1) and (2) above. We're just left with point 3: copying all of the data. Let's search tokio for something to do copying</a>. It looks like tokio::io::copy</code></a> will do. We need to provide it both a reader and writer. Since we're reader from and writing to the same socket, let's just provide the same value for both: let future = listener .incoming() .for_each(|socket| { copy(socket, socket) .map(|_| println!("Connection closed")) }) .map_err(|e| eprintln!("An error occurred: {:?}", e)) ; </code></pre> Are you already laughing at my comically silly mistake? error[E0382]: use of moved value: `socket` --> src/main.rs:13:26 | 13 | copy(socket, socket) | ------ ^^^^^^ value used here after move | | | value moved here </code></pre> Of course we can't use the same value in both positions. Fortunately, when designing Stream</code>s, the authors provided a method called split</code></a> to give us a read and write end of the stream. With that in hand, our echo server becomes trivial: extern crate tokio; use tokio::prelude::*; use tokio::net::TcpListener; use tokio::io::copy; fn main() { let addr = "127.0.0.1:3000".parse().expect("couldn't parse address"); let listener = TcpListener::bind(&addr).expect("couldn't bind address"); let future = listener .incoming() .for_each(|socket| { let (reader, writer) = socket.split(); copy(reader, writer) .map(|_| println!("Connection closed")) }) .map_err(|e| eprintln!("An error occurred: {:?}", e)) ; tokio::run(future) } </code></pre>Writing directly</h2> Using copy</code> kind of ignores the gory details of what's going on under the surface. Let's start off by writing some arbitrary message to the writer side of things, using the write_all</code> function</a>. NOTE The tokio::io::write_all</code> function takes a AsyncWrite</code> and returns a WriteAll</code> Future</code>. Don't be confused by the presence of a write_all</code> method, which in fact is a blocking call. I wasted about 5 minutes fumbling with that while writing this tutorial. extern crate tokio; use tokio::prelude::*; use tokio::net::TcpListener; use tokio::io::{copy, write_all}; fn main() { let addr = "127.0.0.1:3000".parse().expect("couldn't parse address"); let listener = TcpListener::bind(&addr).expect("couldn't bind address"); let future = listener .incoming() .for_each(|socket| { let (reader, writer) = socket.split(); write_all(writer, b"Welcome to the echo server\r\n") .map(|_| println!("Connection closed")) }) .map_err(|e| eprintln!("An error occurred: {:?}", e)) ; tokio::run(future) } </code></pre> Exercise 15 Modify the code above so that, after printing "Welcome to the echo server", it proceeds to actually echo content sent in. write_all(writer, b"Welcome to the echo server\r\n") .and_then(|(writer, _)| { copy(reader, writer) .map(|_| println!("Connection closed")) }) </code></pre>Codecs</h2> Reading from the reader</code> directly is slightly trickier than writing to the writer</code>. We could go play around with the underlying polling reading functions, but we're not going to here. (Feel free to read the official tokio tutorial for more information</a>.) Instead, we're going to introduce a new concept, codecs. So far, we've implicitly been working with the AsyncRead</code> and AsyncWrite</code> traits, which essentially provide the raw polling functions we'd need for building up our own Future</code>s (as we did way long ago at the beginning of this lesson). However, we often don't want to work at that level of abstraction. Instead, we'd like to deal with some kind of framed (or chunked) data. The new abstraction instead will be a Sink</code>, which is "a value into which other values can be sent, asynchronously." We'll continue to use the Stream</code> trait for the read side, which we're already quite familiar with. Let's contrive an example. Our echo server currently provides slightly weird output: Hello Hello There There World World </code></pre> It's hard to tell what I typed in, and what the server responded. Instead, I'd like each line sent back from the server to begin with "You said: ". Doing that with the abstractions we've seen so far would be fairly painful: we'd need to grab chunks of data, look for the newline character, break up the input, splice in the "You said: " message. I know this is a crash course and all, but I'd rather not crash into that. Instead, let's jump straight to the better solution. I want to treat our TCP stream as a stream of lines of data. If I search for the word "lines" (and this is actually how I learned about codecs), I end up with LinesCodec</code></a>. It provides a method new()</code>, as well as new_with_max_length</code>. We'll use new</code> here, but I recommend reading the docs to see why that's a terrible idea in any kind of security sensitive context. The only other method on the type is max_length</code>, which doesn't look like it's going to help us actually deal with a TCP socket as a stream of lines. So let's look down at the trait implementations. We've got all of our usual suspects: Clone</code>, PartialOrd</code>, etc. But two new ones stick out: Decoder</code> and Encoder</code>. Well that certainly looks interesting. Reading through the docs on Decoder</code>, it provides a method called framed</code></a>, which has a description that is great. (Please, take a second to follow that link and read the docs.) Without further ado, let's try adding in our LinesCodec</code> to our echo server: extern crate tokio; use tokio::prelude::*; use tokio::net::TcpListener; use tokio::codec::{Decoder, LinesCodec}; fn main() { let addr = "127.0.0.1:3000".parse().expect("couldn't parse address"); let listener = TcpListener::bind(&addr).expect("couldn't bind address"); let future = listener .incoming() .for_each(|socket| { let lines_codec = LinesCodec::new(); let socket = lines_codec.framed(socket); socket .send(String::from("Welcome to the echo server")) .map(|_| println!("Connection closed")) }) .map_err(|e| eprintln!("An error occurred: {:?}", e)) ; tokio::run(future) } </code></pre> You may have noticed that we no longer have a newline sequence at the end of the "Welcome" string. That's because our lines codec automatically handles that. Additionally, we now need to use String::from</code>, since the Item</code> for this Sink</code> is a String</code>. We can also use split</code> to isolate the Sink</code> from the Stream</code>: let (sink, stream) = lines_codec.framed(socket).split(); sink .send(String::from("Welcome to the echo server")) .map(|_| println!("Connection closed")) </code></pre> And, we can use for_each</code> on the Stream</code> side to get a stream of the lines: extern crate tokio; use tokio::prelude::*; use tokio::net::TcpListener; use tokio::codec::{Decoder, LinesCodec}; fn main() { let addr = "127.0.0.1:3000".parse().expect("couldn't parse address"); let listener = TcpListener::bind(&addr).expect("couldn't bind address"); let future = listener .incoming() .for_each(|socket| { let lines_codec = LinesCodec::new(); let (sink, stream) = lines_codec.framed(socket).split(); sink .send(String::from("Welcome to the echo server")) .and_then(|sink| { stream .for_each(|line| { println!("Received a line: {}", line); future::ok(()) }) .map(|_| println!("Connection closed")) }) }) .map_err(|e| eprintln!("An error occurred: {:?}", e)) ; tokio::run(future) } </code></pre> We're almost done here: we just need to send</code> the lines back to the sink</code> instead of to stdout</code>. Unfortunately, using the send</code> method we've seen so far is going to be tricky, since we'll end up consuming the sink</code> in each iteration of for_each</code>. We could figure out a way to make that all work, but instead, let's just cut to the chase and use send_all</code>. Exercise 16 Modify the code above so that, instead of printing the lines to standard output, they get sent back to the client with the message "You said: ". You'll want to look at send_all</code></a> Solution extern crate tokio; use tokio::prelude::*; use tokio::net::TcpListener; use tokio::codec::{Decoder, LinesCodec}; fn main() { let addr = "127.0.0.1:3000".parse().expect("couldn't parse address"); let listener = TcpListener::bind(&addr).expect("couldn't bind address"); let future = listener .incoming() .for_each(|socket| { let lines_codec = LinesCodec::new(); let (sink, stream) = lines_codec.framed(socket).split(); sink .send(String::from("Welcome to the echo server")) .and_then(|sink| { let stream = stream .map(|line| format!("You said: {}", line)) ; sink.send_all(stream) .map(|_| println!("Connection closed")) }) }) .map_err(|e| eprintln!("An error occurred: {:?}", e)) ; tokio::run(future) } </code></pre>Next time</h2> Whew, that was a big lesson! At this point, you should have a very solid footing in the ins and outs of tokio. It's time to get lots more experience with using the library, and related higher level libraries for doing things like HTTP servers and clients. Depending on reader feedback, the next lesson may either go deeper into tokio and related libraries, or go back to more fundamental aspects of Rust like lifetimes. From the tokio side, we'd play with: Message passing between tasks</li> UDP communications</li> Recursive directory traversal</li> Parallel downloading of files</li> </ul> Rust at FP Complete</a> | Introduction</a>

Solution 3</h3>
Since the `loop</code> will either run forever or be short circuited by an error, any code following loop</code> will never actually be called. Therefore, code placed there will generate a warning.</p>`

Exercise 7</h2>
Use `iter_ok</code> to convert the range 1..11</code> to a Stream</code>, and then collect it as a Vec</code> and print it.</p>`

Arrays, slices, vectors, and String</h2>
Here's another place where we've been cheeky. What's the difference between `String</code> and str</code>? Both of these have popped up quite a bit. We'll get to those in a little bit. First, we need to talk about arrays, slices, and vectors.</p>`

Snoyberg - Michael Snoyman - rust-crash-course

Tokio 0.2 - Rust Crash Course lesson 9

Down and dirty with Future - Rust Crash Course lesson 8

Async, futures, and tokio - Rust Crash Course lesson 7

Lifetimes and Slices - Rust Crash Course lesson 6 - exercise solutions

Lifetimes and Slices - Rust Crash Course lesson 6

Rule of Three - Parameters, Iterators, and Closures - Rust Crash Course lesson 5 - exercise solutions

Rule of Three - Parameters, Iterators, and Closures - Rust Crash Course lesson 5

Crates and more iterators - Rust Crash Course lesson 4 - exercise solutions

Crates and more iterators - Rust Crash Course lesson 4

Iterators and Errors - Rust Crash Course lesson 3 - exercise solutions

Snoyberg - Michael Snoyman - rust-crash-course

Tokio 0.2 - Rust Crash Course lesson 9

Spawning processes</h2> Tokio provides a tokio::process</code> module which resembles the std::process</code> module. We can use this to implement Hello World once again:</p>

Time to spawn</h2> This is all well and good, but we're not really taking advantage of asynchronous programming at all. Let's fix that! We've seen two different interesting programs:</p> Infinitely pausing 1 seconds and calling date</code></li>

Solution 3</h3> Since the loop</code> will either run forever or be short circuited by an error, any code following loop</code> will never actually be called. Therefore, code placed there will generate a warning.</p>

Down and dirty with Future - Rust Crash Course lesson 8

Async, futures, and tokio - Rust Crash Course lesson 7

Interval</h2> We're going to call this thing an Interval</code>, since it somewhat represents a setInterval</code> call in Javascript. Go ahead and start a new project:</p>

Exercise 7</h2> Use iter_ok</code> to convert the range 1..11</code> to a Stream</code>, and then collect it as a Vec</code> and print it.</p>

TCP client</h2> Rust's standard library already provides really nice support for TCP communications out of the box. For example, the following code will print the full response headers and body from an HTTP request to http://httpbin.org/json:</p>

Streaming to a file</h2> I'll repeat for maximum annoyance: tokio is not intended for asynchronous file operations. That said, there is</em> a tokio::fs::File</code> struct which we can use. Let's try to write the response contents to httpbin.json</code> instead:</p>

Lifetimes and Slices - Rust Crash Course lesson 6 - exercise solutions

Exercise 2</h2> You need to provide mutable references for the two arguments to swap</code>. Additionally, in order to get a mutable reference to res</code>, res</code> itself needs to be mut</code>able.</p>

Exercise 3</h2> We need to have two different parameters, and ensure that ret</code> and the return value have the same lifetime parameter:</p>

Lifetimes and Slices - Rust Crash Course lesson 6

Arrays, slices, vectors, and String</h2> Here's another place where we've been cheeky. What's the difference between String</code> and str</code>? Both of these have popped up quite a bit. We'll get to those in a little bit. First, we need to talk about arrays, slices, and vectors.</p>

Vec</h3> A Vec</code> is a "contiguous, growable array type." You can push</code> and pop</code>, check its length, and access via index in O(1) time. We also have a nifty vec!</code> macro for constructing them:</p>

Slices</h3> I'm going to write a helper function that prints all the values in a Vec</code>:</p>

Byte literals</h2> If you put a lower case b</code> in front of a string literal, you'll get a byte array. You can either treat this as a fixed length array or, more commonly, as a slice:</p>

References and slices in APIs</h2> Some general advice which I received and has mostly steered me correctly is:</p> When receiving parameters, prefer slices when possible</p> </blockquote>

Rule of Three - Parameters, Iterators, and Closures - Rust Crash Course lesson 5 - exercise solutions

Exercise 3</h2> The closure version:</p>

Exercise 5</h2> The first error message we get is:</p>

Rule of Three - Parameters, Iterators, and Closures - Rust Crash Course lesson 5

Mutable vs immutable pass-by-value</h3> This one is relatively easy to see. What extra functionality do we get by having a mutable pass-by-value? The ability to mutate the value of course! Let's look at two different ways of implementing a birthday function, which increases someone's age by 1.</p>

Mutable to immutable</h3> Let's point out one final bit:</p>

Summary of the rule of three for parameters</h3> There are three types of parameters:</p> Pass by value</li> Pass by immutable reference</li>

Summary of the rule of three for iterators</h3> Just like function parameters, iterators come in three flavors, corresponding to the following naming scheme:</p>

Exercise 7</h2> The error handling in this program is lackluster. There are three problems:</p> If gtk::init()</code> fails, the exit code of our program is still 0</code> (indicating success).</li> If opening mylog.txt</code> fails, we panic.</li>

Introducing spawn</h3> The simplest way to introduce spawn is to replace the inner()</code> call with spawn(inner)</code>. Replace:</p>

Crates and more iterators - Rust Crash Course lesson 4 - exercise solutions

Solution 3</h3>
Since the `loop</code> will either run forever or be short circuited by an error, any code following loop</code> will never actually be called. Therefore, code placed there will generate a warning.</p>`

Exercise 7</h2>
Use `iter_ok</code> to convert the range 1..11</code> to a Stream</code>, and then collect it as a Vec</code> and print it.</p>`

Arrays, slices, vectors, and String</h2>
Here's another place where we've been cheeky. What's the difference between `String</code> and str</code>? Both of these have popped up quite a bit. We'll get to those in a little bit. First, we need to talk about arrays, slices, and vectors.</p>`