Python 3 asyncio

A while back I wrote a few posts about asynchronous programming:

• Asynchronous Non-blocking I/O Java Echo Server
• Fixing My Asynchronous Non-blocking I/O Callback Hell With Monads
• Scala Asynchronous IO Echo Server
• Reactive Actors

So when I learned that Python 3.5 has added async and await operations I knew I had to check it out to see how it compares. asyncio describes itself as "infrastructure for writing single-threaded concurrent code using coroutines, multiplexing I/O access over sockets and other resources, running network clients and servers, and other related primitives".

Coroutines in Python are similar to generators, but coroutines (a function definition using async def) can control where execution continues after the yield (replaced by await). You await a another coroutine or a future. The coroutine approach can replace callbacks.

If we check the time it takes for the program to run, it's 3 seconds (the longest slow operation) and not 6 seconds (the sum of all the slow operations).

Slow operation sleep 1 complete

Slow operation sleep 2 complete

Slow operation sleep 3 complete

Completed in 3.02 seconds

It's similar if we want to get results from each task. We're even able to get them as they are available. 

Slow operation sleep 1 complete

Got result 1

Slow operation sleep 2 complete

Got result 2

Slow operation sleep 3 complete

Got result 3

Completed in 3.00 seconds

With Python's GIL you've never really been able to run multiple threads in parallel. You've had to run concurrent code in multiple processes to leverage multiple CPU cores. With asyncio we are able to at least make single process IO-bound tasks execute faster because it switches between tasks to bypass GIL contention.

For CPU-bound code you still have to use multiple processes to parallelize your code. Python's parallel API limits your ability to use results from these tasks as they become available, as we did in the example above. Waiting for a process to finish and getting a future result both block. asyncio gives us a way to unify the concurrent and parallel APIs.

Slow operation sum 10000000 complete

Got result 49999995000000

Slow operation sum 20000000 complete

Got result 199999990000000

Slow operation sum 30000000 complete

Got result 449999985000000

Completed in 2.91 seconds

There is some overhead in creating the processes but a single 3 second task is only slightly faster than parallel 1, 2, and 3 second tasks.

Slow operation sum 30000000 complete

Got result 449999985000000

Completed in 2.66 seconds

This hints at another use. Given that asyncio uses a single thread, if there are too many IO-bound tasks or if any of them consumes too much CPU, we can overwhelm it. For such a situation it is possible to create multiple processes, each with its own asyncio event loop.

1 process and 3 tasks:

Got 3 results in 6.20 seconds

1 process and 15 tasks:

Got 15 results in 23.94 seconds

The simulated mix of IO-bound and CPU-bound code is interesting. It's slower than just the CPU-bound code on it's own, but faster than the sum of the IO-bound and CPU-bound code. We see some asyncio benefits up to around 15 tasks and then it levels off.

1 process and 30 tasks:

Got 30 results in 46.58 seconds

4 processes and 15*4 tasks:

Got 60 results in 34.11 seconds

8 processes and 15*8 tasks:

Got 120 results in 52.81 seconds

We similarly see some benefit of parallelizing the code up the number of CPU cores I have, then the time is increasing linearly as we would expect.

16 processes and 15*16 tasks:

Got 120 results in 105.98 seconds

If you are looking for more details the comprehensive Python Concurrency series of articles has additional examples like this and goes into more detail on some of the underlying concepts.

Reactive Actors

I've been meaning to revisit reactive programming and the actor model for a while now. I first learned about them in the Principles of Reactive Programming Coursera class and then actors came up again in the Seven Concurrency Models in Seven Weeks book. The Scala I picked up is quickly being forgotten and I haven't done a post with code in a while, so here I'll get back into that and create a simple application using Akka and RxScala.

Actor model

The developerWorks article JVM Concurrency: Acting asynchronously with Akka gives a good introduction to the actor model:

The actor model for concurrent computations builds up systems based on primitives called actors. Actors take actions in response to inputs called messages. Actions can include changing the actor's own internal state as well as sending off other messages and even creating other actors. All messages are delivered asynchronously, thereby decoupling message senders from receivers. Because of this decoupling, actor systems are inherently concurrent: Any actors that have input messages available can be executed in parallel, without restriction.

Then JVM Concurrency: Building actor applications with Akka goes on to explain the advantages of this approach:

If you compose your actors and messages correctly, you end up with a system in which most things happen asynchronously. Asynchronous operation is harder to understand than a linear approach, but it pays off in scalability. Highly asynchronous programs are better able to use increased system resources (for example, memory and processors) either to accomplish a particular task more quickly or to handle more instances of the task in parallel. With Akka, you can even extend this scalability across multiple systems, by using remoting to work with distributed actors.

