Akka Actor gets a method to interact with Future (block) - can new messages appear before the future is completed?
I wrote the following code to try and figure out the behavior of an actor with respect to incoming messages when calling some asynchronous work in the Future (in fact, async work is meant to model an API for asynchronous asynchronous work):
val actor = system.actorOf(Props[WTFAsyncActor])
for (i <- 1 until 100) {
actor ! 'wtf + i.toString
Thread.sleep(200l) // yes I know actors should not invoke this method
}
Thread.sleep(10000l) // same as above
The Actor code looks like this:
class WTFAsyncActor extends Actor with ActorLogging{
import scala.concurrent.ExecutionContext.Implicits.global
override def receive: Receive = {
case i@_ =>
log.info(s"[$i] external block pre future")
Thread.`yield`() // just trying to relinquish control to induce context switching
Future {
log.info(s"[$i] internal block pre sleep")
Thread.sleep(1000l) // yeah bad - trying to emulate something meaningful like slick
log.info(s"[$i] internal block post sleep")
}
log.info(s"[$i] external block post future")
}
}
I am getting the following logs (excerpt from sample run)
10:17:58.408 [wtf-async-test-akka.actor.default-dispatcher-2] INFO org.wtf.test.WTFAsyncActor - ['wtf1] external block pre future
10:17:58.420 [wtf-async-test-akka.actor.default-dispatcher-2] INFO org.wtf.test.WTFAsyncActor - ['wtf1] external block post future
10:17:58.421 [wtf-async-test-akka.actor.default-dispatcher-2] INFO org.wtf.test.WTFAsyncActor - ['wtf1] internal block pre sleep
10:17:58.599 [wtf-async-test-akka.actor.default-dispatcher-3] INFO org.wtf.test.WTFAsyncActor - ['wtf2] external block pre future
10:17:58.600 [wtf-async-test-akka.actor.default-dispatcher-2] INFO org.wtf.test.WTFAsyncActor - ['wtf2] external block post future
10:17:58.600 [wtf-async-test-akka.actor.default-dispatcher-2] INFO org.wtf.test.WTFAsyncActor - ['wtf2] internal block pre sleep
10:17:58.800 [wtf-async-test-akka.actor.default-dispatcher-2] INFO org.wtf.test.WTFAsyncActor - ['wtf3] external block pre future
10:17:58.801 [wtf-async-test-akka.actor.default-dispatcher-3] INFO org.wtf.test.WTFAsyncActor - ['wtf3] external block post future
10:17:58.801 [wtf-async-test-akka.actor.default-dispatcher-3] INFO org.wtf.test.WTFAsyncActor - ['wtf3] internal block pre sleep
10:17:59.001 [wtf-async-test-akka.actor.default-dispatcher-3] INFO org.wtf.test.WTFAsyncActor - ['wtf4] external block pre future
10:17:59.002 [wtf-async-test-akka.actor.default-dispatcher-3] INFO org.wtf.test.WTFAsyncActor - ['wtf4] external block post future
10:17:59.002 [wtf-async-test-akka.actor.default-dispatcher-3] INFO org.wtf.test.WTFAsyncActor - ['wtf4] internal block pre sleep
10:17:59.202 [wtf-async-test-akka.actor.default-dispatcher-2] INFO org.wtf.test.WTFAsyncActor - ['wtf5] external block pre future
10:17:59.206 [wtf-async-test-akka.actor.default-dispatcher-3] INFO org.wtf.test.WTFAsyncActor - ['wtf5] external block post future
10:17:59.402 [wtf-async-test-akka.actor.default-dispatcher-3] INFO org.wtf.test.WTFAsyncActor - ['wtf6] external block pre future
10:17:59.403 [wtf-async-test-akka.actor.default-dispatcher-2] INFO org.wtf.test.WTFAsyncActor - ['wtf6] external block post future
10:17:59.421 [wtf-async-test-akka.actor.default-dispatcher-3] INFO org.wtf.test.WTFAsyncActor - ['wtf1] internal block post sleep
10:17:59.421 [wtf-async-test-akka.actor.default-dispatcher-3] INFO org.wtf.test.WTFAsyncActor - ['wtf6] internal block pre sleep
10:17:59.607 [wtf-async-test-akka.actor.default-dispatcher-3] INFO org.wtf.test.WTFAsyncActor - ['wtf2] internal block post sleep
10:17:59.607 [wtf-async-test-akka.actor.default-dispatcher-3] INFO org.wtf.test.WTFAsyncActor - ['wtf5] internal block pre sleep
10:17:59.608 [wtf-async-test-akka.actor.default-dispatcher-3] INFO org.wtf.test.WTFAsyncActor - ['wtf7] external block pre future
I think it's safe to say that new messages are handled as they come as long as some thread is available regardless of the Future block that hasn't finished executing yet. I'm right?
What I set for expansion is whether the receiver should wait for the Future block to complete before processing new messages. It seems that it is not . (wtf2 enters dispatcher thread 3 before wtf1 finishes computation on thread 3, for example)
Are there any caveats to this behavior?
Please excuse this may seem like a silly question. I haven't looked under the hood of akka codebase (I'm too new to scala and akka for this, at the moment)
source to share
In this case, subsequent messages can be processed receive
until the completion of any Future. There are no caveats to this, and there are things to watch out for. In particular, don't close any of your mutable states in the future. The future could run concurrently with messages being processed by the actor, violating the guarantee of single-threaded access to the state of the participants and leading to race conditions that could leave your actor state in an invalid position.
If you want to run any work from an actor asynchronously, the big picture is to start a child actor to do that work. A child actor can send messages back to its parent with any results, and you know for sure that a child actor cannot interfere with the parent actor's state.
source to share
You can't talk about it. The future created within the actor and the future created elsewhere are identical in behavior: they create an asynchronous computation that, at some point in time, will return the result of success or failure. It is important to note that there is no guarantee as to what the thread will execute in the future. It is possible that the same thread that is executing your actor's receive block will be allocated for future processing and then again allocated to the receive block, in which case everything appears to be running synchronously. But this is unlikely to happen often unless you are working with a single thread.
source to share