typelevel/cats-effect

Comfy Cats Effect room. Ask questions. Rant about config file formats. Be nice!

Gavin Bisesi

@Daenyth

@gvolpe I believe you need that for List(io1, io2).parSequence ?

whatever, compiler can assist people there if I'm off a little :p

if you have the import and don't need it it's better than you be missing it and not know what to do

Or do you mean that cats-core has the .parTupled etc syntax

oh huh yeah Parallel is in core

TIL

Paul Snively

@paul-snively

Right.

Gavin Bisesi

@Daenyth

I got confused because the static method versions of those for effects are on Concurrent for example

Paul Snively

@paul-snively

Understandable. :-D

Daniel Spiewak

@djspiewak

Generally regarding fibers and concurrency limits and stuff… Fibers are limited by memory, while threads are limited by… much more complex factors. As long as you use Blocker and IOApp correctly, your threads will be optimized for you, so you just need to worry about the memory footprint of all of your fibers. You can literally have tens of millions of them as long as they don't close over large state

Oleg Pyzhcov

@oleg-py

We have parSequenceN and the likes on Concurrent companion, and as syntax extensions, but they are par not in cats.Parallel sense, but rather in "assuming cats.Parallel means concurrent execution, that does execution with bounded concurrency"

But we don't have general parSequence or parTupled on Concurrent companion since those are from core

parTraverseN is just an encoding of a common "parTraverse with semaphore" pattern that people kept asking about all the time they didn't have problems with partial unification

Gavin Bisesi

@Daenyth

Right

Daniel Spiewak

@djspiewak

I feel like every time you start getting into "parTraverse with semaphore" territory, you probably need a richer concurrency framework than just Parallel. The cases get really tricky and it's easy to encode them incorrectly

Oleg Pyzhcov

@oleg-py

Sometimes you need, but sometimes you don't and parTraverse with a semaphore is exactly what you want :)

Gabriel Volpe

@gvolpe

Yeah I thought there was some changes I wasn't aware of cause normally you get that from cats.syntax.parallel._ or cats.implicits._ :smile:

Gavin Bisesi

@Daenyth

nope just my mistake

I usually write F[_] style so I need the imports somewhat regularly

Fabio Labella

@SystemFw

Against all odds (:P) I managed to write down the interruption proposal typelevel/cats-effect#681

Kai(luo) Wang

@kailuowang

Is there a microlib that can periodically run an effect, and allows to be paused and resumed? I am ready to write one but want to check if there is already one out there somewhere.

Christopher Davenport

@ChristopherDavenport

Cron4s, and I believe someone was PRing a pause to agitation

Kai(luo) Wang

@kailuowang

does cron4s provide the execution? or is it only a parser? I know of fs2cron.

Daniel Spiewak

@djspiewak

pause is an interesting one

Gavin Bisesi

@Daenyth

fs2 can enable that

It would be best if your pause/resume/time/whatever logic is simple, but complicated is doable with more effort on your end

The simple case is:

SignallingRef[F, Boolean](false).flatMap { paused =>
  val doStuff = fs2.Stream.eval(myEffect).repeat.pauseWhen(paused)
  pauseLogicStream(paused) concurrently doStuff
}

Kai(luo) Wang

@kailuowang

This message was deleted

Daniel Spiewak

@djspiewak

TIL

Gavin Bisesi

@Daenyth

I gotta say again that TestContext is super cool

I wrote a regression test for a deadlock we had, and just to confirm before I merged, broke the code again. Less than a second of clock time to reproduce the deadlock and detect it :D

using this btw;

  final override implicit def extremelyUnsafeIOAssertionToFuture(
      test: IO[Assertion]
  )(implicit pos: Position): Future[Assertion] = {

    val result: Future[Assertion] = test.unsafeToFuture()
    ctx.tick(1000.day) // Advance the clock

    if (result.value.isDefined)
      result
    else {
      fail(
        s"""Test probably deadlocked. Test `IO` didn't resolve after simulating 1000 days of time.
           | Remaining tasks: ${ctx.state.tasks}""".stripMargin
      )(pos)
    }
  }

Luka Jacobowitz

@LukaJCB

Why not go for something more absurd like 2000 years?

Gavin Bisesi

@Daenyth

I could run it to Long.MaxValue if I wanted :p

at some point there's diminishing returns and I'm not sure how much additional tick() time relates to added real execution time

Kai(luo) Wang

@kailuowang

Thanks @Daenyth that’s very cool.

matfournier

@matfournier

saw the above note about parTraverseN and now I'm curious --what does parTraverse w/ a semaphore look like? E.g. what were people doing before parTraverseN. Not because I want to use it but because I'm curious about library evolution and design.

Oleg Pyzhcov

@oleg-py

@matfournier you can look at the implementation of parTraverseN or parSequenceN which is pretty much what we were recommending before putting them into a lib:
https://github.com/typelevel/cats-effect/blob/dd5ce27a120aca0618fd96dee5f4fc0214c291f8/core/shared/src/main/scala/cats/effect/Concurrent.scala#L602

Piotr Gawryś

@Avasil

Other way was (and still is) to use a stream

And you could say that the next step is something like this: https://github.com/monix/monix/blob/master/monix-eval/shared/src/main/scala/monix/eval/internal/TaskGatherN.scala
parSequence + semaphore has a problem that it starts all IO in the list and then suspends them. If your list is like a 1 000 000 you can get OOM.

Oleg Pyzhcov

@oleg-py

if your list is 1 000 000, you are already in a bad spot :D

Piotr Gawryś

@Avasil

yeah, it's not something I had to ever deal with but someone had an issue

semaphore is totally fine for 99,9% cases

matfournier

@matfournier

@oleg-py thanks!

Kai(luo) Wang

@kailuowang

follow up on the repeating effect thingy. I wrote some 30 lines of code without using fs2. I tested it with some unit tests. Will appreciate if anyone here would take a quick look and point out blunders if any

def create[F[_]](
      effect: F[Unit],
      repeatDuration: FiniteDuration
    )(implicit F: Concurrent[F],
      T: Timer[F]
    ): F[Repeating[F]] = {
    Ref[F]
      .of(none[Fiber[F, Unit]])
      .map { ref =>
        new Repeating[F] {
          def pause: F[Boolean] =
            for {
              fiberO <- ref.modify(existing => (None, existing))
              success <- fiberO.traverse(_.cancel).map(_.isDefined)
            } yield success

          def resume: F[Boolean] = {
            def loop: F[Unit] =
              T.sleep(repeatDuration) >> effect >> loop

            running.ifM(
              false.pure[F],
              for {
                fiber <- F.start(loop)
                success <- ref.modify { existing =>
                  if (existing.isDefined) (existing, false)
                  else (Some(fiber), true)
                }
                _ <- if (!success) fiber.cancel else F.unit
              } yield success
            )
          }

          def running: F[Boolean] = ref.get.map(_.isDefined)
        }
      }
      .flatTap(_.resume)
}

The definition of Repeating is just as the instance suggested.

Fabio Labella

@SystemFw

not interruption safe I'm afraid

it that doesn't matter to you, then it looks fine after a whole 10 seconds looking at it :)

Kai(luo) Wang

@kailuowang

yeah, no interruption safety is my main issue. I couldnt think of a way to achieve it without semaphore. but maybe that’s a must

thanks @SystemFw appreciate it.

Fabio Labella

@SystemFw

I also have a style point if you'd indulge me, I think it's clearer to put the actions inside modify and then flatten, rather than returning booleans and then acting on them

Where communities thrive

People

Repo info

Activity