CODEWITHSUNDEEP
javascalaakkathreadpool
Juh_
Juh_

Reputation: 15539

Running Future.map inside a Future with a single thread ForkJoinPool

I had a code that used an ExecutionContext (EC) build with akka (ActorSystem). This code is doing something quite peculiar: it uses an AkkaForkJoinPool with parallelism-max = 1 and execute something like:

implicit ec = // akka EC backed by AkkaForkJoinPool with parallelism=1

Future{ // (1)
  // (2) get data from DB which uses a separate ExecutionContext for IO
  val data: Future[Data] = getData()

  // (3) use the data
  data.map{ whatEver }

  // etc ...
}

[Edit: I know, put like that this is strange to have the top Future (1). But in reality the code is not my own, it span several functions, and uses more complex operations such as several wrapped for-comprehension. So I won't change that]

Now I moved this code and replace the implicit ExecutionContext (EC) provided by akka by my own following the same rule: I use a (java) ForkJoinPool with parallelism = 1.

As a consequence, this code get stuck at the map (3). My understanding is that when the map (3) is called it requires a thread but the EC cannot provide one because its only available one is taken by the Future (1).

I am not clear how the ForkJoinPool is suppose to work. So my question is did I understand correctly, and:

  1. if not, I am using the java ForkJoinPool incorrectly. I.e. is there a way to make this work?
  2. if yes, how does akka manage it?

I am using akka 2.3.15, scala 2.11.12 and java 8

Upvotes: 0

Views: 129

Answers (2)

Juh_
Juh_

Reputation: 15539

Looking into akka's code, I think I found what it does. I am not completely sure but almost: akka ActorSystem create a Dispatchers which create a MessageDispatcherConfigurator which create a Dispatcher that creates the ExecutorService (I pass on the class hierarchy of that). There are several possible implementations, but this is the most common I think, and this is what happens when using a ForkJoinPool.

Now, the Dispatcher extends BatchingExecutor which is the one that can batch together internal task such as the map in the question (which requires a Thread to run) to the current Thread.

Once again, the code is too complex for me to be sure and I won't investigate more. But indeed akka EC can wrap internal map call to the parent thread which is not what happens with standard (i.e. java) ForkJoinPool.

I think this is a smart trick from akka, and not a typical implementation. The doc of BatchingExecutor says:

/**
 * Mixin trait for an Executor
 * which groups multiple nested `Runnable.run()` calls
 * into a single Runnable passed to the original
 * Executor. This can be a useful optimization
 * because it bypasses the original context's task
 * queue and keeps related (nested) code on a single
 * thread which may improve CPU affinity. However,
 * if tasks passed to the Executor are blocking
 * or expensive, this optimization can prevent work-stealing
 * and make performance worse. Also, some ExecutionContext
 * may be fast enough natively that this optimization just
 * adds overhead.
 * The default ExecutionContext.global is already batching
 * or fast enough not to benefit from it; while
 * `fromExecutor` and `fromExecutorService` do NOT add
 * this optimization since they don't know whether the underlying
 * executor will benefit from it.
 * A batching executor can create deadlocks if code does
 * not use `scala.concurrent.blocking` when it should,
 * because tasks created within other tasks will block
 * on the outer task completing.
 * This executor may run tasks in any order, including LIFO order.
 * There are no ordering guarantees.
 *
 * WARNING: The underlying Executor's execute-method must not execute the submitted Runnable
 * in the calling thread synchronously. It must enqueue/handoff the Runnable.
 */

Upvotes: 0

Jamie
Jamie

Reputation: 6124

Instead of wrapping everything in a future, use a for-comprehension on the result of the first future, since everything depends on it.

for {
  data <- getData()
} yield data.map( whatEver )

or

getData().map { data =>
  data.map { whatEver }
}

Upvotes: 1

Related Questions