Nested Future blocks do not run in parallel in Scala 2.13.x

[lihaoyi]

reproduction steps

In the code below, we can see that in Scala 2.13.2, the three nested Future blocks run sequentially, whereas in Scala 2.12.12 they run in parallel. I would expect them to run in parallel in both cases, as I am spawning a small number of Futures and have 16 cores on this machine that should be able to run them in parallel.

$ sbt '++2.13.2!; console'
Welcome to Scala 2.13.2 (OpenJDK 64-Bit Server VM, Java 1.8.0_252).
Type in expressions for evaluation. Or try :help.

scala> import scala.concurrent._, ExecutionContext.Implicits._, duration.Duration.Inf

scala> def slow(key: String) = Future{ println(s"$key start"); Thread.sleep(1000); println(s"$key end"); key }

scala> def runAsyncSerial(): Future[Seq[String]] = slow("A").flatMap { a => Future.sequence(Seq(slow("B"), slow("C"), slow("D"))) }

scala> Await.result(runAsyncSerial(), Inf)
A start
A end
D start
D end
C start
C end
B start
B end
val res0: Seq[String] = List(B, C, D)

$ sbt '++2.12.12!; console'
Welcome to Scala 2.12.12 (OpenJDK 64-Bit Server VM, Java 1.8.0_252).

scala> import scala.concurrent._, ExecutionContext.Implicits._, duration.Duration.Inf

scala> def slow(key: String) = Future{ println(s"$key start"); Thread.sleep(1000); println(s"$key end"); key }

scala> def runAsyncSerial(): Future[Seq[String]] = slow("A").flatMap { a => Future.sequence(Seq(slow("B"), slow("C"), slow("D"))) }

scala> Await.result(runAsyncSerial(), Inf)
A start
A end
B start
C start
D start
B end
C end
D end
res0: Seq[String] = List(B, C, D)

AFAICT this is a minimal repro; either (A) wrapping the Thread.sleep in blocking() or (B) using an explicit Threadpool execution context or (C) removing the flatMap wrapper all seem to make it run correctly parallel in 2.13.2. This applies to all versions 2.13.x

OTOH -Ydelambdafy:inline or replacing the flatMap with map + flatten seems to make no difference. Neither does changing the JVM version from 8 to 11, or swapping between sbt console and Ammonite, or moving it from the REPL into a .sc script or full .scala file in a Mill project.

In the above example I'm spawning a small number of Futures, but in the case where I hit this in the wild I was spawning 100s of Futures and having them all run serially instead of in parallel, resulting in a 16x slowdown over what I was expecting

Notably, this slowdown applies regardless of how long the operations are: an operation that takes 1000ms to run 16x parallel now would take 16000ms, but an operation that takes 1ms to run 16x parallel would now take 16ms. Both could be equally bad depending on where it happens (e.g. in a lot of backend services, extra 15ms on every operation would violate SLAs)

