monix/monix

Asynchronous, Reactive Programming for Scala and Scala.js

Aleksandr Melnikov

@amelnikov-mylivn

Hello. I wolnder if monix 3.0.0-RC1 is compatiable with cats 1.4.0 and cats-effect 1.0.0?
I got two wierd exceptions:

java.lang.AbstractMethodError: Method monix/eval/instances/CatsConcurrentForTask$.bracketCase(Ljava/lang/Object;Lscala/Function1;Lscala/Function2;)Ljava/lang/Object; is abstract
    at monix.eval.instances.CatsConcurrentForTask$.bracketCase(CatsAsyncForTask.scala)
    at cats.effect.Bracket.bracket(Bracket.scala:64)
    at cats.effect.concurrent.Semaphore$AbstractSemaphore.withPermit(Semaphore.scala:244)

and

java.lang.ClassCastException: monix.eval.Task$Eval cannot be cast to cats.effect.IO
    at monix.eval.internal.TaskEffect$.$anonfun$cancelable$1(TaskEffect.scala:59)
    at monix.eval.internal.TaskEffect$.$anonfun$cancelable$1$adapted(TaskEffect.scala:52)
    at monix.eval.internal.TaskRunLoop$.executeAsyncTask(TaskRunLoop.scala:421)
    at monix.eval.internal.TaskRunLoop$.startFull(TaskRunLoop.scala:110)
    at monix.eval.internal.TaskRunLoop$RestartCallback.onSuccess(TaskRunLoop.scala:537)
    at monix.eval.internal.TaskRunLoop$.startFull(TaskRunLoop.scala:117)
    at monix.eval.internal.TaskMemoize$Register.$anonfun$registerListener$1(TaskMemoize.scala:132)
    at monix.eval.internal.TaskMemoize$Register.$anonfun$registerListener$1$adapted(TaskMemoize.scala:129)
    at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:60)
    at java.util.concurrent.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1402)
    at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
    at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
    at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
    at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)

I got this error after I added Semaphore[Task, Unit] usage in my code

Aleksandr Melnikov

@amelnikov-mylivn

Oh now I see that cats-effect 1.0.0 support is planned for 3.0.0-RC2
Googled by exceptions so missed this monix/monix#710
BTW it does not work with 3.0.0-RC2-d0feeba either

Oleg Pyzhcov

@oleg-py

@DamianReeves IIRC I made it so that monix-java just supports scala 2.11 in the end

Damian Reeves

@DamianReeves

thanks. will be helpful for those of us stuck in the past.

Alexander Khotyanov

@aksharp

@guizmaii a coworker had same exact problem today - canceling tasks/futures, although he's working with an app that uses futures and he used this as a guide to his solution https://viktorklang.com/blog/Futures-in-Scala-protips-6.html

Jules Ivanic

@guizmaii

@aksharp Thanks. I’ll take a look :)

Mateusz Górski

@goral09

@oleg-py @Avasil @alexandru I am looking for a place in the monix runLoop that imitates the green thread management. The docs say that operations are semantically blocking (no thread is blocked) so there must be logic that deals with yielding and resuming the work.

Oleg Pyzhcov

@oleg-py

@goral09 it's spread across TaskCreate, TaskRestartCallback and TaskRunLoop somewhat. There's no concept of green thread in the code, all it does is use Scheduler to, well, schedule the continuation.

Mateusz Górski

@goral09

Thanks @oleg-py

Oleg Pyzhcov

@oleg-py

Eventually, it boils down to usage of java's ScheduledExecutionService or plain JS setTimeout for scheduling

Fabio Labella

@SystemFw

@goral09 the implementation of green threading in monix and cats-effect is different from the one in scalaz

in scalaz IO, there is an actual type for a green thread, and IOs are scheduled manually by the runloop (with round robin last time I checked, but that was a while ago)

otoh monix and cats-effect just schedule the continuation by relying on the JVM (or JS) native tools

and they yield at async boundaries

and therefore Fiber here is just a "view" over the running computation, allowing you to wait on it (join) or interrupt it (cancel)

Fabio Labella

@SystemFw

and they yield at async boundaries

