Using messages with different priorities over event stream in ActorSystem

[Eugene Dzhurinsky]

Hi all!

I need to publish messages of different types to event stream, and those
messages should have different priorities for example, if 10 messages of type
A have been posted, and one message of type B is posted after all, and
priority of B is higher than the priority of A - message B should be picked up
by next actor even if there are 10 messages of type A in queue.

I have read about prioritized messages
(http://doc.akka.io/docs/akka/2.0.3/scala/dispatchers.html#Mailboxes) and
created my simple implementation of that mailbox:

class PrioritizedMailbox(settings: Settings, cfg: Config) extends UnboundedPriorityMailbox(

PriorityGenerator {
case ServerPermanentlyDead => println("Priority:0"); 0
case ServerDead => println("Priority:1"); 1
case _ => println("Default priority"); 10
}

)

then I configured it in application.conf

akka {

actor {

prio-dispatcher {
type = "Dispatcher"
mailbox-type = "animatron.router.actor.mailbox.PrioritizedMailbox"
}

}

}

and wired into my actor:

private val myActor = actors.actorOf(
Props[MyEventHandler[T]].
withRouter(RoundRobinRouter(HIVE)).
withDispatcher("akka.actor.prio-dispatcher").
withCreator(
new Creator[Actor] {
def create() = new MyEventHandler(storage)
}), name = "eventHandler")

I use ActorSystem.eventStream.publish in order to send messages, and my actor
is subscribed to it (I can see in logs that messages are processed, but in
FIFO order).

However looks like it is not enough, because in logs/console I never seen the
messages like "Default priority". Am I missing something here? Does the
described approach work with event streams or just with direct invocations of
sending a message on actor? And how do I get prioritized messages with
eventStream?

--
Eugene N Dzhurinsky

[√iktor Ҡlang]

Hi Eugene,

It's already been answered here: http://stackoverflow.com/questions/12183645/scala-akka-multiple-event-buses-for-actorsystem-or-having-prioritized-events

It's only the mailbox that is prioritized, which means that if you process messages really fast, then they will never stay in the mailbox long enough to be prioritized relative to other entries in the mailbox.

Cheers,

√

 

--
Eugene N Dzhurinsky

--
Viktor Klang

Akka Tech Lead

Typesafe - The software stack for applications that scale

Twitter: @viktorklang

[Eugene Dzhurinsky]

On Thu, Aug 30, 2012 at 10:28:33PM +0200, √iktor Ҡlang wrote:
> Hi Eugene,
>
> It's already been answered
> here: http://stackoverflow.com/questions/12183645/scala-akka-multiple-e
> vent-buses-for-actorsystem-or-having-prioritized-events

Yes, but for some reason the proposed solution doesn't work. I created the
mailbox with priority generation, and added some debug messages to it, but
looks like it is not invoked at all.

--
Eugene N Dzhurinsky

[√iktor Ҡlang]

Works for me:

  trait Foo

  case object X extends Foo

  case object Y extends Foo

  case object Z extends Foo

  class PrioritizedMailbox(settings: ActorSystem.Settings, cfg: Config) extends UnboundedPriorityMailbox(

    PriorityGenerator {

      case X ⇒ 0

      case Y ⇒ 1

      case Z ⇒ 2

      case _ ⇒ 10

    })

val s = ActorSystem("prio", com.typesafe.config.ConfigFactory.parseString(

        """ prio-dispatcher {

        type = "Dispatcher"

          mailbox-type = "%s"

        }""".format(classOf[PrioritizedMailbox].getName)))

      val latch = new java.util.concurrent.CountDownLatch(1)

      val a = s.actorOf(Props(new akka.actor.Actor {

        latch.await // Just wait here so that the messages are queued up inside the mailbox

        def receive = {

          case any ⇒ /*println("Processing: " + any);*/ sender ! any

        }

      }).withDispatcher("prio-dispatcher"))

      implicit val sender = testActor

      a ! "pig"

      a ! Y

      a ! Z

      a ! Y

      a ! X

      a ! Z

      a ! X

      a ! "dog"

      latch.countDown()

      val expectedSequence = Seq(X, X, Y, Y, Z, Z, "pig", "dog")

      expectedSequence foreach { x => expectMsg(x) }

      s.shutdown()

This test passes with flying colors, so either you have something wrong in your config/router, always try to remove as many moving parts as possible when you are debugging, i.e why are you using a router in your example instead of just a plain actor?

Cheers,

√

[Eugene Dzhurinsky]

On Thu, Aug 30, 2012 at 11:27:03PM +0200, √iktor Ҡlang wrote:
> This test passes with flying colors, so either you have something wrong
> in your config/router, always try to remove as many moving parts as
> possible when you are debugging, i.e why are you using a router in your
> example instead of just a plain actor?

You were right, I created test like yours - but with EventStream, and realized
that my mailbox is not invoked, because all actors are picking up messages as
soon as they appear, so no reordering required. Then I updated number of
actors in RoundRobinRouter to 1 - and got messages prioritized.

Thanks for explanations!
--
Eugene N Dzhurinsky