Skip to content

Fix Stream#parJoin for short-circuiting monad transformers - #2895

Merged
SystemFw merged 7 commits into
typelevel:mainfrom
yakivy:fix-parjoin-for-mt
May 10, 2022
Merged

Fix Stream#parJoin for short-circuiting monad transformers#2895
SystemFw merged 7 commits into
typelevel:mainfrom
yakivy:fix-parjoin-for-mt

Conversation

@yakivy

@yakivy yakivy commented Apr 30, 2022

Copy link
Copy Markdown
Contributor

PR fixes hang of the following stream:

case object TestException extends Throwable with NoStackTrace
type F[A] = EitherT[IO, Throwable, A]

Stream.range(1, 64).covary[F]
  .map(i => Stream.eval[F, Unit](
    if (i == 7) EitherT.leftT(TestException)
    else EitherT.right(IO.unit)
  ))
  .parJoin(4)
  .map(_ => Stream.eval[F, Unit](EitherT.right(IO.unit)))
  .parJoin(4)
  .compile
  .drain

Problem description:
For the given stream lease.cancel raises monad transformer error, so the following available.release >> decrementRunning chain is not invoked and therefore output stream is not closed. But this is only a part of the problem, even if we close the stream, it will swallow monad error and return successful result.

Fix description:
To fix the first part of the problem, I replaced >> monad composition with forceR, that is from MonadCancel so should work fine for short-circuiting monad transformers. To fix the second part, I raised lease.cancel failures into effect, separated success cancel with guaranteeCase and composed cancel result with the stream outcome oc <* Outcome.succeeded(fu).

yakivy added 2 commits April 30, 2022 14:50
Comment thread core/shared/src/main/scala/fs2/Stream.scala
Comment thread core/shared/src/test/scala/fs2/StreamParJoinSuite.scala Outdated
@armanbilge
armanbilge marked this pull request as draft May 3, 2022 18:45
Comment thread core/shared/src/main/scala/fs2/Stream.scala Outdated
@yakivy
yakivy marked this pull request as ready for review May 3, 2022 20:41
@yakivy
yakivy requested review from SystemFw and diesalbla May 3, 2022 20:43

@diesalbla diesalbla left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fine to me, but will let @SystemFw give it the final 👍🏼 and merge.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

4 participants