Control-flow combinators do not require thread-...

Control-flow combinators do not require thread-local variable in
Reactor. Review by plocinic.

7 files changed, 105 insertions, 49 deletions

diff --git a/src/actors/scala/actors/Actor.scala b/src/actors/scala/actors/Actor.scala
index 99a23c980f..2d672adb97 100644
--- a/src/actors/scala/actors/Actor.scala
+++ b/src/actors/scala/actors/Actor.scala
@@ -22,7 +22,7 @@ import java.util.concurrent.{ExecutionException, Callable}
  *
  * @author Philipp Haller
  */
-object Actor {
+object Actor extends Combinators {
 
   private[actors] val tl = new ThreadLocal[Reactor]
 
@@ -276,26 +276,6 @@ object Actor {
   }
 
   /**
-   * Causes <code>self</code> to repeatedly execute
-   * <code>body</code>.
-   *
-   * @param body the code block to be executed
-   */
-  def loop(body: => Unit): Unit = body andThen loop(body)
-
-  /**
-   * Causes <code>self</code> to repeatedly execute
-   * <code>body</code> while the condition
-   * <code>cond</code> is <code>true</code>.
-   *
-   * @param cond the condition to test
-   * @param body the code block to be executed
-   */
-  def loopWhile(cond: => Boolean)(body: => Unit): Unit =
-    if (cond) { body andThen loopWhile(cond)(body) }
-    else continue
-
-  /**
    * Links <code>self</code> to actor <code>to</code>.
    *
    * @param  to the actor to link to
@@ -350,7 +330,6 @@ object Actor {
    */
   def exit(): Nothing = self.exit()
 
-  def continue: Unit = throw new KillActorException
 }
 
 /**
@@ -690,13 +669,16 @@ trait Actor extends AbstractActor with ReplyReactor with ReplyableActor {
     // Note that we do *not* reset `trapExit`. The reason is that
     // users should be able to set the field in the constructor
     // and before `act` is called.
-
     exitReason = 'normal
     exiting = false
     shouldExit = false
 
-    scheduler.newActor(this)
-    scheduler.execute(new Reaction(this))
+    scheduler newActor this
+    val task = new Reaction(this)
+    if (Actor.rawSelf(scheduler).isInstanceOf[ActorProxy])
+      scheduler execute task
+    else
+      scheduler executeFromActor task
 
     this
   }
@@ -751,7 +733,7 @@ trait Actor extends AbstractActor with ReplyReactor with ReplyableActor {
   }
 
   var trapExit = false
-  private[actors] var exitReason: AnyRef = 'normal
+  private var exitReason: AnyRef = 'normal
   private[actors] var shouldExit = false
 
   /**
diff --git a/src/actors/scala/actors/ActorTask.scala b/src/actors/scala/actors/ActorTask.scala
index 8d8504245f..cf4283f7de 100644
--- a/src/actors/scala/actors/ActorTask.scala
+++ b/src/actors/scala/actors/ActorTask.scala
@@ -17,14 +17,15 @@ package scala.actors
  *
  *  @author Philipp Haller
  */
-private[actors] class ActorTask(actor: Actor, fun: () => Unit) extends ReactorTask[Actor](actor, fun) {
+private[actors] class ActorTask(actor: Actor, fun: () => Unit) extends ReplyReactorTask[Actor](actor, fun) {
 
-  protected override def beforeExecuting() {
+  protected override def beginExecution() {
+    super.beginExecution()
     if (actor.shouldExit)
       actor.exit()
   }
 
-  protected override def afterExecuting(e: Exception) {
+  protected override def terminateExecution(e: Exception) {
     actor.synchronized {
       if (!actor.links.isEmpty)
         actor.exitLinked(e)
diff --git a/src/actors/scala/actors/Combinators.scala b/src/actors/scala/actors/Combinators.scala
new file mode 100644
index 0000000000..073247827c
--- /dev/null
+++ b/src/actors/scala/actors/Combinators.scala
@@ -0,0 +1,39 @@
+/*                     __                                               *\
+**     ________ ___   / /  ___     Scala API                            **
+**    / __/ __// _ | / /  / _ |    (c) 2005-2010, LAMP/EPFL             **
+**  __\ \/ /__/ __ |/ /__/ __ |    http://scala-lang.org/               **
+** /____/\___/_/ |_/____/_/ | |                                         **
+**                          |/                                          **
+\*                                                                      */
+
+// $Id$
+
+package scala.actors
+
+private[actors] trait Combinators {
+
+  implicit def mkBody[a](body: => a): Actor.Body[a]
+
+  /**
+   * Causes <code>self</code> to repeatedly execute
+   * <code>body</code>.
+   *
+   * @param body the code block to be executed
+   */
+  def loop(body: => Unit): Unit = body andThen loop(body)
+
+  /**
+   * Causes <code>self</code> to repeatedly execute
+   * <code>body</code> while the condition
+   * <code>cond</code> is <code>true</code>.
+   *
+   * @param cond the condition to test
+   * @param body the code block to be executed
+   */
+  def loopWhile(cond: => Boolean)(body: => Unit): Unit =
+    if (cond) { body andThen loopWhile(cond)(body) }
+    else continue
+
+  def continue: Unit = throw new KillActorException
+
+}
diff --git a/src/actors/scala/actors/Reactor.scala b/src/actors/scala/actors/Reactor.scala
index a5cc26949a..a6007a6cd8 100644
--- a/src/actors/scala/actors/Reactor.scala
+++ b/src/actors/scala/actors/Reactor.scala
@@ -45,7 +45,7 @@ private[actors] object Reactor {
  *
  * @author Philipp Haller
  */
-trait Reactor extends OutputChannel[Any] {
+trait Reactor extends OutputChannel[Any] with Combinators {
 
   /* The actor's mailbox. */
   private[actors] val mailbox = new MQueue("Reactor")
@@ -181,7 +181,6 @@ trait Reactor extends OutputChannel[Any] {
   }
 
   protected[actors] def react(f: PartialFunction[Any, Unit]): Nothing = {
-    assert(Actor.rawSelf(scheduler) == this, "react on channel belonging to other actor")
     synchronized { drainSendBuffer(mailbox) }
     searchMailbox(mailbox, f, false)
     throw Actor.suspendException
@@ -195,18 +194,20 @@ trait Reactor extends OutputChannel[Any] {
    * never throws SuspendActorException
    */
   private[actors] def scheduleActor(handler: PartialFunction[Any, Any], msg: Any) = {
-    val fun = () => handler(msg)
-    val task = new ReactorTask(this, fun)
-    scheduler executeFromActor task
+    val fun = () => handler(msg): Unit
+    scheduler executeFromActor makeReaction(fun)
   }
 
   def start(): Reactor = {
-    scheduler.newActor(this)
-    val task = new ReactorTask(this, () => act())
-    scheduler execute task
+    scheduler newActor this
+    scheduler execute makeReaction(() => act())
     this
   }
 
+  implicit def mkBody[A](body: => A) = new Actor.Body[A] {
+    def andThen[B](other: => B): Unit = Reactor.this.seq(body, other)
+  }
+
   /* This closure is used to implement control-flow operations
    * built on top of `seq`. Note that the only invocation of
    * `kill` is supposed to be inside `ReactorTask.run`.
@@ -216,10 +217,9 @@ trait Reactor extends OutputChannel[Any] {
     () => { exit() }
 
   private[actors] def seq[a, b](first: => a, next: => b): Unit = {
-    val s = Actor.rawSelf(scheduler)
-    val killNext = s.kill
-    s.kill = () => {
-      s.kill = killNext
+    val killNext = this.kill
+    this.kill = () => {
+      this.kill = killNext
 
       // to avoid stack overflow:
       // instead of directly executing `next`,
diff --git a/src/actors/scala/actors/ReactorTask.scala b/src/actors/scala/actors/ReactorTask.scala
index aab7d6b72f..3e64eab7a2 100644
--- a/src/actors/scala/actors/ReactorTask.scala
+++ b/src/actors/scala/actors/ReactorTask.scala
@@ -8,7 +8,6 @@
 
 // $Id$
 
-
 package scala.actors
 
 import java.lang.Runnable
@@ -26,10 +25,8 @@ private[actors] class ReactorTask[T >: Null <: Reactor](var reactor: T, var fun:
   extends RecursiveAction with Callable[Unit] with Runnable {
 
   def run() {
-    val saved = Actor.tl.get
-    Actor.tl set reactor
     try {
-      beforeExecuting()
+      beginExecution()
       try {
         try {
           fun()
@@ -50,9 +47,9 @@ private[actors] class ReactorTask[T >: Null <: Reactor](var reactor: T, var fun:
         Debug.info(reactor+": caught "+e)
         Debug.doInfo { e.printStackTrace() }
         reactor.terminated()
-        afterExecuting(e)
+        terminateExecution(e)
     } finally {
-      Actor.tl set saved
+      suspendExecution()
       this.reactor = null
       this.fun = null
     }
@@ -62,8 +59,10 @@ private[actors] class ReactorTask[T >: Null <: Reactor](var reactor: T, var fun:
 
   def compute() = run()
 
-  protected def beforeExecuting() {}
+  protected def beginExecution() {}
+
+  protected def suspendExecution() {}
 
-  protected def afterExecuting(e: Exception) {}
+  protected def terminateExecution(e: Exception) {}
 
 }
diff --git a/src/actors/scala/actors/ReplyReactor.scala b/src/actors/scala/actors/ReplyReactor.scala
index d5936ae662..05e4590162 100644
--- a/src/actors/scala/actors/ReplyReactor.scala
+++ b/src/actors/scala/actors/ReplyReactor.scala
@@ -95,4 +95,7 @@ trait ReplyReactor extends Reactor with ReplyableReactor {
     }
   }
 
+  private[actors] override def makeReaction(fun: () => Unit): Runnable =
+    new ReplyReactorTask(this, fun)
+
 }
diff --git a/src/actors/scala/actors/ReplyReactorTask.scala b/src/actors/scala/actors/ReplyReactorTask.scala
new file mode 100644
index 0000000000..934a87435b
--- /dev/null
+++ b/src/actors/scala/actors/ReplyReactorTask.scala
@@ -0,0 +1,32 @@
+/*                     __                                               *\
+**     ________ ___   / /  ___     Scala API                            **
+**    / __/ __// _ | / /  / _ |    (c) 2005-2010, LAMP/EPFL             **
+**  __\ \/ /__/ __ |/ /__/ __ |                                         **
+** /____/\___/_/ |_/____/_/ | |                                         **
+**                          |/                                          **
+\*                                                                      */
+
+// $Id$
+
+package scala.actors
+
+/** <p>
+ *    The class <code>ReplyReactorTask</code>.
+ *  </p>
+ *
+ *  @author Philipp Haller
+ */
+private[actors] class ReplyReactorTask[T >: Null <: ReplyReactor](reactor: T, fun: () => Unit) extends ReactorTask[ReplyReactor](reactor, fun) {
+
+  var saved: Reactor = _
+
+  protected override def beginExecution() {
+    saved = Actor.tl.get
+    Actor.tl set reactor
+  }
+
+  protected override def suspendExecution() {
+    Actor.tl set saved
+  }
+
+}