Furthermore, the Futures documentation precisely and accurately describes the behavior of the ExecutionContext.global pre-2.13, as @jimm-porch has pointed out (https://docs.scala-lang.org/overviews/core/futures.html):

The number of concurrently blocking computations can exceed the parallelism level only if each blocking call is wrapped inside a blocking call (more on that below).

Last but not least, you must remember that the ForkJoinPool is not designed for long lasting blocking operations.

This description is no longer accurate as of 2.13. From what I have gathered in the discussion on this thread, the 2.13 behavior is better described as:

The number of any concurrent computations can exceed one only if each call is wrapped in a blocking call

Last but not least, you must remember that the ForkJoinPool is not designed for CPU-bound operations.

[SethTisue]

my understanding is that this is working as designed, as per scala/scala#7470, and that you're expected to deal with it with:

wrapping the Thread.sleep in blocking()

cc @viktorklang

[SethTisue]

(Perhaps https://github.com/scala/scala/releases/tag/v2.13.0 should have some wording about this.)

[lihaoyi]

A silent change in time taken to run the simplest of behaviors from O(1) to O(n) is certainly not what I would expect when I read “faster and more robust” 😛

The linked PR talks about "combinators". I'm not using any combinators here, except from the wrapping flatMap which is really irrelevant. The only thing I'm doing is spawning Future{...} blocks

[som-snytt]

On the gitter, I said, "it would be nice if the docs explained the idiom and also why it isn't self-explanatory."

The side-effecting onComplete and foreach are get-out-jail-free, perhaps because of the guidance that you don't care when they run or in what order. Maybe also because it represents extra or other work.

Someone also quoted the code comment for batching that is not in the docs. (I just remembered that I've had to quote before, too. Apparently Scaladoc lets you include --access again for non-public members.) The quote breaks off before the caveat:

However,
 * if tasks passed to the Executor are blocking
 * or expensive, this optimization can prevent work-stealing
 * and make performance worse.
 * 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.

As shown above, the comment should read, "especially LIFO order."

[viktorklang]

A silent change in time taken to run the simplest of behaviors from O(1) to O(n) is certainly not what I would expect when I read “faster and more robust” 😛

If you didn't do the Thread.sleep() it would most likely be faster. Also, not surrounding it in blocking{} is inadvisable.

The linked PR talks about "combinators". I'm not using any combinators here, except from the wrapping flatMap which is really irrelevant. The only thing I'm doing is spawning Future{...} blocks

From the Future.apply scaladoc:

The following expressions are equivalent:

val f1 = Future(expr)
val f2 = Future.unit.map(_ => expr)
val f3 = Future.unit.transform(_ => Success(expr))

If you want to run a side-effect you should consider Future.unit.foreach(_ => effect) or Future.unit.onComplete(_ => effect)

[lihaoyi]

I don't want to run a side effect, I just want to be able to parallelize these computations using Futures, the same way I have been doing since Scala 2 introduced Futures and Promises in SIP14. What puzzles me most is that surely I can't be the first person bumping into these issues? I assume that other people also rely on Futures to parallelize things? The above code snippet isn't exactly an esoteric edge case.

The time taken to run each individual task seems irrelevant here: we have code that used to be parallel, but is now not, and the slowdown can be made arbitrarily large depending on how many Futures I kick off. I hit this kicking off 100ish Futures on a 16 core machine and having the all complete sequentially, a 16x slowdown. If I was running on a beefy EC2 instance like a m5.24xlarge with 96 cores it would have been a 96x slowdown. It would take quite a lot of reduction in fixed overhead to overcome a 96x slowdown due to loss of parallelization.

Do Futures created in other Futures never run in parallel any more when using the default ExecutionContext? If not, then when do they still in parallel? This isn't just a theoretical concern: in one part of our system it was causing 16x slowdowns, in another it was causing mysterious 5-10 second delays in our asynchronous push-update code paths. Both of these are real user-facing consequences that I as the product owner can't brush under the rug, so I'd like to be able to understand the mechanism that's causing this to behave as it does

[viktorklang]

@lihaoyi Whether or not something will run in parallel is up to the ExecutionContext. With the default EC it will parallelize "top-most" Futures (assuming that it has more than one thread) and break off eligible "callbacks" once a blocking{} is encountered. Assuming you have written your code to take an implicit ExecutionContext you can pass in one which gives you the desired behavior—there is always a tradeoff between performance and coordination overhead.

What could be considered is a system property to turn batching off, if one wants to maximize parallelism.

[lihaoyi]

With the default EC it will parallelize "top-most" Futures (assuming that it has more than one thread) and break off eligible "callbacks" once a blocking{} is encountered

Thanks Viktor! This is exactly what I wanted to know. I'm aware of the tradeoff between parallelism and coordination overhead. I must say that I have never seen the blocking{} semantics explained before in such a clear and concise fashion

While I don't really agree with changing the choice of tradeoff so drastically for the global execution context in Scala 2.13, it is what it is.

@SethTisue is there any way we can advertise this large change in behavior more broadly? I do think of myself as someone who knows my way around Scala and Futures, but this change in Scala 2.13 completely blindsided me. I spent months wondering why my Futures were sometimes not scheduling promptly (but never in a minimal dev environment, only in production under load!) and a full day trying to minimize the test case presented above from a program behaving 16x slower than it should. I imagine the vast majority of Scala developers would have even more difficulty than I would if they bump into this behavior change in the wild.

[lihaoyi]

Also, if this is expected behavior should we close this issue?

[som-snytt]

I was going to suggest keeping this issue open for improving the docs for clarification. After re-reading the Scaladocs, Javadocs, and the overview section, I understand that task scheduling is up to the execution context, that fork-join tasks should be of an appropriate size, that I don't know what API incurs a submit or a fork in the global context, and that I'm not convinced blocking is a great way to say "this bit of code doesn't block but is an appropriate unit of work for work-stealing." (The Javadocs say a unit of work should be yea-big so scheduling overhead doesn't swamp useful work. What if future used a macro that examined byte code to estimate benefitsFromBatching?)

The overview has a stub section for "extensions", so maybe a point of adumbration would show how to write some code to improve fork-join ergonomics.

Alternatively, a blog post about the Klang levels, K0 thru K3, to correlate expertise with what usages one should attempt. Maybe the API should all take an implicit K-level, import concurrent.K1 would enable certain combinators but no blocking.

I see I was due for a semi-annual review of how do futures work again? because my last was for #11827 which was also unexpectedly expected behavior. There must be an Onion headline along the lines of, "Coder most surprised to learn they violated the principle of least surprise."

[viktorklang]

@som-snytt

I'm not convinced blocking is a great way to say "this bit of code doesn't block but is an appropriate unit of work for work-stealing."

That's not really what happens though—if there are enqueued operations locally (on the current thread) then those will be submitted to be executed to the pool prior to running the blocking chunk of code, in order to preserve liveness under blocking. This paired with ManagedBlockers in FJP works rather nicely together by trying to ensure that there are threads available to process non-blocking operations in the face of blocking operations.

[nadavwr]

Prior to 2.13 I remember first seeing blocking used to convey non-blocking fork in the context of Akka's default dispatcher, which at the time was already batching.

Now that ExecutionContext is batching in Scala 2.13, perhaps adding fork as an alias to blocking would be advisable? This would better convey user intent—the user intends the nested Future not to be linked to the same fiber.

[mushtaq]

@viktorklang a clarifying question:

If the slow method above does not use Thread.sleep, instead it represents a non-blocking web-service call (say using Akka Http client), then the nested invocations of slow within runAsyncSerial will run in parallel, right?

[viktorklang]

@mushtaq It depends on how/where the async web-service calls are processed—they would be initiated serially by the Global EC, but their processing may occur on some other pool in parallel.

[viktorklang]

@nadavwr

Now that ExecutionContext is batching in Scala 2.13, perhaps adding fork as an alias to blocking would be advisable? This would better convey user intent—the user intends the nested Future not to be linked to the same fiber.

fork wouldn't be semantically correct since the block of code inside blocking{} is not forked, but any locally (thread locally) enqueued Future-operations would be resubmitted to the pool, so that they can be processed by some other thread.

[viktorklang]

@lihaoyi One option would be to treat Future.apply as a fork (semantically) which would then remove the similarity from Future.unit.map(_ => block) in which case users could themselves select if they wanted fork-like semantics or not.

[alexandru]

Future.apply being a "fork" might be the correct option because users expect it.

In tutorials on Future I've seen it mentioned that Future.apply forks and that if people don't want that then Future.successful should be used.

If an alternative is desired that's only memory-safe (i.e. one that registers a Runnable with Batching in the underlying thread-pool), then that alternative should be separate from Future.apply. I'd suggest Future.trampoline 🙂

[nadavwr]

@viktorklang Right. Perhaps def fork[A](f: => Future[A]): Future[A] = blocking(f) then?

Future.apply does sound pleasantly unsurprising.

[mushtaq]

@mushtaq It depends on how/where the async web-service calls are processed—they would be initiated serially by the Global EC, but their processing may occur on some other pool in parallel.

We are using the same EC (ActorSystem.executionContext ) for all non-blocking calls, including the web-client. If I understand you @viktorklang, nested Future.sequence will not parallelise in that case. If so, that is really sad 😞

I feel that this is a very fundamental change and even the long time scala.concurrent.Future users may not have understood this implication yet.

I think some guidance on how to retain the parallel nature of nested Future.sequence calls in purely asynchronous applications will help. Is there any role of blocking in such cases? Wrapping non-blocking web-client calls with blocking does not look right. Maybe, we should use a separate EC even for non-blocking web calls?

[viktorklang]

@mushtaq It completely depends on how ActorSystem.executionContext is implemented.

[viktorklang]

@nadavwr

Right. Perhaps def fork[A](f: => Future[A]): Future[A] = blocking(f) then?

No, that doesn't help.

This would though:

https://github.com/scala/scala/pull/9129/files

[lihaoyi]

I don't think special casing Future.apply is the right way to go. I would expect the following to run in parallel as well:

def slow(key: String) = Future{ println(s"$key start") }.map{_ => Thread.sleep(1000); println(s"$key end"); key }

def runAsyncSerial(): Future[Seq[String]] = slow("A").flatMap { a => Future.sequence(Seq(slow("B"), slow("C"), slow("D"))) }

Is it possible to only batch things together when we're actually chaining operations on the same Future? I want this to be one batch

Future{ Future(...).map(...).map(...).map(...).map(...) }

And I want this to be two batches:

Future{ (Future(...).map(...).map(...), Future(...).map(...).map(...)) }

I think that would be intuitive and would not surprise anyone, whereas the current batching of completely unrelated futures is very surprising

[jrudolph]

At least for Future.apply the 2.13 behavior is unfortunate, indeed.

I think that would be intuitive and would not surprise anyone, whereas the current batching of completely unrelated futures is very surprising

AFAICS this works as expected with Viktor's patch as long as you have a Future.apply in the beginning of each branch that you want to run in parallel. I'm not sure if it's guaranteed to be run in parallel or if it is just more likely. I'm inclined to think that it's guaranteed because all those forked initial future thunks will not be batched but will be put into task submission queues of the underlying EC. And each thread will first work on its batching queue before looking at other tasks, so the parallel tasks should be up for grabs for other threads.

Is it possible to only batch things together when we're actually chaining operations on the same Future? I want this to be one batch

Future{ Future(...).map(...).map(...).map(...).map(...) }

And I want this to be two batches:

Future{ (Future(...).map(...).map(...), Future(...).map(...).map(...)) }

The first example doesn't make sense because there's a single dependency chain between all tasks, so, of course, they must be run sequentially. The question is more if in

val a : Future[...] == ...
Future.sequence(a.map(longRunning), a.map(longRunning2))

those two long-running tasks would be run in parallel because they depend on the same original future. If that's not working with Viktor's patch, I'd still find it acceptable if it would require to use Future.apply around those long running chunks of work as well.

In general, it's tempting to simply use Future.apply or (map or whatever) for long-running CPU intensive tasks for parallelization but for best performance you will always need fine tuning for these kind of loads. Putting long running tasks on the main EC of your application will have almost the same bad effects of thread starvation as blocking tasks. Using blocking still wouldn't be the right thing because you exactly don't want to spawn extra threads if you are already CPU-bound. In many cases, you might want to either break down tasks into smaller pieces or run them on dedicated ECs to keep the main ECs free for asynchronous task handling.

[lihaoyi]

AFAICS this works as expected with Viktor's patch as long as you have a Future.apply in the beginning of each branch that you want to run in parallel.

This is good enough. As you said, if both sides starts with a : Future[...], I would expect the maps to run in parallel. Future.apply shouldn't be special, what I want is unrelated tasks in the computation graph to run in parallel.

In general, it's tempting to simply use Future.apply or (map or whatever) for long-running CPU intensive tasks for parallelization but for best performance you will always need fine tuning for these kind of loads. Putting long running tasks on the main EC of your application will have almost the same bad effects of thread starvation as blocking tasks. Using blocking still wouldn't be the right thing because you exactly don't want to spawn extra threads if you are already CPU-bound. In many cases, you might want to either break down tasks into smaller pieces or run them on dedicated ECs to keep the main ECs free for asynchronous task handling.

This may or may not be true, but it's somewhat irrelevant here: these problems are well known, and well understood, well beyond the Scala ecosystem. I can take a colleague who's familiar with Node.js, Java's ThreadPoolExecutors, or Python's multiprocessing.pool and have them immediately understand the problems with thread starvation.

What I'm asking for isn't to magically solve thread starvation, what I'm asking for is to have a predictable behavior that unrelated parts of an asynchronous computation will proceed in parallel, as far as the threads allow. This was the case for the global ExecutionContext for 7 years from Jan 2013, and only changed in June 2019. We can wring our hands forever about what the "ideal" solution would be, but ExecutionContext.global was a working solution that I had put into production many times over the past 7 years, and used in countless onboarding and teaching material. I'm rather unhappy to see my 7 years of production systems, blog posts, tech talks, and a book so casually broken in the name of saving a few nanoseconds.

[jrudolph]

saving a few nanoseconds

What I wanted to say is that in the end, there are at least two different use cases for Futures: one is efficient management of asynchronous callbacks (that are usually short running) vs. using the same infrastructure for efficiently running long-running tasks. Maybe people who mostly run things with async in mind are rather glad about the performance improvements (also because they already manage the long-running tasks in special ways anyway). So, it really is a trade-off that probably went too far with the change in 2.13. With the suggested change to Future.apply, maybe the default solution is better balanced for most cases? But I wouldn't want to be the judge here, it might be that there's no solution that fits all (in which case avoiding breaking stuff might be the better choice).

We can wring our hands forever about what the "ideal" solution would be, this was a working solution

I think we need to wring out hands exactly, because things always change and we need to argue about what improvements we can make and which ones we might not want to make. Stability is a important but there must be room for improvements.

[eed3si9n]

.. ExecutionContext.global was a working solution that I had put into production many times over the past 7 years, and used in countless onboarding and teaching material. I'm rather unhappy to see my 7 years of production systems, blog posts, tech talks, and a book so casually broken in the name of saving a few nanoseconds.

+1 on Haoyi here.

Making any changes to the standard library should be done in a careful way such that it won't change the semantics of production code especially for a popular feature like ExecutionContext.global. Given that the semantics can change with bringing in new ExecutionContext, would it be possible to make the BatchedExecutor an explicit opt-in instead?

[mdedetrich]

@lrytz I understand the issue, the point is that the new batching executor is designed to not parallelize certain tasks because doing so is slower than parallelizing them, i.e. the batching executor is meant to solve the problem where someone tries to calculate Fibonacci using Future's, in such a case you don't want such tasks to be calculated on seperate threads/pools and ideally you would to calculate the entire Fibonacci on a single thread (and also the same thread you are currently on).

This works fine assuming you use blocking correctly, yes its true that you may have less parallel tasks running but if you are doing something like

Future { 1 }.flatMap { x => Future { x + 1 } }

You don't want the x + 1 to be parallelized and to go onto another thread because its really slow to do so (and that is how the default global ExecutionContext worked prior to Scala 2.13.x) i.e.. x+1 is an incredibly fast operation so its better to just run this on the current thread. The issue is there is no way figure out what is "fast" and what is "slow" for thread blocking operations (i.e. x+1 vs Thread.sleep(1000), hence why blocking exists)

[mdedetrich]

The default execution context never had a “not suitable for tasks longer than 10us” warning (or however much the scheduling overhead is). If it did, this discussion would be different, but it didn’t.

This was documented in Future, see #12089 (comment) . There is an argument the documentation is unclear, hence my previous comment about linting (i.e. it being a shame that there isn't a standard Scala linter being used with such a rule to pick this up).

[lihaoyi]

@mdedetrich that quoted documentation perfectly describes the limitations of ExecutionContext.global pre-2.13. It does not reflect the limitations since 2.13.0

[mdedetrich]

Sure, I said it was unclear. The critical point being made is that this limitation (as you call it) applies to Future generally, irrespective of what ExecutionContext is being used and what Scala version you are using. The design of Future makes it very clear why this is the case.

Its always that when using blocking operations with Future you should either use blocking or put the blocking computation onto a specific ExecutionContext using map/flatMap and explicitly specifying the suitable ExecutionContext (i.e. a fixed/cached thread pool).

Again its a big shame that this convention wasn't documented clearly enough.

EDIT: Even for other IO types this is the case, i.e. see https://tpolecat.github.io/doobie/docs/14-Managing-Connections.html where they recommend using a fixedThreadPool for JDBC. This is a general problem with asyc code, you get into problems with mixing long blocking code with async code (and this is irrespective of Scala's Future, even Node.js has the same problem).

[lihaoyi]

The thing is, it was not unclear. The linked eocumentation is precise and accurate and clear as day. Just because we don’t like the documented behavior doesnt mean we get to retro-actively declare it “unclear”.

[mariogalic]

May I suggest linking the advice regarding usage of blocking for Future(blocking(longOp)) directly at ExecutionContext scaladoc, or maybe even Future scaladoc, instead of just concurrent package scaladoc. Even from concurrent package doc it is not actually directly stated, but one must follow the link to see first concrete example at https://docs.scala-lang.org/overviews/core/futures.html#blocking-inside-a-future.

[mdedetrich]

The thing is, it was not unclear. The linked eocumentation is precise and accurate and clear as day.

Okay, replace unclear with misleading. The same point stands and we are kinda grasping at straws here.

Just because we don’t like the documented behavior doesnt mean we get to retro-actively declare it “unclear”.

Its not about "not liking", its about whether its intended behavior. Abstractions are designed to have an intended behavior, otherwise they lose all meaning/reasoning.

And yes I am saying that we should also update the documentation and any other materials (as much as possible) to make this ultra clear. Future does not work well when mixing blocking and non blocking operations unless you explicitly tell it (one way or another) what operations are blocking, otherwise you get problems like this. This is how its designed and coded.

As I said before, you can keep on doing what you are doing (and btw I personally don't have an issue with reverting the default ExecutionContext) but regardless if this gets changed or not you are still misusing the abstraction. Its just a very unfortunate thing that this was documented badly/incorrectly etc etc.

[sjrd]

I would like to echo one of @lihaoyi's points about the documentation.

To me it was always very clear that you should wrap blocking operations with blocking {}.
It was however also very clear that you did not have to use blocking {} for long-running operations, and in fact shouldn't use it.

In other words, as long as the task keeps the CPU busy, there should be no reason to use blocking {} (if there was it would be called longRunning, not blocking), because I'm still maximizing throughput.

If I have a blocking operation, i.e., one that leaves the CPU idle, then I need blocking {} to keep maximizing throughput.

The documentation was never saying that non-blocking long-running operations should be wrapped in blocking {}. In fact, since it keeps repeating that blocking operations should be wrapped in blocking {}, and never mentions long-running operations, it is clearly telling people not to use blocking {} for non-blocking long-running operations.

Whether or not Futures were intended to require blocking {} for long-running operations is irrelevant today, because both the documentation and the implementation were in favor of not using blocking {} for long-running operations.

[lihaoyi]

@sjrd’s comment makes me realize one more thing: it is impossible to write code using ExecutionContext.global that behaves properly in 2.12 and 2.13, no matter how we treat CPU-bound tasks (doesn’t need to be long running)

• If we use blocking{} around CPU bound tasks, in Scala 2.13 we’ll get a thread pool explosion and have many many more threads than cores

• If we do not use blocking{} around CPU bound tasks, in 2.13 we get zero parallelism

Breakage aside, or even the inability to easily migrate old code to preserve 2.12 parallelism behavior in 2.13 , this kind of impossible-to-make-cross-version-compatible code is another level of inconvenience

[viktorklang]

In other words, as long as the task keeps the CPU busy, there should be no reason to use blocking {} (if there was it would be called longRunning, not blocking), because I'm still maximizing throughput.

Throughput for that specific task, but you're impacting fairness negatively by excessively occupying a shared resource.
It is also important to remember that "blocking" describes a behavior, not an implementation (as it can be implemented in the VM or OS as spin-locks, back-off spin-locks, mutexes, etc). In this case the documentation should be clarified to mean "blocking progress for other tasks" (these are hard to word, because there is no universal metric of fairness.

Whether or not Futures were intended to require blocking {} for long-running operations is irrelevant today, because both the documentation and the implementation were in favor of not using blocking {} for long-running operations.

And it was also clear that global was not necessarily a good place to issue long-running operations. Ultimately documentation will never prevent all unintended uses, nor relying on runtime behaviors which are not documented—alas that doesn't mean that we shouldn't strive to always improve it.

All that said, all while it is good to discuss how documentation can be improved (it almost certainly always can be), I think it is more constructive to discuss the path forward.

I think the least disruptive option is to: introduce a batched version of global separately (it will still run on the same thread pool), so that those who are mindful about the impact of their tasks on other tasks can regain the same performance as 2.13.0-1-2 by switching to it, while restoring the <2.13 behavior for existing code. Improving the documentation for all the parts can be done as a part of this change (be clearer about expected behavior of tasks, the recommendation of blocking {} in more places, clarification of what blocking {} means, etc).

[viktorklang]

If we use blocking{} around CPU bound tasks, in Scala 2.13 we’ll get a thread pool explosion and have many many more threads than cores

@lihaoyi You can specify the number of extra threads:

scala.concurrent.context.maxExtraThreads = defaults to "256"
[..]
The maxExtraThreads is the maximum number of extra threads to have at any given time to evade deadlock, see scala.concurrent.BlockContext.

• https://www.scala-lang.org/api/2.13.3/scala/concurrent/ExecutionContext$.html#global:scala.concurrent.ExecutionContextExecutor

[mdedetrich]

@lihaoyi

If we use blocking{} around CPU bound tasks, in Scala 2.13 we’ll get a thread pool explosion and have many many more threads than cores

You can also specify an explicit ExecutionContext , i.e. cached thread pool would likely be appropriate for your usecase. That way you wont run out of threads (also depends on what blocking operations you are doing).

[lihaoyi]

@viktorklang I know I can tweak maxExtraThreads, but that's still not sufficient.

My understanding is that the ideal thread pool size is num_threads = num_cores + num_io_blocking_operations. This is the semantics provided by the pre-2.13 ExecutionContext.global, and works to ensure all cores are fully utilized without having extraneous threads to add contention and context-switching overhead. This is common doctrine in any asynchronous programming environment, beyond just Scala.

This is implemented by having a default thread pool which is the size of num_cores, and asking the developer to mark blocking IO with blocking{} blocks so the runtime can correctly spin off the more threads as necessary (up to maxExtraThreads).

With the 2.13 blocking semantics, I am being told that I should use blocking{} for every CPU bound operation as well as every long IO-bound operation. That means that the runtime is unable to distinguish blocking-IO and CPU-bound operations:

• If maxExtraThreads is high, I get num_threads > num_cores + num_io_blocking_operations, resulting in too many threads fighting for a small number of CPUs, and thus too much context switching and inefficiency

• If maxExtraThreads is low, I end up with num_threads = num_cores < num_io_blocking_operations, and thus idle CPUs

In pre-2.13, the blocking{...} blocks were what told the runtime what num_io_blocking_operations was at any one point in time, so num_threads could dynamically adjust in response. In 2.13, the runtime no longer gets this signal: blocking{} now is used for both IO-bound and CPU-bound actions, and the runtime can no longer ensure num_threads = num_cores + num_io_blocking_operations.

Maybe this num_threads = num_cores + num_io_blocking_operations doctrine has been wrong all along, but it's served well for a long time and doesn't seem unreasonable to me.

[viktorklang]

@lihaoyi In my experience, if you're CPU-bound you want about num_threads = 0.7 x num_cores since there are GC-threads and other threads who also need to be able to get CPU-time to avoid STW (stop-the-world) pauses.

Your situation w.r.t. num_threads and num_extra_threads seems to assume that there is no other tasks which may need to run on global—generally no such guarantee can be made since that assumes a closed-world assumption, which global cannot make. (See argument previously made w.r.t. that)

the runtime can no longer ensure num_threads = num_cores + num_io_blocking_operations.

It never could, as the pool does not know whether semantic blocking is implemented with spin-locks or otherwise.
In general, since you cannot really have preemptive scheduling in user-land (especially not if you allow native code execution, see Erlang etc) the way to handle cooperative scheduling is for all parties to remain cooperative, which means avoiding "hogging" shared resources, and ideally divide longer "processes" into discrete steps to allow for other tasks to make progress.

In any case, let's try to find a path forward, so what do you think about my proposition?

[lihaoyi]

In any case, let's try to find a path forward, so what do you think about my proposition?

A lot of things have been discussed, so I'll go through in one in turn.

1. As mentioned earlier, I do not think special casing Future.apply is the right thing to do. It doesn't really solve the problem, and adds more complexity onto a really simple underlying model.

2. If there's some way to preserve batching while still allowing the parallelism we were getting before, I would be all for it. But the Future internals are gnarly and I do not feel competent enough to really discuss that option. If you say it's not feasible, I have to believe you.

3. As far as introduce a batched version of global separately, I think that's the right thing to do. Ideally I'd like there to be a half-dozen different ExecutionContext.*s that I can import and try out. We already have .global and .parasitic, adding .batching and others would fit nicely.

4. However, due to forward-and-backwards binary compatibility requirements, I do not believe this is possible for the next few years? I recall being told the Standard Library was frozen until Scala 3. Waiting until Scala 3.1 to solve the problem (2022? 2023?) seems like too long to wait for a regression of this sort.

5. Similarly, while I agree @alexandru's design ideas for how Futures can be improved, the standard library isn't really the place to experiment with those improvements, and anyway it isn't up for change until Scala 3.1 anyway. e.g. I have my own ideas of how we can extend scala.concurrent in a bunch of useful ways, but that kind of discussion is beyond the scope of this ticket.

6. That leaves switching between batching and non-batching executors using an environment variable or JVM property, as @eed3si9n proposed. I think that sounds reasonable.

   • People who want to try it in some super-high-performance asynchronous system to improve performance can do so, but others would be able to migrate their 2.11/2.12 code to 2.13 without issue.
   • We already have precedence for configuring ExecutionContext.global via scala.concurrent.context.maxExtraThreads and other system properties, so this isn't unheard of.
   • This is basically a process-global version of swapping between import ExecutionContext.Implicits.global and import ExecutionContext.Implicits.batched. Not pretty, but it'll do as a stopgap until the standard library is unfrozen.

7. In terms of documentation, as I mentioned to Matthew the documentation on how the ExecutionContext.global works is great https://docs.scala-lang.org/overviews/core/futures.html: precise, accurate, and tells me exactly how it behaves and what the limitations are. This doesn't stop us from documenting a new ExecutionContext.batching in the same way. As I mentioned above, I would prefer having a buffet of different ExecutionContexts each thoroughly documented and with clear instructions of what the different alternatives are and why you may choose each one.

That's my current understanding of our possible paths forward, feel free to correct anything that's mistaken :) I trust you guys have a better understanding of the tradeoffs than I do!

[mdedetrich]

I think thats reasonable although I personally find the JVM property configuration unnecessary. Just change the default ExecutionContext to what it was pre Scala 2.13.x but leave the batching one

[julienrf]

4. However, due to forward-and-backwards binary compatibility requirements, I do not believe this is possible for the next few years? I recall being told the Standard Library was frozen until Scala 3. Waiting until Scala 3.1 to solve the problem (2022? 2023?) seems like too long to wait for a regression of this sort.

We can put it in scala-library-compat, which only has to be backward compatible.

[lrytz]

If really needed we can add special compiler support so that scala.concurrent.ExecutionContext.batching translates to some bytecode that's binary compatible with 2.13. The JVM property solution seems rather awkward.

I like the idea of using scala-library-compat, however we almost decided to rename it back to scala-collection-compat and use it only for backports, not for new 2.13 library features. But we could use this incident as the starting point for a new scala-library-future library that contains things that will become part of the next standard library.