Return to chapter index

7 Customizing Actor Execution

7.1 Pluggable schedulers

your name: joe jr.
your name: joe jr.


your name: joe jr.
your name: john


object ActorWithThreadLocalWrong extends Application {
  val tname = new ThreadLocal[String] {
    override protected def initialValue() = "john"
  }
  val joeActor = actor {
    react { case 'YourName =>
      tname set "joe jr."
      sender ! tname.get
      react { case 'YourName =>
        sender ! tname.get
      }
    }
  }
  actor {
    println("your name: " + (joeActor !? 'YourName))
    println("your name: " + (joeActor !? 'YourName))
  }
}


abstract class ActorWithThreadLocal(private var name: String)
  extends Actor {
  override val scheduler = new SchedulerAdapter {
    def execute(codeBlock: => Unit): Unit =
      ActorWithThreadLocal.super.scheduler execute {
        tname set name
        codeBlock
        name = tname.get
      }
  }
}


abstract class SwingActor extends Actor {
  override val scheduler = new SchedulerAdapter {
    def execute(codeBlock: => Unit): Unit =
      java.awt.EventQueue.invokeLater(
        new Runnable() {
          def run() = codeBlock
        }
      )
  }
}


import scala.actors.Actor
import scala.actors.scheduler.DaemonScheduler

object DaemonActors {

  class MyDaemon extends Actor {
    override def scheduler = DaemonScheduler
    def act() {
      loop {
        react { case num: Int => reply(num + 1) }
      }
    }
  }

  def main(args: Array[String]) {
    val d = (new MyDaemon).start()
    println(d !? 41)
  }
}


  case class SyncGear(s: Int)
  case class SyncDone(g: Gear)

  class Gear(val id: Int, var speed: Int, val controller: Actor)
      extends Actor {

    def act() {
      loop {
        react {
          case SyncGear(targetSpeed: Int) =>
            println("[Gear "+id+
                    "] synchronize from current speed " + speed +
                    " to target speed " + targetSpeed)
            adjustSpeedTo(targetSpeed)
        }
      }
    }

    def adjustSpeedTo(targetSpeed: Int) {
      if (targetSpeed > speed) {
        speed += 1
        self ! SyncGear(targetSpeed)
      } else if (targetSpeed < speed) {
        speed -= 1
        self ! SyncGear(targetSpeed)
      } else if (targetSpeed == speed) {
        println("[Gear " + id + "] has target speed")
        controller ! SyncDone(this)
        exit()
      }
    }
  }


object NonDeterministicGears {
  def main(args: Array[String]) {
    actor {
      val g1 = (new Gear(1, 7, self)).start()
      val g2 = (new Gear(2, 1, self)).start()
      g1 ! SyncGear(5)
      g2 ! SyncGear(5)
      react { case SyncDone(_) =>
        react { case SyncDone(_) => }
      }
    }
  }
}


[Gear 2] synchronize from current speed 1 to target speed 5
[Gear 2] synchronize from current speed 2 to target speed 5
[Gear 1] synchronize from current speed 7 to target speed 5
[Gear 1] synchronize from current speed 6 to target speed 5
[Gear 1] synchronize from current speed 5 to target speed 5
[Gear 2] synchronize from current speed 3 to target speed 5
[Gear 1] has target speed
[Gear 2] synchronize from current speed 4 to target speed 5
[Gear 2] synchronize from current speed 5 to target speed 5
[Gear 2] has target speed


  

7.2 Managed blocking

import scala.actors.Actor._
import java.util.concurrent.CountDownLatch

object PoolLockup {

  def main(args: Array[String]) {
    val numCores = Runtime.getRuntime().availableProcessors()
    println("available cores: " + numCores)

    val latch = new CountDownLatch(1)
    for (i <- 1 to (numCores * 2)) actor {
      latch.await()
      println("actor " + i + " done")
    }

    actor { latch.countDown() }
  }

}


import scala.actors.Actor
import scala.actors.Actor._
import scala.concurrent.ManagedBlocker
import java.util.concurrent.CountDownLatch

object ManagedBlocking {

  class BlockingActor(i: Int, latch: CountDownLatch)
        extends Actor {
    val blocker = new ManagedBlocker {
      def block() = { latch.await(); true }
      def isReleasable = { latch.getCount == 0 }
    }
    def act() {
      scheduler.managedBlock(blocker)
      println("actor " + i + " done")
    }
  }

  def main(args: Array[String]) {
    val numCores = Runtime.getRuntime().availableProcessors()
    println("available cores: " + numCores)

    val latch = new CountDownLatch(1)
    for (i <- 1 to (numCores * 2))
      (new BlockingActor(i, latch)).start()

    actor { latch.countDown() }
  }

}