How to split and dispatch an async control-flow using Continuations?

Question

I have an asynchronous control-flow like the following:

ActorA ! DoA(dataA, callback1, callbackOnErrorA)

def callback1() = {
  ...
  ActorB ! DoB(dataB, callback2, callbackOnErrorB)
}

def callback2() = {
  ActorC ! DoC(dataC, callback3, callbackOnErrorC)
} 

...

How would I divide this flow into several parts (continuations) and sequentially dispatch these to different actors (or threads/tasks) while maintaining the overall state?

Any hint appreciated, Thanks

Tosha

Answer 1

This is very simplified, but shows how to split up a single control flow among three actors, passing the state along to each:

package blevins.example

import scala.continuations._
import scala.continuations.ControlContext._
import scala.actors.Actor._
import scala.actors._

object App extends Application {

  val actorA, actorB, actorC = actor {
    receive {
      case f: Function1[Unit,Unit] => { f() }
    }
  }

  def handle(a: Actor) = shift { k: (Unit=>Unit) =>
    a ! k
  }

  // Control flow to split up
  reset {
      // this is not handled by any actor
      var x = 1
      println("a: " + x)

      handle(actorA)  // actorA handles the below
      x += 4
      println("b: " + x)

      handle(actorB) // then, actorB handles the rest
      var y = 2
      x += 2
      println("c: " + x)

      handle(actorC) // and so on...
      y += 1
      println("d: " + x + ":" + y)
  }

}

Engstrom

Answer 2

I like to use scalaz.concurrent.Promise. This example isn't exactly like the one in your question, but it gives you the idea.

object Async extends Application {
  import scalaz._
  import Scalaz._
  import concurrent._
  import concurrent.strategy._
  import java.util.concurrent.{ExecutorService, Executors}

  case class ResultA(resultb: ResultB, resulta: ResultC)
  case class ResultB()
  case class ResultC()

  run

  def run {
    implicit val executor: ExecutorService = Executors.newFixedThreadPool(8)
    import Executor.strategy

    val promiseA = doA
    println("waiting for results")
    val a: ResultA = promiseA.get
    println("got " + a)
    executor.shutdown    
  }

  def doA(implicit s: Strategy[Unit]): Promise[ResultA] = {
    println("triggered A")
    val b = doB
    val c = doC
    for {bb <- b; cc <- c} yield ResultA(bb, cc)
  }

  def doB(implicit s: Strategy[Unit]): Promise[ResultB] = {
    println("triggered B")
    promise { Thread.sleep(1000); println("returning B"); ResultB() }
  }

  def doC(implicit s: Strategy[Unit]): Promise[ResultC] = {
    println("triggered C")
    promise { Thread.sleep(1000); println("returning C"); ResultC() }
  }
}

Output:

triggered A
triggered B
triggered C
waiting for results
returning B
returning C
got ResultA(ResultB(),ResultC())

You'll find an introduction to Scalaz concurrency in this presentation from Runar.

This approach isn't as flexible as Actors, but composes better and can't deadlock.

Accentuate

Nice Scalaz Promise-Example, thank you. However I want to get a deeper understanding of the new Scala2.8-CPS thing and would appreciate a CPS-specific answer. – Tosha

+1 for mentioning the advantages of Futures over Actors, and the use of implicit vals for defining the strategies. – Synclastic

Answer 3

See Akka's Futures and how to compose them or scalaz's Promises, they are nearly the same, there are only slight differences.

Synclastic