At first the actor model may sound the same as what I described in my Communicating Sequential Processes post because both involve message passing, but the two concurrency models have several differences:

• Actors have identities while CSPs are anonymous
• Actors transmit messages to named actors while CSPs transmit messages using channels
• Actors transmit messages asynchronously while CSPs can't transmit a message until the sender is ready to receive it

My impression, and I could be wrong, is that actors more naturally extend beyond a single machine to a distributed system since the sending and receiving of messages is decoupled. A quick search does turn up distributed channels in pycsp, though, so it seems that both can be distributed.

Reactive applications

The Reactive Manifesto details four qualities of reactive applications:

• responsive - the system responds in a timely manner if at all possible
• resilient - the system stays responsive in the face of failure
• elastic - the system stays responsive under varying workloads
• message driven - the system relies on asynchronous message passing between components

Where Akka describes itself as a toolkit and runtime, RxScala only claims to be a library for composing asynchronous and event-based programs using observable sequences. To me it's not clear how it helps us achieve all four qualities (or if it even intends to).  Nevertheless, the ReactiveX introduction explains their advantages:

The ReactiveX Observable model allows you to treat streams of asynchronous events with the same sort of simple, composable operations that you use for collections of data items like arrays. It frees you from tangled webs of callbacks, and thereby makes your code more readable and less prone to bugs.

This means is that the methods returning Observables can be implemented using thread pools, non-blocking I/O, actors, or anything else. This is how ReactiveX and Akka will be used together: Actors are the concurrency implementation for services communicating with asynchronous messages.

Combining Akka and RxScala

I came up with the following short example. First I wrote a couple of methods returning an Observable to get a feel for it, then added the stockQuote() method which also uses an actor in it's implementation:


Running it produces the expected output, something like:

6

8

broken service

GOOG: 253.22

I can really see the potential in the Observable model, especially after reading more about it at The Netflix Tech Blog. If you were already using actors maybe combining them like this could make sense. I also need to checkout Akka Streams which seems like a similar idea.

UPDATE

Ray is a framework for parallelizing ML workloads. They use the actor model as a way of coordinating work and maintaining state.

Communicating Sequential Processes: Goroutines and Channels

This post, like Software Transactional Memory: Dining Philosophers, is motivated by the book Seven Concurrency Models in Seven Weeks. One of the concurrency models discussed is communicating sequential processes (CSP). Even though it was completely new to me, the idea has been around for several decades. I've wanted to check out Go for a while now and since it's a language whose design was influenced by CSP, it seems like a good time to write a Go program.

For concurrent programming, Go encourages shared values to be passed around on channels instead of by sharing memory. The value can only be accessed by one goroutine at a time. Unbuffered channels combine communication with synchronization, and buffered channels can be used like a semaphore. A goroutine is a function executing concurrently with other goroutines in the same address space. It's multiplexed onto multiple OS threads so a blocking goroutine doesn't hold up other goroutines.

In addition to the Seven Concurrency Models in Seven Weeks book, I think the Clojure core.async Channels blog sums up the motivation for channels well. Basically, they are an alternative to using queues to communicate between different components and to using events/callbacks. You avoid avoid thread overhead and callback hell.

I wrote a short program for a vending machine where money deposits and soda dispenses are values passed on channels. It was more difficult than expected to think this way, but the two goroutines only communicate over channels which was the intent.

Software Transactional Memory: Dining Philosophers

Software transactional memory (STM) is a concurrency alternative to lock-based synchronization. It is similar to database transactions except instead of modifying database records, you are modifying variables in shared memory. STM transactions are atomic (a transaction is all or nothing), consistent (all changes made from a transaction must preserve invariants), and isolated (concurrent transactions result in the same system state as transactions executed serially). Since they are in-memory they are not durable.

With STM you are separating identity from state with persistent data structures. While state is constantly changing, the current state for an identity is immutable. Changes made by one thread will not affect another with a reference to the same data structure.¹

Without locks multiple threads can also modify different parts of a data structure that would normally share the same lock. Optimistic concurrency control assumes multiple transactions can run in parallel and if wrong, they will be retried.²

Example Code

Motivated by the book Seven Concurrency Models In Seven Weeks, I have implemented the dining philosophers problem in Scala. The book uses Clojure, but ScalaSTM is inspired by the STMs in Haskell and Clojure and I am trying to get better with Scala, so this seemed like a good way to reinforce what I was reading. It turns out that a dining philosophers solution is in the ScalaSTM documentation, but nonetheless it was a worthwhile exercise.

