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
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
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
I usually write
F[_] style so I need the imports somewhat regularly
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
}
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)
}
}
at some point there's diminishing returns and I'm not sure how much additional
tick() time relates to added real execution time
@matfournier you can look at the implementation of
https://github.com/typelevel/cats-effect/blob/dd5ce27a120aca0618fd96dee5f4fc0214c291f8/core/shared/src/main/scala/cats/effect/Concurrent.scala#L602
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
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.
semaphore is totally fine for 99,9% cases
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.
it that doesn't matter to you, then it looks fine after a whole 10 seconds looking at it :)
thanks @SystemFw appreciate it.