Dataflow Concurrency (Scala)
Description
Akka implements Oz-style dataflow concurrency by using a special API for Futures (Scala) that enables a complimentary way of writing synchronous-looking code that in reality is asynchronous.
The benefit of Dataflow concurrency is that it is deterministic;, i.
Getting Started (SBT)" % "2.10.0") }, scalacOptions += "-P:continuations:enable",
You will also need to include a dependency on akka-dataflow:
"com.typesafe.akka" %% "akka-dataflow" % "2.1.0"
Dataflow variables
A Dataflow variable can be read any number of times but only be written to once, which maps very well to the concept of Futures/Promises Futures (Scala). Conversion from Future and Promise to Dataflow Variables is implicit and is invisible to the user (after importing akka.dataflow._).
The mapping from Promise and Future is as follows:
- Futures are readable-many, using the apply method, inside flow blocks.
- Promises are readable-many, just like Futures.
- Promises are writable-once, using the << operator, inside flow blocks. Writing to an already written Promise throws a java.lang.IllegalStateException, this has the effect that races to write a promise will be deterministic, only one of the writers will succeed and the others will fail.
The flow
The flow method acts as the delimiter of dataflow expressions (this also neatly aligns with the concept of delimited continuations), and flow-expressions compose. At this point you might wonder what the flow-construct brings to the table that for-comprehensions don't, and that is the use of the CPS plugin that makes the look like it is synchronous, but in reality is asynchronous and non-blocking. The result of a call to flow is a Future with the resulting value of the flow.
To be able to use the flow method, you need to import:
import akka.dataflow._ //to get the flow method and implicit conversions
The flow method will, just like Futures and Promises, require an implicit ExecutionContext in scope. For the examples here we will use:
import scala.concurrent.ExecutionContext.Implicits.global
Using flow
First off we have the obligatory "Hello world!":
flow { "Hello world!" } onComplete println
You can also refer to the results of other flows within flows:
flow { val f1 = flow { "Hello" } f1() + " world!" } onComplete println
… or:
flow { val f1 = flow { "Hello" } val f2 = flow { "world!" } f1() + " " + f2() } onComplete println
Working with variables
Inside the flow method you can use Promises as Dataflow variables:
val v1, v2 = Promise[Int]() flow { // v1 will become the value of v2 + 10 when v2 gets a value v1 << v2() + 10 v1() + v2() } onComplete println flow { v2 << 5 } // As you can see, no blocking above!
Flow compared to for
Should I use Dataflow or for-comprehensions?
val f1, f2 = Future { 1 } val usingFor = for { v1 ← f1; v2 ← f2 } yield v1 + v2 val usingFlow = flow { f1() + f2() } usingFor onComplete println usingFlow onComplete println
Conclusions:
- Dataflow has a smaller code footprint and arguably is easier to reason about.
- For-comprehensions are more general than Dataflow, and can operate on a wide array of types.
Contents