Require Timer and shift automatically in concurrent IO operations #232

oleg-py · 2018-05-19T12:49:57Z

Fixes #229.

LukaJCB · 2018-05-19T14:46:08Z

core/shared/src/test/scala/cats/effect/internals/TestUtils.scala

+    val fakeErr = new PrintStream(outStream)
+    System.setErr(fakeErr)
+
+    Future.fromTry(Try(thunk)).flatten.andThen {


Flatten isn't available on 2.11 unless you import cats syntax :)

Thanks, will fix when I get to code :)

johnynek · 2018-05-19T18:45:11Z

core/shared/src/main/scala/cats/effect/IO.scala

-  def race[A, B](lh: IO[A], rh: IO[B]): IO[Either[A, B]] =
-    IORace.simple(lh, rh)
+  def race[A, B](lh: IO[A], rh: IO[B])(implicit timer: Timer[IO]): IO[Either[A, B]] =
+    IORace.simple(timer.shift *> lh, timer.shift *> rh)


I wonder if we should just take timer, lh, and rh directly on simple and delay or even avoid the allocation of *>

johnynek · 2018-05-19T18:48:01Z

core/shared/src/main/scala/cats/effect/IO.scala

-  final def start: IO[Fiber[IO, A @uncheckedVariance]] =
-    IOStart(this)
+  final def start(implicit timer: Timer[IO]): IO[Fiber[IO, A @uncheckedVariance]] =
+    IOStart(timer.shift *> this)


What if we just make IOStart(timer, this) so if we never actually evaluatue the result we save some allocations.

👍 agree with @johnynek

johnynek · 2018-05-19T18:49:03Z

core/shared/src/main/scala/cats/effect/IO.scala

-  def racePair[A, B](lh: IO[A], rh: IO[B]): IO[Either[(A, Fiber[IO, B]), (Fiber[IO, A], B)]] =
-    IORace.pair(lh, rh)
+  def racePair[A, B](lh: IO[A], rh: IO[B])(implicit timer: Timer[IO]): IO[Either[(A, Fiber[IO, B]), (Fiber[IO, A], B)]] =
+    IORace.pair(timer.shift *> lh, timer.shift *> rh)


Similar comment about just taking the timer here vs manually shifting.

alexandru · 2018-05-20T07:26:53Z

core/shared/src/main/scala/cats/effect/IO.scala

-  final def start: IO[Fiber[IO, A @uncheckedVariance]] =
-    IOStart(this)
+  final def start(implicit timer: Timer[IO]): IO[Fiber[IO, A @uncheckedVariance]] =
+    IOStart(timer.shift *> this)


👍 agree with @johnynek

alexandru · 2018-05-20T07:30:42Z

core/shared/src/main/scala/cats/effect/IO.scala

@@ -674,17 +671,63 @@ private[effect] abstract class IOLowPriorityInstances extends IOParallelNewtype
  }

  implicit def ioSemigroup[A: Semigroup]: Semigroup[IO[A]] = new IOSemigroup[A]
+
+  implicit val ioEffect: Effect[IO] = new IOEffect


The nice thing about this is that we get to keep the old ioEffect, with the previous type.

alexandru · 2018-05-20T07:31:10Z

core/shared/src/main/scala/cats/effect/IO.scala

