{"action":"created","issue":{"url":"https://api.github.com/repos/typelevel/cats-effect/issues/121","repository_url":"https://api.github.com/repos/typelevel/cats-effect","labels_url":"https://api.github.com/repos/typelevel/cats-effect/issues/121/labels{/name}","comments_url":"https://api.github.com/repos/typelevel/cats-effect/issues/121/comments","events_url":"https://api.github.com/repos/typelevel/cats-effect/issues/121/events","html_url":"https://github.com/typelevel/cats-effect/pull/121","id":292147339,"number":121,"title":"Proof of concept: cancelable cats.effect.IO","user":{"login":"alexandru","id":11753,"avatar_url":"https://avatars2.githubusercontent.com/u/11753?v=4","gravatar_id":"","url":"https://api.github.com/users/alexandru","html_url":"https://github.com/alexandru","followers_url":"https://api.github.com/users/alexandru/followers","following_url":"https://api.github.com/users/alexandru/following{/other_user}","gists_url":"https://api.github.com/users/alexandru/gists{/gist_id}","starred_url":"https://api.github.com/users/alexandru/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/alexandru/subscriptions","organizations_url":"https://api.github.com/users/alexandru/orgs","repos_url":"https://api.github.com/users/alexandru/repos","events_url":"https://api.github.com/users/alexandru/events{/privacy}","received_events_url":"https://api.github.com/users/alexandru/received_events","type":"User","site_admin":false},"labels":[{"id":579526313,"url":"https://api.github.com/repos/typelevel/cats-effect/labels/enhancement","name":"enhancement","color":"84b6eb","default":true}],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":41,"created_at":"2018-01-27T19:49:04Z","updated_at":"2018-02-19T09:11:22Z","closed_at":null,"author_association":"COLLABORATOR","pull_request":{"url":"https://api.github.com/repos/typelevel/cats-effect/pulls/121","html_url":"https://github.com/typelevel/cats-effect/pull/121","diff_url":"https://github.com/typelevel/cats-effect/pull/121.diff","patch_url":"https://github.com/typelevel/cats-effect/pull/121.patch"},"body":"If there's consensus, this is a proof of concept for a cancelable `cats.effect.IO`.\r\n\r\nIt is inspired by [Monix](https://github.com/monix/monix/)'s `Task`, using approximately the same cancellation mechanism, minus some features.\r\n\r\nThis doesn't necessarily have to be approved, however the implementation is solid and I'm pushing it because I've come to the conclusion that the type class hierarchy in `cats-effect` is effectively crippled due to `IO` not being cancelable.\r\n\r\nI came to this conclusion after trying to describe this data type:\r\n\r\n```scala\r\ntrait Timer[F[_]] {\r\n  def async: Async[F]\r\n  def shift: F[Unit]\r\n\r\n  def sleep(timespan: FiniteDuration): F[Unit]\r\n}\r\n``` \r\n\r\nThis is basically the pure equivalent of [ScheduledExecutorService](https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ScheduledExecutorService.html) and of JavaScript's [setTimeout](https://developer.mozilla.org/en-US/docs/Web/API/WindowOrWorkerGlobalScope/setTimeout), an operation that's fundamental for *any platform* capable of asynchronous processing and that we need to describe in a pure way.\r\n\r\n## Implementation notes\r\n\r\n1. not all `IO` tasks are cancelable, only tasks built with `IO.cancelable` can cancel the evaluation (see below)\r\n  - should go without saying (after point 1) that `flatMap` chains are not auto-cancelable, and compared with Monix's `Task` there's no way to configure an `IO` task to be auto-cancelable\r\n  - if this is a problem, `flatMap` loops can be made cancelable by using `IO.cancelBoundary`, see below\r\n2. `IO` tasks that are cancelable, become non-terminating on `cancel`\r\n  - they can be turned into tasks that trigger an error with `onCancelRaiseError`, which can be used for implementing `bracket` or other operations, see below\r\n3. this PR does NOT break API compatibility\r\n4. this PR does NOT break semantics for current code, given that cancelable tasks can only be built with `IO.cancelable`, whereas tasks built with `IO.async`, `IO.apply` or other builders are not cancelable\r\n\r\n### Versus the \"async interruption\" from Haskell and Scalaz 8\r\n\r\nOn \"cancel\" what happens is that whatever you specify in the `IO.cancelable` builder gets executed. And depending on the implementation of an `IO.cancelable` task, it can become non-terminating. And our impure `cancel()` is a `() => Unit`.\r\n\r\nBy comparison Haskell and Scalaz sends an error, a `Throwable` on interruption and cancelled tasks get completed with that `Throwable`. Their impure cancel is a `Throwable => Unit`.\r\n\r\nWe on the other hand have an `onCancelRaiseError(e: Throwable)`, which can transform a task that's non-terminating on cancel, into one that raises an error on cancel. This operation also creates a race condition - cutting off the signaling to downstream, even if the source is uncancelable.\r\n\r\n`Throwable => Unit` allows the task's logic to know the cancellation reason, however I have yet to see a good use case for it — usually cancellation is about cutting the connection to the producer, closing all resources as soon as possible, because you're no longer interested in the result, due to some race condition that happened.\r\n\r\n`Throwable => Unit` is also a little confusing, being too broad in scope. Users might be tricked into sending messages back to the producer via this channel, in order to steer it, to change its outcome - however cancellation is cancellation, we're doing it for the purpose of releasing resources and the implementation of race conditions will end up closing the connection, disallowing the cancelled task to send anything downstream.\r\n\r\nTherefore it's confusing for the user and the only practical use is to release resources differently, based on the received error. But that's not a use-case I feel that's worth pursuing, given the increase in complexity.\r\n\r\n## IO.cancelable builder\r\n\r\n`IO.async` remains the same and we now have a new `IO.cancelable` builder - and in the provided function the user is expected to return a \"cancelable\", which is a `() => Unit` function that is supposed to cancel the started asynchronous computation. For example:\r\n\r\n```scala\r\ndef sleep(duration: FiniteDuration)(implicit sc: ScheduledExecutorService): IO[Unit] =\r\n  IO.cancelable { cb => \r\n    val c = sc.schedule(() => cb(Right()), duration.delay, duration.unit)\r\n    // Returning a function that can cancel our IO evaluation\r\n    () => c.cancel()\r\n  }\r\n```\r\n\r\n## start + Fiber + cancel as a pure action\r\n\r\n`IO` values that get started in the `IO` context (e.g. forked) get represented with a new interface:\r\n\r\n```scala\r\ntrait Fiber[F[+_], +A] {\r\n\r\n  def cancel: F[Unit]\r\n\r\n  def join: F[A]\r\n}\r\n```\r\n\r\nAnd `IO` now has a new `start` method on it exposing it:\r\n\r\n```scala\r\nsealed abstract class IO[+A] {\r\n\r\n  def start: IO[Fiber[IO, A]]\r\n}\r\n```\r\n\r\nA `Fiber` can be cancelled or joined, representing a background process that may run concurrently.\r\nNOTE: we don't fork threads automatically, we just start the `IO` and if the `IO` happens to run asynchronously, then we have actual parallelism, otherwise we don't.\r\n\r\nExample:\r\n\r\n```scala\r\nfor {\r\n  fiber <- IO.shift *> launchMissiles.start\r\n  _  <- runToBunker.handleErrorWith { e =>\r\n    // Retreating to bunker failed, abort!\r\n    fiber.cancel *> IO.raiseError(e)\r\n  }\r\n  aftermath <- fiber.join\r\n} yield {\r\n  aftermath\r\n}\r\n```\r\n\r\n## unsafeRunCancelable\r\n\r\n```scala\r\n// This is the sleep \r\nval io: IO[Unit] = sleep(10.seconds) *> IO.eval(println(\"Hello!\")\r\n\r\nval cancel: () => Unit =\r\n  io.unsafeRunCancelable(r => println(s\"Done: $r\"))\r\n\r\n// ... if a race condition happens, we can cancel it ...\r\ncancel()\r\n```\r\n\r\n## uncancelable marker\r\n\r\nGiven a cancelable `IO`, we can turn it into an `IO` that cannot be cancelled:\r\n\r\n```scala\r\n// Our reference from above\r\nval io: IO[Unit] = sleep(10.seconds) *> IO.eval(println(\"Hello!\")\r\n\r\n// This IO can't be canceled, even if we try\r\nio.uncancelable\r\n```\r\n\r\n## onCancelRaiseError\r\n\r\nNormally `IO` tasks that are cancelable, like our `sleep` above, become non-terminating on cancellation.\r\n\r\nGiven any `IO` (cancelable or not), the `onCancelRaiseError` operator creates a race condition between the source and cancellation ... and if cancellation happens, the cancel signal gets sent to the source and the resulting IO completes with the specified error.\r\n\r\n```scala\r\nimport java.util.concurrent.CancellationException \r\n\r\n// Our reference from above\r\nval io: IO[Unit] = sleep(10.seconds) *> IO.eval(println(\"Hello!\")\r\n\r\n// In case of cancellation, complete in error\r\nval io2 = io.onCancelRaiseError(new CancellationException(\"Boo\"))\r\n\r\nval cancel: () => Unit =\r\n  io.unsafeRunCancelable(r => println(s\"Done: $r\"))\r\n\r\n// ... if a race condition happens, we can cancel it ...\r\ncancel()\r\n// Prints:\r\n// Done: Left(CancellationException: Boo)\r\n```\r\n\r\n## IO.cancelBoundary\r\n\r\nReturns a cancelable boundary — an `IO` task that checks for the cancellation status of the run-loop and does not allow for the bind continuation to keep executing in case cancellation happened.\r\n\r\nThis operation is very similar to `IO.shift`, as it can be dropped in `flatMap` chains in order to make loops cancelable.\r\n\r\n```scala\r\ndef fib(n: Int, a: Long, b: Long): IO[Long] =\r\n  IO.suspend {\r\n    if (n <= 0) IO.pure(a) else {\r\n      val next = fib(n - 1, b, a + b)\r\n\r\n      // Every 100-th cycle, check cancellation status\r\n      if (n % 100 == 0)\r\n        IO.cancelBoundary *> next\r\n      else\r\n        next\r\n    }\r\n  }\r\n```\r\n\r\n## parMap2\r\n\r\nThis PR updates the `parMap2` implementation to cancel the other task if any of them fails — thus no longer waiting on both to complete in order to signal an error."},"comment":{"url":"https://api.github.com/repos/typelevel/cats-effect/issues/comments/366628341","html_url":"https://github.com/typelevel/cats-effect/pull/121#issuecomment-366628341","issue_url":"https://api.github.com/repos/typelevel/cats-effect/issues/121","id":366628341,"user":{"login":"alexandru","id":11753,"avatar_url":"https://avatars2.githubusercontent.com/u/11753?v=4","gravatar_id":"","url":"https://api.github.com/users/alexandru","html_url":"https://github.com/alexandru","followers_url":"https://api.github.com/users/alexandru/followers","following_url":"https://api.github.com/users/alexandru/following{/other_user}","gists_url":"https://api.github.com/users/alexandru/gists{/gist_id}","starred_url":"https://api.github.com/users/alexandru/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/alexandru/subscriptions","organizations_url":"https://api.github.com/users/alexandru/orgs","repos_url":"https://api.github.com/users/alexandru/repos","events_url":"https://api.github.com/users/alexandru/events{/privacy}","received_events_url":"https://api.github.com/users/alexandru/received_events","type":"User","site_admin":false},"created_at":"2018-02-19T09:11:22Z","updated_at":"2018-02-19T09:11:22Z","author_association":"COLLABORATOR","body":"@pchlupacek \r\n\r\n> To build the IO, that is cancellable, I need to use IO.cancellable, and then how I precisely cancel this ?\r\n\r\nAs a side-effectful action you get `unsafeRunCancellable`, which gives you a `() => Unit` back that can be cancelled. This is a low-level API for `IO` specifically and won't get exposed in type classes.\r\n\r\nUsually cancellation will happen under the hood on race conditions. For example the implementation of `parMap2` was changed in this PR to cancel the active task in case one of them finishes in error. \r\n\r\nAt some point we'll also have an `IO.race` that will cancel the loser. Based on that you could then express a `timeout` operation that cancels the source in case a certain timespan was exceeded.\r\nAll of these are race conditions for which cancellation happens automatically.\r\n\r\nNow there are instances in which an `IO` value gets exposed that's attached to a running process. For example this would be the result of a `start` operation, which in FS2 currently produces an `IO[IO[A]]`. These are `IO` values that are running concurrently and that the user might wish to cancel manually, therefore an operation like `start` is producing an `IO[Fiber[A]]`, where `Fiber` has a `cancel: IO[Unit]` on it.\r\n\r\nI think the best thing you can do right now is to download this branch and play with it.\r\n\r\n> I am also puzzled with api cancellable/uncancelable.\r\n\r\nWhat exactly is confusing?\r\n\r\nThe `uncancelable` operation does what it says. Consider this very simple analogy:\r\n\r\n```scala\r\ntype Run = () => (() => Unit)\r\n\r\ndef buildCancelable(): Run = {\r\n  ???\r\n}\r\n\r\ndef uncancelable(run: Run): Run = {\r\n  () => {\r\n    // execution\r\n    run()\r\n    // make cancellation a no-op\r\n   () => {}\r\n  }\r\n}\r\n```"}}
{"actor":{"display_login":"alexandru"}}