StackOverflowError in chained Future.flatMap calls

[scabug]

First, a simple reproduction of the bug:

val promise = Promise[Int]
List.range(0, 1000).map(i => Future(i)).foldLeft(promise.future)((f1, f2) => f2.flatMap(i => f1))
promise.success(-1)

This will throw a crazy exception with 1000 causes, the root being a StackOverflowError. Note if running this in the repl, or anywhere, you may instead get NoClassDefFoundErrors. This is because something is trying to handle the initial error deep in the stack (probably shouldn't be), which triggers a class to be loaded, which throws another StackOverflowError, which results in the NoClassDefFoundError.

Here's the implementation of flatMap, which clearly shows the issue:

def flatMap[S](f: T => Future[S])(implicit executor: ExecutionContext): Future[S] = {
  val p = Promise[S]()

  onComplete {
    case f: Failure[_] => p complete f.asInstanceOf[Failure[S]]
    case Success(v) =>
      try {
        f(v).onComplete({
          case f: Failure[_] => p complete f.asInstanceOf[Failure[S]]
          case Success(v) => p success v
        })(internalExecutor)
      } catch {
        case NonFatal(t) => p failure t
      }
  }(executor)

  p.future
}

The internalExecutor executes the onComplete callback in the same thread. So, when promises returned by flatMap are used as the promise to be returned by another flatMap callback function, and this is done enough times, you get a StackOverflowError.

This is not an uncommon situation in iteratees, which use many many futures and often chain them together in very long chains (for very long streams), and both the Play core developers and Play users have found this to be an issue all over the place. Currently we fix it with a call like this:

.flatMap(a => Future.successful(a))

But I don't think this is the right solution to the problem, and it seems to come up all over the place.

Suggested solution is to use an execution context that doesn't redeem the flatMap promise is the same thread.

[scabug]

Imported From: https://issues.scala-lang.org/browse/SI-6932?orig=1
Reporter: @jroper
Affected Versions: 2.10.0
Other Milestones: 2.10.1-RC1

[scabug]

@retronym said (edited on Jan 7, 2013 1:48:51 PM UTC):
See my comments on scala/scala#1686 (comment), in which NonFatal was complicit in a similar story.

[scabug]

@retronym said:
Not sure if this is a home run idea, but we might want to eagerly load NonFatal to avoid obscuring these problems.

retronym/scala@scala:2.10.x...retronym:topic/eager-load-non-fatal-2

[scabug]

@jroper said:
Yes, NonFatal is the class that wasn't loaded in the repl, and I got around that by loading it as soon as I started the repl. Agreed it would be a good idea to eagerly load it.

[scabug]

@viktorklang said:
Proposed fix is here: https://github.com/viktorklang/scala/pull/5/commits

[scabug]

@viktorklang said:
We need to try to get this fixed in 2.10.1 and also backport the fix to the backport of SIP-14 for 2.9.3 (If possible)

[scabug]

@retronym said:
scala/scala#1941

[scabug]

@adriaanm said (edited on Jan 24, 2013 11:17:24 PM UTC):
for 2.9.3-RC2: scala/scala#1962

[scabug]

@pchiusano said:
I've come across another way of fixing this problem, which is to build a 'trampolined Future' type that associates its flatMap calls to the right, similar the trampoline data type that Runar has written up. The interplay between the trampolining and the asynchronous computation is pretty subtle and tricky to get right, but it's only like 50 LOC. Here's a working gist illustrating the idea. There's some examples in there too, including the example that was problematic above. One advantage to it is that new tasks are only submitted to the thread pool when explicitly forked (notice that flatMap does not require an implicit ExecutorService or anything), rather than on every flatMap call, which I suspect is going to be much faster for many use cases.

[scabug]

@viktorklang said:
"One advantage to it is that new tasks are only submitted to the thread pool when explicitly forked "

The reason we didn't go that route is that we consider that to be a disadvantage, i.e. the programmer has to make choices about execution rather than flow.

[scabug]

@pchiusano said:
I thought about this some more, and I believe the degree of parallelism is identical with the approach I gave (updated version below). I am just reusing the spawned thread for the 'rest' of the computation and avoiding a needless task submit cycle. That is, if I have a f.flatMap(g), the Future produced by g must be run in sequence after f. Therefore, there is no advantage to always spawning a separate logical thread to run the Future produced by g - we would be better off just reusing the thread that f was just about to relinquish, unless g explicitly indicates it wishes to refork.

Here's updated code, if you are interested:

https://github.com/pchiusano/fpinscala/blob/master/answers/src/main/scala/fpinscala/iomonad/Future.scala

It is a different library than what you have in the sense that Future does not necessarily represent a running computation. Until you call start, run, or runAsync, nothing is happening. This greatly simplifies the implementation - there are no race conditions to worry about, where a listener registers itself at the same time the promise is completing on its own.

I'll probably port this to scalaz pretty soon so we can see how it works out there.