well, I guess earlier than that, since it's trampolined as well

Alexandru Nedelcu

@alexandru

The implementation of Scalaz IO cannot be that different, since it has to rely on the same primitives.

Fabio Labella

@SystemFw

I think at some point it actually wanted to move away from EC altogether

Mateusz Górski

@goral09

in theory, it (scalaz's) should be faster

Alexandru Nedelcu

@alexandru

Why do you say that?

Mateusz Górski

@goral09

because it doesn't have to go to EC and back?

Fabio Labella

@SystemFw

in theory, it (scalaz's) should be faster

citation needed

and also it does use an EC right now

Mateusz Górski

@goral09

@SystemFw are you looking at https://github.com/scalaz/scalaz-zio/blob/master/core/jvm/src/main/scala/scalaz/zio/RTS.scala ?

Alexandru Nedelcu

@alexandru

"Green threads" basically means units of work that do "cooperative multi-threading", which effectively means some way to execute multiple things without necessarily scheduling those on separate OS threads.

An executing Task or IO is essentially a "green thread".

Nothing magical about it.

Mateusz Górski

@goral09

yes, I get that part @alexandru , I was interested seeing in actual mechanics of how this is managed

Alexandru Nedelcu

@alexandru

When you end up with a "run loop" like we have, you've essentially overriden Java's call-stack, which means you have a green threads implementation on your hands.

Well, yeah, it's in TaskRunLoop. It's not going to be easy to read though, due to all the speed improvements.

Fabio Labella

@SystemFw

it's not that bad once you get used to the style

and all the "ugliness" is 100% necessary there

all the implementations have a similar strategy at the core: a current IO[Any], and a Stack[Any => IO[Any]]]

the runloop really is a run...loop

in Zio you have this bit of code

Mateusz Górski

@goral09

I wonder it this choice was an informed one (from scalaz-io RTS) :

val threadPool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors().max(2))

Fabio Labella

@SystemFw

 // Fiber does not need to be interrupted, but might need to yield:
              if (opcount == maxopcount) {
                // Cooperatively yield to other fibers currently suspended.
                // FIXME: Replace with the new design.
                opcount = 0

                // Cannot capture `curIo` since it will be boxed into `ObjectRef`,
                // which destroys performance, so we create a temp val here.
                val tmpIo = curIo

                rts.submit(evaluate(tmpIo))

                curIo = null

Alexandru Nedelcu

@alexandru

Btw, in Monix we also have Scheduler, which is capable of executing Runnable units on the current thread.

This is then used in the implementation for stack safety and error reporting.

Fabio Labella

@SystemFw

yeah, I misspoke slightly

the biggest difference is that Fiber in scalaz is a concrete type that holds the runloop logic

whereas in monix and cats effect is mostly a view over the runloop

Alexandru Nedelcu

@alexandru

Ah, right, that makes sense.

I have some thoughts on improving the run loop, but I don't have time for experiments right now. I'm thinking of introducing the "visitor pattern" we now have in Iterant too ... when you end up with an ADT with a lot of states, the JVM can do that pattern matching better than you via plain old method dispatches, however for tight flatMap loops pattern matching is faster (saw it in testing) so I have to juggle those modes somewhat ... to switch between method dispatches (visitor) and pattern matching in hot loops.

Not sure how well that will work out without experiments. Plain old method dispatches would be cool for adding more specialized states (ContextSwitch is a recent example).

Fabio Labella

@SystemFw

the key idea is the stack of continuations in any case, that's just a really powerful interpretation technique (which I suspect is related to CEK/CESK machines)

I think that approach makes sense

finding out the exact threshold it's going to be tough though :)

Alexandru Nedelcu

@alexandru

Doing performance work is kind of stressful, since you try out things and most of the times you discover that you're wrong. Not fun.

Fabio Labella

@SystemFw

:+1:

Fabio Labella

@SystemFw

I wonder it this choice was an informed one (from scalaz-io RTS) :
val threadPool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors().max(2))

@goral09 This is pretty std for CPU-bound work, when you have nonblocking implementations for most other things

Where communities thrive

People

Repo info

Activity