Cats-effect: Unexpected execution context after calling IO.fromFuture

[christianharrington]

I was surprised to find that after calling IO.fromFuture, computation is sometimes continued on the ExecutionContext that the Future was run on.

import cats.effect._
import scala.concurrent._
import scala.concurrent.ExecutionContext.global

// Slow Future
IO.fromFuture(IO(Future(Thread.sleep(500))(global)))
  .flatMap(_ => IO.delay(println(Thread.currentThread.getName)))
  .unsafeRunSync()

// Fast Future
IO.fromFuture(IO(Future(())(global)))
  .flatMap(_ => IO.delay(println(Thread.currentThread.getName)))
  .unsafeRunSync()

The above prints something like:

main

I looked through the commits, and there used to be a notice concerning this, but it was removed at some point.

I feel that the current behavior is a bit scary. I ran into it while using Slick, which maintains its own ExecutionContext for blocking operations. I was surprised when I suddenly saw some of my code executing there. I got around it by always shifting back after calling IO.fromFuture, but it took a bit of digging to find out what was happening.

I was wondering if this is the intended behavior. If it is, maybe the original notice should be restored?

[alexandru]

IO's execution model is different.

1. in your first sample you always get the execution happening on global because the future's callback gets registered before the completion of that future happens, which then gets called on global
2. in your second sample that Future's execution can be fast and converting from an already complete future ends up being the equivalent of IO.pure, which is why the execution can happen on main (but this is non-deterministic behavior, depending on whether that future is complete or not on evaluation)

IO.fromFuture(IO.pure(Future.successful("sample"))) <-> IO.pure("sample")

If you want to ensure that execution always shifts in the bind (flatMap) continuation, then indeed doing a manual shift after fromFuture is the right approach.

[christianharrington]

That makes total sense. Thanks for the explanation. If I open a PR that adds a note of that to the microsite documentation for IO.fromFuture and the ScalaDoc for IO.fromFuture, is that something that could be merged in?

[alexandru]

More documentation is always good 👍

[djspiewak]

It is worth noting that this particular problem is so pervasive that we have (in quasar) a bit of syntactic sugar implicitly added to the IO companion object which does the shift automatically.

There really are very few cases where you want the current fromFuture semantic. You almost always want to shift, otherwise your pools may become unbalanced over time (as more futures are converted) and it becomes quite difficult to track down the problem.

[alexandru]

@djspiewak indeed, I wonder what we should do in the future ...

Should we go back to requiring that ExecutionContext, as we did originally, or should we require a ContextShift?

[djspiewak]

ContextShift definitely. Following fs2's lead, our internal stuff has been settling into a pattern of "explicit ExecutionContext for blocking things, and implicit ContextShift representing the CPU pool". So then ContextShift actually does end up being coherent. We also use tagged types (i.e.BlockingContext, which is just ExecutionContext @@ Blocking) just to be extra explicit about it, to avoid issues with the wrong ExecutionContext being accidentally passed to the wrong thing.

[Daenyth]

@djspiewak I'd love to see a code sample of that if you can share.

[djspiewak]

@Daenyth Here's fromFutureShift: https://github.com/djspiewak/quasar/blob/master/foundation/src/main/scala/quasar/contrib/cats/effect/package.scala#L36-L39 Unfortunately, all of its use-sites are now closed-source, but you get the idea. You just call it like IO.fromFutureShift, the same way you might call IO.fromFuture.

BlockingContext is defined here, and here's a decent example of how it gets used.

[Daenyth]

Thanks very much!

[Gr0Bert]

For those who decided to seek an answer and dig a little bit:
The root of the issue is in the futures itself with the combination of the trampolined execution context. Trampolined execution context executes all stuff on the current thread.
When future's callback is registered after it's completion (via onComplete) - callback starts immediately on the passed executor, but our executor is trampolined thus - on the calling thread.
When future callback is registered before it's completion - there is another thread in action - the one from which complete was called and all callbacks are started to run within executors passed in onComplete, but our executors are trampolined - thus callbacks starting to run on the thread, from which complete was called.
You can see it from the following snippet:

object AB extends App {
  import scala.concurrent._
  import scala.concurrent.ExecutionContext.global
  import scala.concurrent.duration._

  val a = Promise.apply[Int]()
  val b = Future{
    Thread.sleep(500)
    a.complete(Success(2))
  }(global)
  a.future.onComplete(_ => println(Thread.currentThread().getName))(Execution.Implicits.trampoline)

  Await.ready(b, 1 second)
}

If you comment Thread.sleep(500) you will get different results from time to time.
Trampolined context is taken from https://github.com/yanns/trampoline-EC
Please, correct if I'm wrong in my statements.

[djspiewak]

@robgubin I'm pretty sure the root of the problem is considerably simpler: IO.fromFuture delegates continuation-thread responsibility to the Future. The fact that Future has nondeterministic semantics for continuations is deeply unfortunate, and I would argue not a particularly good design, but ultimately I believe it should be the responsibility of IO.fromFuture to smooth out that inconsistency.

If you would like to resolve the problem in user-space, a simple inclusion of IO.shift will resolve the issue:

IO.fromFuture(IO(myFuture) >> IO.shift

Assuming you have an implicit ContextShift[IO] in scope, this will convert myFuture into an IO, where the continuation of the IO is on a deterministic thread pool: namely, the one wrapped by the ContextShift.

[sloshy]

Leaving a note so that anyone who googles this can find a result:

Got tripped up by this today when using Slick with cats-effect. I was writing a very linear application that only makes a single DB call, and I did not want to pass ExecutionContext around everywhere when I didn't really need excessive concurrency so I used IO.fromFuture to process query results, sort of like this:

IO.fromFuture(IO(db.run(query.result))

The problem with this is, I got a java.lang.InterruptionException on some thread I did not recognize after I closed the DB connection. What I didn't realize at the time was that after the IO.fromFuture call I was actually executing still on the database connection thread from Slick's connection pool.

The workaround suggested here worked perfectly, btw. Thanks for discussing it in such detail so I can understand what's going on!

---

(as an aside, the thread name confused me because it was -1 and I thought I was executing in some nether-zone for a minute, but it was actually because Slick defaults the pool name to the name of your config path and I was passing in a raw config object, so the pool name was effectively "" + s"$threadNum" - a very silly mistake, but arguably it would have been helped if Slick had a much more transparent configuration API that did not make such weird assumptions [lesson learned, use doobie next time])