make cleanUp of systemMessages atomic

- extend systemDrain to take the new contents which shall be switched in
- make NoMessage placeholder which will signal final closing of the
  mailbox
- put that in when cleaning up, and check it when enqueuing

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)

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
   }
 

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 =

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(NoMessage)
-        var message = systemDrain()
+      while (message ne null) {
-        while (message ne null) {
+        // message must be “virgin” before being able to systemEnqueue again
-          // message must be “virgin” before being able to systemEnqueue again
+        val next = message.next
-          val next = message.next
+        message.next = null
-          message.next = null
+        dlq.systemEnqueue(actor.self, message)
-          dlq.systemEnqueue(actor.self, message)
+        message = next
-          message = next
-        }
       }
     }
   }

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(NoMessage)
-        var message = systemDrain()
+      while (message ne null) {
-        while (message ne null) {
+        // message must be “virgin” before being able to systemEnqueue again
-          // message must be “virgin” before being able to systemEnqueue again
+        val next = message.next
-          val next = message.next
+        message.next = null
-          message.next = null
+        dlm.systemEnqueue(actor.self, message)
-          dlm.systemEnqueue(actor.self, message)
+        message = next
-          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
-    /*
+    if (head == NoMessage) actor.system.deadLetterMailbox.systemEnqueue(receiver, message)
-     * this write is safely published by the compareAndSet contained within
+    else {
-     * systemQueuePut; “Intra-Thread Semantics” on page 12 of the JSR133 spec
+      /*
-     * guarantees that “head” uses the value obtained from systemQueueGet above.
+       * this write is safely published by the compareAndSet contained within
-     * Hence, SystemMessage.next does not need to be volatile.
+       * systemQueuePut; “Intra-Thread Semantics” on page 12 of the JSR133 spec
-     */
+       * guarantees that “head” uses the value obtained from systemQueueGet above.
-    message.next = head
+       * Hence, SystemMessage.next does not need to be volatile.
-    if (!systemQueuePut(head, message)) {
+       */
-      message.next = null
+      message.next = head
-      systemEnqueue(receiver, message)
+      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
+
 }
 
 /**