Actor starvation when self-messages and different dispatchers are involved

[clockfly]

Problem Statement:

We have a benchmark test which process 10 million message per second(project https://github.com/intel-hadoop/gearpump), during the test, I found sometimes some message cannot be delivered to target actor.

Test case description:

The test is like this:
I have 600 upstream actor, and 600 downstream actors distribued in 60 JVM. the upstream actors will crazily send data to downstream actors in a shuffle way. The upstream actors rely on self-sending message to trigger the repeation of sending.

upstream tasks and downstream tasks will share a custom dispatcher(not default dispatcher, or remote-default-dispatcher).

Upstream Actor logic:

1. upstream actor A send itself a tick message
2. in receive of A, sending one messge to downstream actors. After sending, send itself another tick message
3. repeat step 2

Trouble-shooting and findings

I compiled akka source to add log trace, and confirmed that the lost message is already inserted into target Mailbox queue(SingleConsumerOnlyUnboundedMailbox), and the Malibox(Runnable) is already inserted into internal task queue of fork-join pool. The strange part is that the mbox was NOT get scheduled to a real thead and run.

Finding1: about akka Version

I find version akka 2.3.6 don't have this problem, while version 2.3.9 has.

Finding2: about self-sending message

For my test, the upstream actors need to send themselves repeatedly tick message so that they can trigger themselves to send more message downstream. If I change the self-sending message, to be an indirect self sending, then the problem is gone.
For example, instead of a sending directly to itself A->A, I create a intermediate actor C, A->C->A,

Finding 3: About changes in latest akka version

Since 2.3.6 doesn't have this problem, I browser through the akka history, and find a suspicious change. https://github.com/akka/akka/pull/16152/files#diff-2c9d0638ce3a1f007623ac9d1ead1725R379

The intent of this PR is to make MailBox a subclass of ForkjoinTask, so that we can optimize the performance by fully using fork-join ideas. I doubt the changes in this PR will also create fairness problem.

Here is some change in AkkaForkJoinPool

     extends ForkJoinPool(parallelism, threadFactory, unhandledExceptionHandler, true) with LoadMetrics {
-    override def execute(r: Runnable): Unit =      +    override def execute(r: Runnable): Unit = {
-      if (r eq null) throw new NullPointerException else super.execute(new AkkaForkJoinTask(r))        +      if (r eq null) throw new NullPointerException("The Runnable must not be null")
+      val task =
+        if (r.isInstanceOf[ForkJoinTask[_]]) r.asInstanceOf[ForkJoinTask[Any]]
+        else new AkkaForkJoinTask(r)
+      Thread.currentThread match {
+        case worker: ForkJoinWorkerThread if worker.getPool eq this ⇒ task.fork()
+        case _ ⇒ super.execute(task)
+      }

If actor A is sending itself message, then it will trigger AkkaForkJoinPool.execute, as the calling thread is already in the fork-join pool,

+      Thread.currentThread match {
+        case worker: ForkJoinWorkerThread if worker.getPool eq this ⇒ task.fork()
+        case _ ⇒ super.execute(task)
+      }

So it will pick first branch which execute task.fork()

I doubt task.fork() will have higher priority than the other branch "super.execute(task)", so possibly the later will not get scheduled.

Thanks in advance.
Sean

[MQ-EL]

I think when Actor processing received message in forkjoin pool, nobody include itself can tell msg to it process in parallel.
so, actor A ! msg not trigger AkkaForkJoinPool.execute, it just enqueue msg into mailbox, and just return.
you can see registerForExecution method in Dispatcher class.

you can switch config from forkjoin to normal pool in 2.3.9, then test. So, exclude other suspect firstly.

Just some of my little thinks

[clockfly]

@MQ-EL

Dispatcher.registerForExecution will try to execute mbox with executorservice.
Check:

protected[akka] override def registerForExecution(mbox: Mailbox, hasMessageHint: Boolean, hasSystemMessageHint: Boolean): Boolean = {
    if (mbox.canBeScheduledForExecution(hasMessageHint, hasSystemMessageHint)) { //This needs to be here to ensure thread safety and no races
      if (mbox.setAsScheduled()) {
        try {
          executorService execute mbox
          true
        } catch {
          case e: RejectedExecutionException ⇒
            try {
              executorService execute mbox
              true
            } catch { //Retry once
              case e: RejectedExecutionException ⇒
                mbox.setAsIdle()
                eventStream.publish(Error(e, getClass.getName, getClass, "registerForExecution was rejected twice!"))
                throw e
            }
        }
      } else false
    } else false
  }

[MQ-EL]

when A tell msg to itself, A's mbox.setAsScheduled() will return false. because mailbox has set currentStatus flag with Scheduled value

[MQ-EL]

eh.... A ! msg to itself in scheduler's thread in your test?

I mean in step 2(After sending, send itself ...):
Upstream Actor logic:
1. upstream actor A send itself a tick message
2. in receive of A, sending one messge to downstream actors. After sending, send itself another tick message
3. repeat step 2

[clockfly]

@MQ-EL

In Mailbox.run

  final def run = {
    try {
      if (!isClosed) { //Volatile read, needed here
        processAllSystemMessages() //First, deal with any system messages
        processMailbox() //Then deal with messages
      }
    } finally {
      setAsIdle() //Volatile write, needed here
      dispatcher.registerForExecution(this, false, false)
    }
  }

in "finally", it will set the status as idle, and then do dispatcher.registerForExecution(this, false, false),
This call will schedule the mbox to Executor Service again.

[MQ-EL]

Yes, you are right. but it's not the problem. it's sequence!!! The msg will be exec eventually, that's we wanted.

I want to know which msg is lost? tick or A->B msg?

[clockfly]

@MQ-EL

I want to know which msg is lost? tick or A->B msg?

I see your questions now.

Suppose there is another dispatcher(for example: remote dispatcher) sending a ad-hoc message to A, that message may not be delivered to A.

[drewhk]

Hi Sean,

Is there a short reproducer we can try? I would like to see the whole
scenario in detail, preferably in code.

-Endre

On Thu, Apr 30, 2015 at 11:51 AM, Sean Zhong notifications@github.com
wrote:

@MQ-EL https://github.com/MQ-EL

I want to know which msg is lost? tick or A->B msg?

I see your questions now.

Suppose there is another dispatcher(for example: from other remote actors)
sending a ad-hoc message to A, that message may not be delivered to A.

—
Reply to this email directly or view it on GitHub
#17341 (comment).

[clockfly]

@drewhk

I will try to make one.

[clockfly]

Here is the reproducer I just wrote:

Test env: akka 2.3.9

package test 

import akka.actor.{ActorSystem, Props, Actor}
import com.typesafe.config.ConfigFactory

object ActorHang {

  val config = ConfigFactory.parseString(
"""
  |actorhang {
  |
  |  task-dispatcher {
  |    mailbox-type = "akka.dispatch.SingleConsumerOnlyUnboundedMailbox"
  |    throughput = 100
  |    fork-join-executor {
  |      parallelism-factor = 2
  |      parallelism-max = 2
  |      parallelism-min = 2
  |    }
  |  }
  |}
""".stripMargin
  )

  def main(args: Array[String]): Unit = {
    val system = ActorSystem("hang", config)

    val tasks = (0 until 20).map { i =>
      system.actorOf(Props(new Task(i)).withDispatcher("actorhang.task-dispatcher"))
    }.toList

    val messageLossTask = system.actorOf(Props(new MessageLossTask).withDispatcher("actorhang.task-dispatcher"))

    //wait until all task actor is running
    Thread.sleep(3000)

    Console.println("Sending CheckMessageLoss to target task")
    messageLossTask  ! CheckMessageLoss

    system.awaitTermination()
  }

  class Task(id: Int) extends Actor {
    private var sent = 0

    self ! "tick"

    override def receive: Receive = {
      case "tick" =>
        sent += 1
        self ! "tick"
    }
  }

  class MessageLossTask extends Actor {

    override def receive: Receive = {
      case msg: Any =>
        // In 2.3.9, CheckMessageLoss will not be received
        Console.println(s"Receiving message $msg")
    }
  }

  case object CheckMessageLoss
}