Parallel[IO, IO.Par] implementation #115

alexandru · 2018-01-12T18:18:52Z

Fixes #83

This adds:

internals.TrampolineEC: an internal ExecutionContext that's based on a trampoline, for "light async boundaries" in order to prevent stack overflow
the low level implementation of parMap2, review IOParMap.scala
a newtype called IO.Par, with an encoding inspired by alexknvl/newtypes
Parallel[IO, IO.Par] and Applicative[IO.Par] implementations

As a matter of implementation detail:

state is shared with an AtomicReference
it's based on getAndSet, which in terms of performance is pretty good (versus compareAndSet) because it uses platform intrinsics on Java 8
to remain stack safe, the implementation needs 2 light async boundaries (by means of TrampolineEC)
we are not forking any threads
it doesn't matter if one of the tasks fails, it still waits for the other to finish

Waiting for the other to finish, even in the face of error, is necessary in order to avoid leaks.
Consider this:

def loop(io1: IO[Int], io2: IO[Int]): IO[Int] =
  (io1, io2).parMap2(_ + _).handleWith(_ => loop(io1, io2))

If any of these IO values finish in error, if we don't wait for the other one, we've got ourselves a memory leak because the loop wouldn't back-pressure and it wouldn't cancel the other one. By waiting on both we avoid memory leaks and is the way to make this safe for IO.

Speaking of TrampolineEC, this raises the question — why isn't IO.async stack safe? Because we can definitely do it, no implicit ExecutionContext needed 😉

durban · 2018-01-12T20:17:42Z

core/js/src/main/scala/cats/effect/internals/TrampolineEC.scala

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2017 Typelevel


durban · 2018-01-12T20:23:16Z

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

