Introduce GenDeferred

[armanbilge]

This PR introduces GenDeferred as a generalization of Deferred:

trait GenDeferred[F[_], G[_], A] extends DeferredSource[F, A] with DeferredSink[G, A]

Conceptually, this enables two independent effect systems to communicate/synchronize via a shared data structure.

In practice, there are often situations where it is useful to instantiate a GenDeferred where F is any Async type and G is SyncIO when interopping with unsafe code. This makes it possible to complete the deferred "in-place" via SyncIO#unsafeRunSync without requiring a Dispatcher and avoiding a thread-shift to the compute pool to run this trivial operation.

I would like to follow-up with a similar GenQueue[F, G, A] such that an unbounded queue can be implemented for Async[F] and Sync[G].

Example use-cases include fs2-grpc (cc @ahjohannessen) and fs2-io.js which both rely heavily on Deferred and Queue to interop with unsafe code.

Credit to @ChristopherDavenport for first exploring this idea in https://github.com/davenverse/condemned.

[durban]

I didn't look too closely, but I think I've already did this in #1583 (it wasn't merged).

[armanbilge]

@durban thanks for chiming in, yes the concept looks the same :) apologies that I wasn't aware of your prior art.

One year on, with at least two applications, I wonder if opinions have changed? 😅

[ahjohannessen]

I hope this and the follow up could land in cats-effect.
Library authors apparently have a need for something like this. I mean @armanbilge does this, @ChristopherDavenport makes condemned and fs2-grpc could use this.

[durban]

One year on, with at least two applications, I wonder if opinions have changed?

My opinion haven't changed: I still think it's a good idea 😄 (Although as it turned out, my original use case required even more than this...)

[bplommer]

Chiming in with another closed PR in the same vicinity: #1889 is a bit unfocused, but among other things it takes a slightly different approach of just letting you change the effect type of an existing SyncDeferred/AsyncDeferred to an arbitrary effect with the relevant instance. It looks like your approach is simpler to implement though 😅

[armanbilge]

Wow, another one! Thanks for the pointer. It seems you were actually able to execute the changes for Queue as well which is awesome since I shied away from it here. Also Daniel seemed slightly less skeptical in that PR which is a big win in of itself 😉

[bplommer]

It seems you were actually able to execute the changes for Queue as well which is awesome since I shied away from it here.

Ah yes - not sure whether the Deferred part is any more complex than your version 🤔

[armanbilge]

Yes, I held off since we are getting a real AsyncQueue as part of #2771. So it doesn't make sense to rework the current Concurrent implementation.

[SystemFw]

Quick comment I've already mentioned on Discord: I don't have issues with the idea, but I'd rather we didn't have the subtyping hierarchy

[armanbilge]

Big thanks to Daniel for patiently walking me through exactly what happens when you complete a Deferred and why doing this outside of the compute pool (i.e., not via Dispatcher) can actually make things worse.

A key part of completing a Deferred is notifying any waiting readers.

cats-effect/kernel/shared/src/main/scala/cats/effect/kernel/Deferred.scala

Lines 157 to 158 in f423ac7

	def complete(a: A): F[Boolean] = {
	def notifyReaders(readers: LongMap[A => Unit]): F[Unit] = {

To do so, you fulfill a callback next with the completed value a, for each reader.

cats-effect/kernel/shared/src/main/scala/cats/effect/kernel/Deferred.scala

Line 167 in f423ac7

val task = F.delay(next(a))

The callback function itself is ultimately defined here.

cats-effect/core/shared/src/main/scala/cats/effect/IOFiber.scala

Line 626 in f423ac7

val cb: Either[Throwable, Any] => Unit = { e =>

Which involves scheduling the semantically-blocked fiber in question.

cats-effect/core/shared/src/main/scala/cats/effect/IOFiber.scala

Line 667 in f423ac7

scheduleFiber(ec, this)

Which takes us to here.

cats-effect/core/shared/src/main/scala/cats/effect/IOFiber.scala

Lines 1284 to 1287 in f423ac7

	private[this] def scheduleFiber(ec: ExecutionContext, fiber: IOFiber[_]): Unit = {
	if (ec.isInstanceOf[WorkStealingThreadPool]) {
	val wstp = ec.asInstanceOf[WorkStealingThreadPool]
	wstp.scheduleFiber(fiber)

And finally, to here (assuming our fiber is indeed assigned to the WSTP).

cats-effect/core/jvm/src/main/scala/cats/effect/unsafe/WorkStealingThreadPool.scala

Lines 418 to 431 in f423ac7

	private[effect] def scheduleFiber(fiber: IOFiber[_]): Unit = {
	val pool = this
	val thread = Thread.currentThread()
	if (thread.isInstanceOf[WorkerThread]) {
	val worker = thread.asInstanceOf[WorkerThread]
	if (worker.isOwnedBy(pool)) {
	worker.schedule(fiber)
	} else {
	scheduleExternal(fiber)
	}
	} else {
	scheduleExternal(fiber)
	}

And suddenly, it's obvious why it makes a huge difference whether the current thread we are running on belongs to the compute pool (where Dispatcher places us) or we are completing these callbacks from an external thread. It's the difference between scheduling on the same worker thread vs submitting to the WSTP external queue.

In summary: the context shift is fundamental as Daniel has been saying all along. So, better to get on the compute pool ASAP.

Sadly, this still leaves JavaScript in an awkward place since there is no WSTP-fastpath. The initial yield is wasteful since it's not taking us to a better place like it does on JVM. But I'm not sure if that's good enough reason to muck up the hiearchy with this stuff.

Note: edited to clarify the final hop to actually scheduling on the WSTP.

[djspiewak]

@armanbilge What if we have a JS specialization of Dispatcher?

[armanbilge]

Sure, but what would it do / how would it work?

[armanbilge]

Trying a different idea in #2835.