Conclusion

Atomic variables are enough for a lot of problems and functional programming discourages mutable data anyway, but if it is simpler to use multiple mutable variables then STM might be the way to go. As always, you need to compare performance of both though.

1 http://clojure.org/state
2 https://nbronson.github.io/scala-stm/intro.html

Scala Asynchronous IO Echo Server

In previous posts Asynchronous Non-blocking I/O Java Echo Server and Fixing My Asynchronous Non-blocking IO Callback Hell With Monads, I compared using callbacks and futures/promises as concurrency alternatives to creating using a thread pool in a basic echo server. I have now decided to rewrite the server from the second post in Scala since it has better support for futures and promises than Java 8.

Tthe idiomatic way to write an echo server in Scala would be to use the actor model, but that is a topic for another day. My implementation using futures and promises still turned out to be faster than the similar Java 8 implementation (in fact roughly as fast as the true asynchronous NIO implementation) and the code is more concise.

There is still the need to block on accept, read, and write, but I also still think that is unavoidable with the AsynchronousSocketChannel API.

Java 7 Fork/Join and Java 8 Streams: Merge Sort

Fork/Join

The fork/join framework from Java 7 has been on my list of things to check out for a while now. Basically, it provides an easy way to parallelize work that can be broken up into smaller parts. Work is distributed to threads in a pool, and idle threads can steal work from busy threads (work stealing). There are many parallel algorithms and I decided to try parallel merge sort. I have two processors, so I would expect to see some speedup over the canonical top-down merge sort.


My fork/join parallel implementation does not become faster than the non-parallel implementation until I try to sort over about 25 million integers. It is obvious fork/join comes with significant overhead, but I suspect that would be less noticeable if I had more processors to work with.

Streams

As I have been trying to become more adept at functional programming and am also a Java fan, I have been starting to explore what's new in Java 8 as well. This probably is not at all practical, but I had the idea to try to re-implement merge sort using streams. To do this I had to use the non-recursive bottom-up merge sort algorithm.

This is faster than the fork/join version up to about 3 million elements, then it becomes slower. It was not really clear to me how to properly parallelize this one. Converting it to a parallel stream before doing the merge seems to only slightly improve performance. Again, maybe it is difficult to really see the benefit with only two processors.

It is also worth noting that since all parallel streams use the same fork/join thread pool, they will be affected by each other's performance if used concurrently as explained in Java Parallel Streams Are Bad for Your Health.

Notes

For completeness, here is the merge method I used in all three merge sort implementations:

Fixing My Asynchronous Non-blocking I/O Callback Hell With Monads

In my previous post Asynchronous Non-blocking I/O Java Echo Server, I looked at non-blocking I/O from the Java perspective. My echo server used the callback functionality known as a CompletionHandler in Java 7, but it left something to be desired. Mainly I had a callback inside a callback inside a callback. Well, now Java 8 is released and I set out to fix it inspired by Monadic futures in Java 8: How to organize your data and avoid callback hell.

Monads surround computations and can take actions on them. They help chain computations together because they take in an action and then return a new monad surrounding the computations that had the action executed on them.

I am still trying to fully understand monads myself, so I am not going to act like I already do and give detailed information here. I think it's easier to understand from examples, anyway, instead of reading about category theory. The above definition will make more sense after looking at my new server code and seeing the chaining in action.

My New Server Code

A promise is a type of monad common around asynchronous operations. I used the new monadic CompletableFuture class to wrap my asynchronous computations.

My Thoughts

1. I like this better than my Java 7 version in terms of readability, but it would still be a lot cleaner without checked exceptions like in Scala.
2. This is not really non-blocking. Even though it is done asynchronously, I block waiting for a Future to complete three times. There is not a good relationship between Future and CompletableFuture to help avoid this as far as I can tell.
3. This Java 8 version is slower than the Java 7 version. My theory is that, even though the blocking calls are handled asynchronously by the CompletableFuture class, explicitly blocking the thread with the get() instead of letting Java handle it with a callback is what makes the difference.

There is not much Java 8 code out there to look at, so what would make this better? Can I get rid of the blocking get() calls somehow? Why is it slower than the Java 7 version?

Notes

A future represents a result that does not yet exist (and might not ever exist). Futures are read-only. A promise is the container that completes a future. Promises are writable.

UPDATE:

I posted a Scala version of this at Scala Asynchronous IO Echo Server.

Asynchronous Non-blocking I/O Java Echo Server