+                case error @ Left(_) =>
+                  cb(error)
+                  rb match {
+                    case Left(error2) => throw error2


Doesn't this make it impossible to handle any errors in fb? Won't this simply kill a thread just because fb fails?

Right, was thinking that if the second error happens, then it needs to be handled somehow and we don't have a reporter.

Is ignoring this second error the right approach?

Or maybe a printStackTrace()?

I'm not sure. If f would have type (Attempt[A], Attempt[B]) => C, it could handle both errors. But that probably won't work with Parallel ...

I've been thinking about it and I think the right approach is to wrap both errors into a CompositeException and throw that one instead.

I've had this approach in Monix in cases where multiple operations have to be executed without interruption even if any of them fails — collecting thrown exceptions into a list, then wrapping that into a CompositeException and throwing that.

durban · 2018-01-12T20:25:07Z

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

+          fb.unsafeRunAsync { attemptB =>
+            // Using Java 8 platform intrinsics
+            state.getAndSet(Right(attemptB)) match {
+              case null => () // wait for B


wait for A?

alexandru · 2018-01-16T12:46:48Z

PR is ready, asking for feedback.

mpilquist · 2018-01-16T13:06:54Z

core/shared/src/main/scala/cats/effect/util/CompositeException.scala

+  * caught from evaluating multiple independent IO actions
+  * and that need to be signaled together.
+  */
+class CompositeException(val errors: List[Throwable])


FYI, in FS2 we have this same type but we require 2+ throwables in order to construct it. https://github.com/functional-streams-for-scala/fs2/blob/180b811083296762a2270afcef38d7fe4b278a1a/core/shared/src/main/scala/fs2/CompositeFailure.scala

👍 that's a good idea — might do that here as well.

Any reasonably sized project that cares about exception handling has this type 😜

ChristopherDavenport

I'm in favor something stronger than List for CompositeException, otherwise I think this looks good. Questions are to ensure I understand behavior correctly.

ChristopherDavenport · 2018-01-16T15:08:18Z

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

+                    case Right(b) =>
+                      cb(try Right(f(a, b)) catch { case NonFatal(e) => Left(e) })
+                    case error @ Left(_) =>
+                      cb(error.asInstanceOf[Left[Throwable, C]])


Why is this asInstanceOf necessary?

Because I want to reuse that value, as I know it's correct, but the compiler isn't smart enough 🙃

[error] cats-effect/core/shared/src/main/scala/cats/effect/internals/IOParMap.scala:48:26: type mismatch; [error] found : scala.util.Left[Throwable,B] [error] required: Either[Throwable,C] [error] cb(error) [error] ^

ChristopherDavenport · 2018-01-16T15:10:14Z

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

+            state.getAndSet(Left(attemptA)) match {
+              case null => () // wait for B
+              case Right(attemptB) => complete(attemptA, attemptB)
+              case left =>


These blocks are due to Either[Either[Throwable, A], Either[Throwable, B]] as we expect asynchronously to expect both values, if we already have a State of an A set it should be impossible to receive another?

So these cannot happen if the IO.async callback contract is respected, if that callback gets called at most once.

But the protocol can get violated, as the type system can't prevent it. Well, not by users because they are protected by that callback wrapper (IOPlatform.onceOnly) injected in IO.async. But if you workaround the library's encapsulation, or by mistakes from the library authors, then it's possible.

ChristopherDavenport · 2018-01-16T15:15:38Z

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

+                case Right(a) =>
+                  rb match {
+                    case Right(b) =>
+                      cb(try Right(f(a, b)) catch { case NonFatal(e) => Left(e) })


This is what restores the error handling behavior, as our case NonFatal(e) => Left(e) enables the recovery from the eventual possible failure conditions within IO.async?

IMO I wouldn't leave IO.async to catch exceptions. Currently the exception handling is reliable due to being protected by IOPlatform.onlyOnce, but that requires extra synchronization.

In Monix there is no onlyOnce backed by an AtomicReference, that check being only a plain variable. And in this case what happens is that the error simply gets reported with the provided Scheduler and thus it can't be recovered. From a usability perspective I find that reasonable, as the user is given a callback and you can't expect a Task to complete without ensuring that the callback gets called.

Also in this particular case that error is triggered asynchronously. It might happen from another thread, depending on the IO values it evaluated. So it won't be caught by IO.async.

johnynek · 2018-01-16T17:40:46Z

core/js/src/main/scala/cats/effect/internals/TrampolineEC.scala

+private[effect] final class TrampolineEC private (underlying: ExecutionContext)
+  extends ExecutionContext {
+
+  // Starts with `null`!


can you comment why vs justing using Nil? I guess you are using null to signal that you are not in a localRunLoop. Is that right? Can you comment?

Could it be NonEmptyList[Runnable] I wonder, and then use the null trick on that?

Yes, that's the difference between null and Nil, null signaling that we aren't in a run-loop and Nil signaling that a Runnable is in progress. This optimizes for the shallow case, as the first Runnable being executed doesn't get stored and retrieved from that List.

Since this isn't so obvious, guess at least a code comment is needed.

alexandru · 2018-01-17T06:27:30Z

Status — having problems with the newtype encoding for Scala 2.10, investigating options.

codecov-io · 2018-01-17T08:22:55Z

Codecov Report

Merging #115 into master will increase coverage by 0.87%.
The diff coverage is 60.29%.

@@            Coverage Diff             @@
##           master     #115      +/-   ##
==========================================
+ Coverage   88.49%   89.36%   +0.87%     
==========================================
  Files          20       23       +3     
  Lines         452      489      +37     
  Branches       41       36       -5     
==========================================
+ Hits          400      437      +37     
  Misses         52       52

alexandru · 2018-01-17T08:26:23Z

@mpilquist @durban I have now modified CompositeException to be like the one in FS2. In case you want to switch to this one in cats-effect for code reuse, I think we can add extra builders, but that can also come as a different PR.

Only difference is that I do prefer Exception as a suffix instead of Failure, because it's a well established JVM convention. Plus I made it inherit from RuntimeException, but I don't think that's a problem.

alexandru · 2018-01-17T08:26:48Z

I have now fixed the newtype encoding for Scala 2.10. It just required an indirection to avoid a small limitation of the compiler.

ChristopherDavenport

👍 on green. Since travis is misbehaving, harder to tell.

alexandru added 3 commits January 12, 2018 19:52

Add trampolined execution context

b801b56

Add internal parMap2 implementation

eecdf49

Oops, deleted tests by mistake

1c1afb4

durban reviewed Jan 12, 2018

View reviewed changes

Add newtype and Parallel instance

6f86cc9

alexandru changed the title ~~WIP: Parallel[IO, ?] implementation~~ Parallel[IO, IO.Par] implementation Jan 16, 2018

alexandru added 3 commits January 16, 2018 14:49

Fix comment

1a44533

Fix comment

2362fcb

Fix test for JS

aa38bd2

mpilquist reviewed Jan 16, 2018

View reviewed changes

mpilquist approved these changes Jan 16, 2018

View reviewed changes

Fix ScalaDoc

29a7e23

ChristopherDavenport reviewed Jan 16, 2018

View reviewed changes

johnynek reviewed Jan 16, 2018

View reviewed changes

Fix newtype for Scala 2.10, modify CompositeException

368cca3

alexandru added 2 commits January 17, 2018 10:33

Fix scaladoc

a91c9d8

Make FunctionK vals in Parallel[IO]

3f510f1

This was referenced Jan 17, 2018

Add a Parallel for Observable for using combineLatest monix/monix#536

Merged

Add newtype for Task.Par, using same encoding used in cats-effect monix/monix#538

Closed

Got rid of extension methods

73648aa

ChristopherDavenport approved these changes Jan 17, 2018

View reviewed changes

alexandru added 2 commits January 17, 2018 17:52

Fix Scaladoc

fd1352f

Fix Scaladoc

122405f

ChristopherDavenport merged commit 8c07424 into typelevel:master Jan 17, 2018

alexandru mentioned this pull request Feb 14, 2018

Should we provide a cats.Parallel instance for IO? #83

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Parallel[IO, IO.Par] implementation #115

Parallel[IO, IO.Par] implementation #115

alexandru commented Jan 12, 2018 •

edited

Loading

durban Jan 12, 2018

durban Jan 12, 2018

alexandru Jan 13, 2018

alexandru Jan 13, 2018

durban Jan 13, 2018

alexandru Jan 16, 2018

durban Jan 12, 2018

alexandru commented Jan 16, 2018

mpilquist Jan 16, 2018

alexandru Jan 16, 2018

ChristopherDavenport left a comment

ChristopherDavenport Jan 16, 2018

alexandru Jan 16, 2018

ChristopherDavenport Jan 16, 2018

alexandru Jan 16, 2018

ChristopherDavenport Jan 16, 2018

alexandru Jan 16, 2018

johnynek Jan 16, 2018

alexandru Jan 17, 2018

alexandru Jan 17, 2018

alexandru commented Jan 17, 2018

codecov-io commented Jan 17, 2018 •

edited

Loading

alexandru commented Jan 17, 2018

alexandru commented Jan 17, 2018

ChristopherDavenport left a comment

Parallel[IO, IO.Par] implementation #115

Parallel[IO, IO.Par] implementation #115

Conversation

alexandru commented Jan 12, 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

alexandru commented Jan 16, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ChristopherDavenport left a 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

alexandru commented Jan 17, 2018

codecov-io commented Jan 17, 2018 • edited Loading

Codecov Report

alexandru commented Jan 17, 2018

alexandru commented Jan 17, 2018

ChristopherDavenport left a comment

Choose a reason for hiding this comment

alexandru commented Jan 12, 2018 •

edited

Loading

codecov-io commented Jan 17, 2018 •

edited

Loading