+
+  implicit val ioEffect: Effect[IO] = new IOEffect
+
+  private[effect] class IOEffect extends Effect[IO] with StackSafeMonad[IO] {


WTF I did not know there's a StackSafeMonad in Cats since Jun 20, 2017.

It's not that known because it's not appearing in user code, I think. But yeah, it's nice to not copy-paste same tailRecM across all your types :)

alexandru · 2018-05-20T07:35:34Z

core/shared/src/main/scala/cats/effect/IO.scala

+    override def unit: IO[Unit] =
+      IO.unit
+
+    override def map[A, B](fa: IO[A])(f: A => B): IO[B] =


I would mark all of these defs as final.

Supposedly it helps the JVM optimize code. Never seen concrete evidence of it, but unless you want to leave the door open for overrides, there's no point in keeping them non-final.

alexandru · 2018-05-20T07:36:49Z

core/shared/src/main/scala/cats/effect/IO.scala

    import IO.Par.{unwrap, apply => par}

    override def pure[A](x: A): IO.Par[A] =
      par(IO.pure(x))
    override def map2[A, B, Z](fa: IO.Par[A], fb: IO.Par[B])(f: (A, B) => Z): IO.Par[Z] =
-      par(IOParMap(unwrap(fa), unwrap(fb))(f))
+      par(IOParMap(timer.shift *> unwrap(fa), timer.shift *> unwrap(fb))(f))


We might want to deploy the same trick I did in Monix for detecting already forked tasks.

But it doesn't have to happen in this PR, plus it probably needs good validation that it works in Monix first 🙂

oleg-py · 2018-05-20T10:41:19Z

@alexandru there's a problem with MVar tests on JS with auto-shifting, can you take a look? 😄

johnynek · 2018-05-20T15:50:04Z

core/shared/src/main/scala/cats/effect/internals/IOParMap.scala

-          IORunLoop.startCancelable(fa, connA, callbackA(connB))
-          IORunLoop.startCancelable(fb, connB, callbackB(connA))
+          IORunLoop.startCancelable(timer.shift *> fa, connA, callbackA(connB))
+          IORunLoop.startCancelable(timer.shift *> fb, connB, callbackB(connA))


Do we need to shift both? Isn’t shifting just fa enough?

johnynek · 2018-05-20T15:51:32Z

core/shared/src/main/scala/cats/effect/internals/IORace.scala

        case Right(a) =>
          onSuccess(active, connR, cb, Left(a))
        case Left(err) =>
          onError(active, cb, connR, err)
      })

      // Starts concurrent execution for the right value