After writing a Node.js blog a couple weeks ago I wanted to revisit non-blocking I/O. I was first introduced to the topic in a college class where we were using Java, but that was quite a while ago, so I wondered what a non-blocking I/O server looks like in Java these days (and of course Node.js handles a lot of these details for you, so it is probably not the best place to learn about it). Non-blocking I/O was added in Java 1.4 (NIO) and then in Java 7 asynchronous non-blocking I/O (NIO2) was added.

I set out to write a simple echo server that can at least handle simultaneous connections so as to not be completely trivial. I would use the NIO2 APIs so that it is completely asynchronous to be most like the Node.js TCP echo server.

My Server Code


This code is disappointingly difficult to read. I think this is what JavaScript folks call callback hell: three levels of callbacks. Although if you can get through all the boilerplate and exception handling code, you will see I really only had to write a few lines of "real" code which is pretty nice. It is more concise than a similar server written using the older NIO APIs too.

Differences Between Blocking I/O, Non-blocking I/O, and Asynchronous Non-blocking I/O

With blocking I/O when a thread does a read or write it is blocked until some data is read or the data is written. The canonical Java web server example spawns a new thread for each request so that the main thread will not be blocked from accepting new connections.

With non-blocking I/O you request a read and you get what is available (maybe nothing is available) and the thread can continue on. You request a write and whatever can be written is written and the thread continues on.¹ In other words, a single thread can manage multiple connections, but it might have to call read and write multiple times to completely read and write for each request and response.

With asynchronous non-blocking I/O an I/O operation always returns immediately--before anything is read/written--and the thread continues on. The I/O operation is handled in the background and the callback you provide eventually handles the result. This is how I wrote my server shown above.

These real-world analogies are helpful to at least understand the differences between blocking and non-blocking.

Conclusion

Non-blocking I/O is good when you need to manage many long-lived concurrent connections like a chat server or a P2P network. A use case where blocking I/O is probably better is one where more data is transferred at once, but connections are short-lived, like a traditional web server.¹

It also seem like web sockets would be implemented with non-blocking I/O. Can anyone point me to more information about that?

Update

I posted a Java 8 version here.


Resources

1 more info at http://java.dzone.com/articles/java-nio-vs-io

Node.js CRUD API

I have been wanting to check out Node for quite a while now. I am used to the more common OS thread-based concurrency model that we see in Java application servers. Node is instead based on an event-driven, non-blocking I/O model where all connections are handled in one thread.¹ As long as the processing for each request is not CPU-intensive, it can handle thousands of concurrent requests.²

Another more obvious advantage of Node is that your application is written in Javascript. Anyone who has ever written client-side Javascript can immediately begin writing server-side code. You could also, for example, use the same input validation code in your client and server.³

Setup

Never having written a Node app before, I based my first one largely off this great blog Creating a REST API using Node.js, Express, and MongoDB by Christophe Coenraets. I hope to explore MongoDB more in another post. For now I will ignore it and just say that it is simple to install and Chritophe's code to talk to it works as is.

I ran my app on Windows and I did get tripped up on installing the Node MongoDB driver with npm. I had to follow these steps and run "set npm_config_arch=x86" before running "npm mongodb".

The Code

The server is pretty straightforward. Map each of your routes to a function that handles the request and listen on a specific port.

I updated the functions that talk to MongoDB to (I think) be more RESTful.⁴ The non-GET requests will respond with more meaningful HTTP response codes instead of JSON.

Node is able to handle multiple connections in one thread because of the use of callbacks. Any time you do something in your app that may take a long time you, tell Node what to do and provide a function to call when it is finished. It can handle other requests while the long running operation (like reading a file) is completing.

Testing

Bringing in a testing framework would probably be overkill for this simple app, so cURL will suffice. I got it for Windows here. An example for each of the routes:

curl -i -X GET http://localhost:3000/wines
curl -i -X GET http://localhost:3000/wines/<id>
curl -i -X DELETE http://localhost:3000/wines/<id>
curl -i -X POST -H "Content-Type: application/json" -d @new-wine.json http://localhost:3000/wines
curl -i -X PUT -H "Content-Type: application/json" -d @new-wine.json http://localhost:3000/wines/<id>

1 http://nodejs.org/

2 http://blog.caustik.com/2012/08/19/node-js-w1m-concurrent-connections/

3 http://www.theprojectspot.com/tutorial-post/nodejs-for-beginners-callbacks/4

4 http://williamdurand.fr/2012/08/02/rest-apis-with-symfony2-the-right-way/