CODEWITHSUNDEEP
scalaakkaactor
James Davies
James Davies

Reputation: 9839

Wait on asynchronous Future call before processing next message in Akka

When receiving events, Akka Actors will process one message at a time, blocking until a request has been completed before moving onto the next message.

This works well for synchronous/blocking tasks, however if I want to perform an asynchronous/non-blocking request, Akka will continue processing without waiting for the task to complete.

For example:

 def doThing():Future[Unit] = /* Non blocking request here */

 def receive = {
     case DoThing => doThing() pipeTo sender
 }

This will call doThing() and start processing the future, but will not wait for it to complete before processing the next message - it will simple execute the next messages in the queue as fast as possible.

In essence, it appears that Akka considers "returning a future" to be "finished processing" and moves onto the next message.

In order to process one message at a time, it appears I need to actively block the Actor thread to stop it doing so

def receive = {
    case DoThing => sender ! blocking(Await.result(doThing()))
}

This feels like a very wrong approach - It's artificially blocking a thread in code that should otherwise be completely non-blocking.

When comparing Akka to, say, Elixir actors, we can easily avoid this problem in the first place by using a tail call to request the next message without needing to artificially block.

Is there any way in Akka to either

a) Wait for a Future to complete before processing the next message without blocking the thread.

b) Use an explicit tail call or some other mechanism to use a pull-based workflow instead of push based?

Upvotes: 19

Views: 5951

Answers (5)

James Brems
James Brems

Reputation: 1

context.become and context.unbecome are not thread safe so we must not call them directly from the completion context of a local future, but we could send ourselves an actor message and handle the context in a receive function:

  override def receive: Receive = {
    case DoThing =>
      val future = doThing() pipeTo sender

      //when future finishes stop stashing and unstash all messages
      future.onComplete {
        case Failure(e) => {
          self ! StopStash
        }
        case Success(value) =>  {
          self ! StopStash
        }
      }
      // Stash incoming messages until finished with future,
      context.become(stashing, discardOld = false)
  }

  def stashing: Receive = {
    case StopStash =>
      context.unbecome() // context.unbecome is not thread safe so we cannot call it in the success context of a future
      unstashAll()
    case _ =>
      stash()
  }

Upvotes: 0

Jeffrey Chung
Jeffrey Chung

Reputation: 19497

With Akka Streams, you could use mapAsync:

import akka.actor.ActorSystem
import akka.stream._
import akka.stream.scaladsl._
import scala.concurrent.Future

implicit val system = ActorSystem("ThingDoer")
implicit val materializer = ActorMaterializer()
implicit val ec = system.dispatcher

case object DoThing

def doThing(): Future[Unit] = Future {
  println("doing its thing")
}

Source((1 to 10).map(_ => DoThing))
  .mapAsync(parallelism = 1)(_ => doThing())
  .runWith(Sink.ignore)

Upvotes: 3

Levijatanu
Levijatanu

Reputation: 418

As David suggested but if you like to stash all messages 

class MyActor extends Actor with Stash {
  def doThing(): Future[WrappedFuture] = {}

  override def receive: Receive = {
    case msg: DoThing =>
      val future = doThing() pipeTo sender

      //when future finishes unstash all messages
      future.onComplete {
        case Failure(e) => {
          unstashAll()
          context.unbecome()
        }
        case Success(value) =>  {
          unstashAll()
          context.unbecome()
        }
      }
      // Stash incoming messeges until finished with future,
      context.become({
        case _ ⇒ stash()
      }, discardOld = false)
  }
}

Upvotes: 0

egbokul
egbokul

Reputation: 3974

Instead of having one actor dealing with this problem, have a chain of two:

  • Actor 1 receives the initial messages, starts all the IO calls and merges them in a Future
  • Actor 2 receives the results of the merged Futures.

This does not guarantee the preservation of message ordering, so if you need that then Actor 2 will have to be aware of the messages Actor 1 has seen, and potentially stash early messages on itself.

I am not aware of anything in Akka that solves this problem. Maybe there is a library that implements a pattern like this?

Upvotes: 0

David Caseria
David Caseria

Reputation: 261

Like it was suggested in the comments, you can use the Stash (http://doc.akka.io/docs/akka/current/scala/actors.html#Stash) trait to store incoming messages as you wait for the Future to resolve.

It's required to save the current sender so that you don't improperly close over the sender actor reference. You can achieve this through a simple case class like the one defined below.

class MyActor extends Actor with Stash {

  import context.dispatcher

  // Save the correct sender ref by using something like
  // case class WrappedFuture(senderRef: ActorRef, result: Any)
  def doThing(): Future[WrappedFuture] = ???

  override def receive: Receive = {
    case msg: DoThing =>
      doThing() pipeTo self

      context.become({
        case WrappedFuture(senderRef, result) =>
          senderRef ! result
          unstashAll()
          context.unbecome()
        case newMsg: DoThing =>
          stash()
      }, discardOld = false)
  }
}

Upvotes: 12

Related Questions