An example of lightweight (Green) application threads using the continuation monad

GreenThreads.scala

/**
 * An example of lightweight (Green) application threads using the continuation monad,
 * which gives us scalable non-blocking IO without nested callbacks.
 * @author willtim
 */
object GreenThreads {

  // a thread (continuation) represents the rest of the computation
  sealed abstract class Thread
  case class Print(str: String, rest: Thread) extends Thread
  case class Fork(t1: Thread, t2: Thread) extends Thread
  case object End extends Thread

  case class CM[A](cont: (A => Thread) => Thread) { ma =>
    def map[B](f: A => B): CM[B] =
      CM[B](k => ma.cont(x => k(f(x))))

    def flatMap[B](f: A => CM[B]): CM[B] =
      CM[B](k => ma.cont(x => f(x).cont(k)))
  }

  def thread[A](cm: CM[A]): Thread = cm.cont(_ => End)

  // atomic operations
  def cPrint(str: String): CM[Unit] = CM(k => Print(str, k()))
  def cFork[A](m: CM[A]): CM[Unit] = CM(k => Fork(thread(m), k()))
  def cEnd: CM[Unit] = CM(k => End)

  // running a computation
  type Output = List[String]
  type ThreadQueue = List[Thread]
  type State = (Output, ThreadQueue)

  def runCM[A](m: CM[A]): Output = dispatch(Nil, thread(m))

  def dispatch(q: ThreadQueue, t: Thread): Output = t match {
    case Print(c, t) => c :: schedule(q ++ List(t))
    case Fork(t1, t2) => schedule(q ++ List(t1, t2))
    case End => schedule(q)
  }

  def schedule: ThreadQueue => Output = {
    case Nil => Nil
    case t :: ts => dispatch(ts, t)
  }

  def p1: CM[Unit] = for {
    _ <- cPrint("Hello")
    _ <- cPrint("World")
  } yield ()

  def p2: CM[Unit] = for {
    _ <- cPrint("1")
    _ <- cPrint("2")
  } yield ()

  def p3: CM[Unit] = for {
    _ <- cPrint("A")
    _ <- cFork(p1)
    _ <- cPrint("B")
    _ <- cFork(p2)
    _ <- cPrint("C")
  } yield ()

  def main(args: Array[String]) {
    println {
      runCM(p3)
    }
  }
}