diff --git a/akka-actor-tests/src/test/scala/akka/actor/dispatch/ActorModelSpec.scala b/akka-actor-tests/src/test/scala/akka/actor/dispatch/ActorModelSpec.scala index acc416f04f..4d83c85b82 100644 --- a/akka-actor-tests/src/test/scala/akka/actor/dispatch/ActorModelSpec.scala +++ b/akka-actor-tests/src/test/scala/akka/actor/dispatch/ActorModelSpec.scala @@ -374,7 +374,7 @@ abstract class ActorModelSpec(config: String) extends AkkaSpec(config) with Defa def compare(l: AnyRef, r: AnyRef) = (l, r) match { case (ll: ActorCell, rr: ActorCell) ⇒ ll.self.path compareTo rr.self.path } } foreach { case cell: ActorCell ⇒ - System.err.println(" - " + cell.self.path + " " + cell.isTerminated + " " + cell.mailbox.status + " " + cell.mailbox.numberOfMessages + " " + SystemMessage.size(cell.mailbox.systemDrain())) + System.err.println(" - " + cell.self.path + " " + cell.isTerminated + " " + cell.mailbox.status + " " + cell.mailbox.numberOfMessages + " " + SystemMessage.size(cell.mailbox.systemDrain(null))) } System.err.println("Mailbox: " + mq.numberOfMessages + " " + mq.hasMessages) diff --git a/akka-actor/src/main/scala/akka/actor/ActorSystem.scala b/akka-actor/src/main/scala/akka/actor/ActorSystem.scala index 008610c333..af7313b41e 100644 --- a/akka-actor/src/main/scala/akka/actor/ActorSystem.scala +++ b/akka-actor/src/main/scala/akka/actor/ActorSystem.scala @@ -545,7 +545,7 @@ private[akka] class ActorSystemImpl(val name: String, applicationConfig: Config, becomeClosed() def systemEnqueue(receiver: ActorRef, handle: SystemMessage): Unit = deadLetters ! DeadLetter(handle, receiver, receiver) - def systemDrain(): SystemMessage = null + def systemDrain(newContents: SystemMessage): SystemMessage = null def hasSystemMessages = false } diff --git a/akka-actor/src/main/scala/akka/dispatch/AbstractDispatcher.scala b/akka-actor/src/main/scala/akka/dispatch/AbstractDispatcher.scala index 48a91dd00c..12eea14ffc 100644 --- a/akka-actor/src/main/scala/akka/dispatch/AbstractDispatcher.scala +++ b/akka-actor/src/main/scala/akka/dispatch/AbstractDispatcher.scala @@ -107,6 +107,10 @@ private[akka] case class Watch(watchee: ActorRef, watcher: ActorRef) extends Sys * INTERNAL API */ private[akka] case class Unwatch(watchee: ActorRef, watcher: ActorRef) extends SystemMessage // sent to tear down a DeathWatch +/** + * INTERNAL API + */ +private[akka] case object NoMessage extends SystemMessage // switched into the mailbox to signal termination final case class TaskInvocation(eventStream: EventStream, runnable: Runnable, cleanup: () ⇒ Unit) extends Runnable { def run(): Unit = diff --git a/akka-actor/src/main/scala/akka/dispatch/BalancingDispatcher.scala b/akka-actor/src/main/scala/akka/dispatch/BalancingDispatcher.scala index e50f9150a4..6beee3c9da 100644 --- a/akka-actor/src/main/scala/akka/dispatch/BalancingDispatcher.scala +++ b/akka-actor/src/main/scala/akka/dispatch/BalancingDispatcher.scala @@ -52,15 +52,13 @@ class BalancingDispatcher( override def cleanUp(): Unit = { val dlq = actor.systemImpl.deadLetterMailbox //Don't call the original implementation of this since it scraps all messages, and we don't want to do that - while (hasSystemMessages) { - var message = systemDrain() - while (message ne null) { - // message must be “virgin” before being able to systemEnqueue again - val next = message.next - message.next = null - dlq.systemEnqueue(actor.self, message) - message = next - } + var message = systemDrain(NoMessage) + while (message ne null) { + // message must be “virgin” before being able to systemEnqueue again + val next = message.next + message.next = null + dlq.systemEnqueue(actor.self, message) + message = next } } } diff --git a/akka-actor/src/main/scala/akka/dispatch/Mailbox.scala b/akka-actor/src/main/scala/akka/dispatch/Mailbox.scala index b81a2fc0ba..d26e7b2afc 100644 --- a/akka-actor/src/main/scala/akka/dispatch/Mailbox.scala +++ b/akka-actor/src/main/scala/akka/dispatch/Mailbox.scala @@ -169,6 +169,7 @@ private[akka] abstract class Mailbox(val actor: ActorCell, val messageQueue: Mes */ protected final def systemQueueGet: SystemMessage = Unsafe.instance.getObjectVolatile(this, AbstractMailbox.systemMessageOffset).asInstanceOf[SystemMessage] + protected final def systemQueuePut(_old: SystemMessage, _new: SystemMessage): Boolean = Unsafe.instance.compareAndSwapObject(this, AbstractMailbox.systemMessageOffset, _old, _new) @@ -208,14 +209,14 @@ private[akka] abstract class Mailbox(val actor: ActorCell, val messageQueue: Mes } final def processAllSystemMessages() { - var nextMessage = systemDrain() + var nextMessage = systemDrain(null) try { while ((nextMessage ne null) && !isClosed) { if (debug) println(actor.self + " processing system message " + nextMessage + " with " + actor.childrenRefs) actor systemInvoke nextMessage nextMessage = nextMessage.next // don’t ever execute normal message when system message present! - if (nextMessage eq null) nextMessage = systemDrain() + if (nextMessage eq null) nextMessage = systemDrain(null) } } catch { case NonFatal(e) ⇒ @@ -235,15 +236,13 @@ private[akka] abstract class Mailbox(val actor: ActorCell, val messageQueue: Mes protected[dispatch] def cleanUp(): Unit = if (actor ne null) { // actor is null for the deadLetterMailbox val dlm = actor.systemImpl.deadLetterMailbox - while (hasSystemMessages) { - var message = systemDrain() - while (message ne null) { - // message must be “virgin” before being able to systemEnqueue again - val next = message.next - message.next = null - dlm.systemEnqueue(actor.self, message) - message = next - } + var message = systemDrain(NoMessage) + while (message ne null) { + // message must be “virgin” before being able to systemEnqueue again + val next = message.next + message.next = null + dlm.systemEnqueue(actor.self, message) + message = next } if (messageQueue ne null) // needed for CallingThreadDispatcher, which never calls Mailbox.run() @@ -300,7 +299,7 @@ private[akka] trait SystemMessageQueue { /** * Dequeue all messages from system queue and return them as single-linked list. */ - def systemDrain(): SystemMessage + def systemDrain(newContents: SystemMessage): SystemMessage def hasSystemMessages: Boolean } @@ -315,26 +314,30 @@ private[akka] trait DefaultSystemMessageQueue { self: Mailbox ⇒ assert(message.next eq null) if (Mailbox.debug) println(actor.self + " having enqueued " + message) val head = systemQueueGet - /* - * this write is safely published by the compareAndSet contained within - * systemQueuePut; “Intra-Thread Semantics” on page 12 of the JSR133 spec - * guarantees that “head” uses the value obtained from systemQueueGet above. - * Hence, SystemMessage.next does not need to be volatile. - */ - message.next = head - if (!systemQueuePut(head, message)) { - message.next = null - systemEnqueue(receiver, message) + if (head == NoMessage) actor.system.deadLetterMailbox.systemEnqueue(receiver, message) + else { + /* + * this write is safely published by the compareAndSet contained within + * systemQueuePut; “Intra-Thread Semantics” on page 12 of the JSR133 spec + * guarantees that “head” uses the value obtained from systemQueueGet above. + * Hence, SystemMessage.next does not need to be volatile. + */ + message.next = head + if (!systemQueuePut(head, message)) { + message.next = null + systemEnqueue(receiver, message) + } } } @tailrec - final def systemDrain(): SystemMessage = { + final def systemDrain(newContents: SystemMessage): SystemMessage = { val head = systemQueueGet - if (systemQueuePut(head, null)) SystemMessage.reverse(head) else systemDrain() + if (systemQueuePut(head, newContents)) SystemMessage.reverse(head) else systemDrain(newContents) } def hasSystemMessages: Boolean = systemQueueGet ne null + } /**