-      IORunLoop.startCancelable[B](rh, connR, {
+      IORunLoop.startCancelable[B](timer.shift *> rh, connR, {


Why isn’t shifting just the left side enough?

It might be enough, but I'm not sure (my understanding of concurrency and internals is still a little bit off 😅 ), and I also didn't want to introduce a behavior (right-hand side start immediately on the current thread) which would break the symmetry between IO.race(fa, fb) and IO.race(fb, fa).map(_.swap).

And if optimizing it away would not be a breaking change, we can always do it later.

I agree. Also if you fork one thread, then execute the other immediately, it's not really a fair race 🙂

I think that we'll also be able to do an optimization like I introduced in Monix, where we can detect tasks that are forked for the obvious cases. For example we can detect shift *> delay(f) and prevent an extra shift.

... such optimizations are dirty though, I wouldn't do it in this PR. I'd rather merge the PR ASAP, because @mpilquist needs it in FS2.

johnynek · 2018-05-20T15:53:44Z

core/shared/src/test/scala/cats/effect/IOAppTests.scala


 class IOAppTests extends AsyncFunSuite with Matchers with BeforeAndAfterAll with TestUtils {
  test("exits with specified code") {
    IOAppPlatform.mainFiber(Array.empty, Eval.now(implicitly))(_ => IO.pure(ExitCode(42)))
      .flatMap(_.join)
      .unsafeToFuture
-      .value shouldEqual (Some(Success(42)))
+      .map(_ shouldEqual 42)


Shouldn’t this be an onComplete or something? If we get an exception will this be called?

Also, does the test framework just automatically handle Future? Why don’t we need to Await?

Test framework does. You return a Future[Assertion], and this is what I'm creating with map. A failed Future gets reported as a failed test.

gvolpe · 2018-05-21T11:54:59Z

site/src/main/tut/datatypes/io.md

@@ -1277,11 +1260,11 @@ There is also `cats.Traverse.sequence` which does this synchronously.

 ### parTraverse

-If you have a list of data and a way of turning each item into an IO, but you want a single IO for the results you can use `parTraverse` to run the steps in parallel. The IO tasks must be asynchronous, which if they are not you can use shift.
+If you have a list of data and a way of turning each item into an IO, but you want a single IO for the results you can use `parSequence` to run the steps in parallel.


parTraverse?

Thanks for noticing 👍

- Reverted misfix in documentation. Added note on creating Timer[IO]s. - Fixed failing MVar tests - Moved timer.shift from IO.start deeper to IOStart

codecov-io · 2018-05-22T13:47:10Z

Codecov Report

Merging #232 into master will increase coverage by 0.92%.
The diff coverage is 95.74%.

@@            Coverage Diff             @@
##           master     #232      +/-   ##
==========================================
+ Coverage   87.81%   88.73%   +0.92%     
==========================================
  Files          60       58       -2     
  Lines        1494     1492       -2     
  Branches      148      145       -3     
==========================================
+ Hits         1312     1324      +12     
+ Misses        182      168      -14

Require Timer and shift automatically in concurrent IO operations

0934ef3

LukaJCB reviewed May 19, 2018

View reviewed changes

mpilquist self-requested a review May 19, 2018 14:59

mpilquist approved these changes May 19, 2018

View reviewed changes

johnynek reviewed May 19, 2018

View reviewed changes

oleg-py mentioned this pull request May 20, 2018

Cancelation laws presuppose no interruption #192

Closed

alexandru reviewed May 20, 2018

View reviewed changes

oleg-py added 3 commits May 20, 2018 11:41

Undo Future silencing, use immediate EC in IOAppTests

f07fcee

Make methods in TC instances final

1729f6b

Move "shift *>" down the call stack; fix nondeterministic test failure

5107d19

johnynek reviewed May 20, 2018

View reviewed changes

gvolpe reviewed May 21, 2018

View reviewed changes

oleg-py added 2 commits May 22, 2018 16:23

Final batch of fixes

a3b9e3b

- Reverted misfix in documentation. Added note on creating Timer[IO]s. - Fixed failing MVar tests - Moved timer.shift from IO.start deeper to IOStart

Silence MiMa checks

010d6d7

alexandru approved these changes May 22, 2018

View reviewed changes

ChristopherDavenport approved these changes May 22, 2018

View reviewed changes

alexandru added this to the 1.0.0-RC2 milestone May 23, 2018

alexandru merged commit 8ed6e71 into typelevel:master May 23, 2018

oleg-py mentioned this pull request May 23, 2018

Move uncancelable down to Bracket. Add laws specifying interaction. #241

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Require Timer and shift automatically in concurrent IO operations #232

Require Timer and shift automatically in concurrent IO operations #232

oleg-py commented May 19, 2018

LukaJCB May 19, 2018

oleg-py May 19, 2018

johnynek May 19, 2018

johnynek May 19, 2018

alexandru May 20, 2018

johnynek May 19, 2018

alexandru May 20, 2018

alexandru May 20, 2018

alexandru May 20, 2018

oleg-py May 20, 2018

alexandru May 20, 2018

alexandru May 20, 2018

oleg-py commented May 20, 2018

johnynek May 20, 2018

johnynek May 20, 2018

oleg-py May 20, 2018 •

edited

Loading

alexandru May 22, 2018

alexandru May 22, 2018

johnynek May 20, 2018

oleg-py May 20, 2018

gvolpe May 21, 2018

oleg-py May 21, 2018

codecov-io commented May 22, 2018


		implicit val ioEffect: Effect[IO] = new IOEffect

		private[effect] class IOEffect extends Effect[IO] with StackSafeMonad[IO] {

Require Timer and shift automatically in concurrent IO operations #232

Require Timer and shift automatically in concurrent IO operations #232

Conversation

oleg-py commented May 19, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

oleg-py commented May 20, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

oleg-py May 20, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov-io commented May 22, 2018

Codecov Report

oleg-py May 20, 2018 •

edited

Loading