Topic/interrupt scope #1019

pchlupacek · 2017-12-15T18:01:51Z

Resolves #916

This still needs a cleanup and perhaps more cases.

Still the drained streams won't get interrupted.

@mpilquist @pchiusano @SystemFw pls feedback.

Thanks.

+ simplified concurrently definition with new Promise

mpilquist · 2017-12-15T18:48:23Z

core/shared/src/main/scala/fs2/internal/Algebra.scala

@@ -114,32 +130,45 @@ private[fs2] object Algebra {
    , g: (B, O) => B
    , v: FreeC.ViewL[Algebra[F,O,?], Option[(Segment[O,Unit], FreeC[Algebra[F,O,?],Unit])]]
  )(implicit F: Sync[F]): F[B] = {
-
+  //  println(s"RFL(${scope.id}): ${v.get}")


mpilquist · 2017-12-15T18:49:25Z

core/shared/src/main/scala/fs2/Stream.scala

-      haltWhenTrue.noneTerminate.either(self.noneTerminate).
-        takeWhile(_.fold(halt => halt.map(!_).getOrElse(false), o => o.isDefined)).
-        collect { case Right(Some(i)) => i }
+    def interruptWhen(haltWhenTrue: Stream[F,Boolean])(implicit F: Effect[F], ec: ExecutionContext): Stream[F,O] = {


Do we need a similar refactoring for pauseWhen?

I would like to go over all nondet combinators when that is done and clean them up. I think we may need to touch mergeXX as well. But prefferably I would like to do this as separate pr ?

mpilquist · 2017-12-15T18:51:54Z

core/shared/src/main/scala/fs2/internal/Algebra.scala

@@ -175,7 +202,8 @@ private[fs2] object Algebra {
            }

          case o: Algebra.OpenScope[F,_] =>
-            F.flatMap(scope.open) { innerScope =>
+            F.flatMap(scope.open(o.interruptible)) { innerScope =>
+              //println(s"RFL(${scope.id}): --- OPEN SCOPE -- ${innerScope.id} interruptible: ${innerScope.interruptible.nonEmpty}")


mpilquist · 2017-12-15T18:52:35Z

core/shared/src/main/scala/fs2/internal/RunFoldScope.scala

@@ -88,11 +97,12 @@ private[internal] final class RunFoldScope[F[_]] private (val id: Token, private
   * If this scope is currently closed, then the child scope is opened on the first
   * open ancestor of this scope.
   */
-  def open: F[RunFoldScope[F]] = {
+  def open(interruptible: Option[(Effect[F], ExecutionContext, Promise[F, Throwable])]): F[RunFoldScope[F]] = {
+   // println(s"OPENING new scope at $id")


mpilquist · 2017-12-15T19:00:08Z

core/shared/src/main/scala/fs2/internal/Algebra.scala

+    * Like `scope` but allows this scope to be interrupted.
+    * Note that this may fail with `Interrupted` when interruption occured
+    */
+  private[fs2] def interruptScope[F[_], O, R](pull: FreeC[Algebra[F,O,?],R])(implicit effect: Effect[F], ec: ExecutionContext): FreeC[Algebra[F,O,?],R] = {


Can we define both this and scope in terms of a new method that has the bulk of this body?

mpilquist

Looks great!

+ modified Stream#append to resume from intettruption + modified Stream#flatMap to not use Stream#append, but own interruptible version of append + added flatMap2 to FreeC

pchlupacek · 2017-12-16T19:06:29Z

@mpilquist pushed changes, I think this shall be good to go.
Now I will review mergeXXX pause, either etc to make sure all of them works as expected
That will come as another PRs

pchlupacek · 2017-12-16T19:08:02Z

@mpilquist of the final build failed, so I need to find out reason why again ...

mpilquist · 2017-12-19T15:19:53Z

core/shared/src/main/scala/fs2/internal/Algebra.scala

+          Either[Throwable,Any] => FreeC[Algebra[F,O,?], Unit]]
+        val fx = bound.fx.asInstanceOf[FreeC.Eval[Algebra[F,O,?],_]].fr
+        fx match {
+          case _: Algebra.CloseScope[F,_] => FreeC.Bind(FreeC.Eval(fx), { (_: Either[Throwable, Any]) => f(Left(e)) })


What if there's another CloseScope in the result of f(fx)?

Not sure if we need deep search this. I am quite certain we always get the closest to root scope that may be needed to close when flatMap fails. I was not able to come with stream structure where that would not be enough.

Why do we only care about the closest CloseScope?

mpilquist · 2017-12-19T15:27:44Z

core/shared/src/main/scala/fs2/internal/FreeC.scala

@@ -14,6 +14,11 @@ private[fs2] sealed abstract class FreeC[F[_], +R] {
      case Left(e) => FreeC.Fail(e)
    })

+  def flatMap2[R2](f: Either[Throwable, R] => FreeC[F, R2]): FreeC[F, R2] =


Rename to transformWith (matching naming convention used in other types)

mpilquist · 2017-12-19T15:30:15Z

core/shared/src/main/scala/fs2/internal/Algebra.scala

+    * Specific version of `Free.asHandler` that will output any CloseScope that may have
+    * been opened before the `rsn` was yielded
+    */
+  def asHandler[F[_], O](free: FreeC[Algebra[F,O,?],Unit], e: Throwable): FreeC[Algebra[F,O,?],Unit] = {


This seems too adhoc to me, like we're working around a flaw somewhere else in the design.

Well this is all realted to #1022. Any shortcuting of propert CloseScope causes the (A) resource leakage (B) interuption to be not possible. May be there is more princcipled approach, but that may // likely will require to touch algebra of either FreeC or Algebra itself

Re: #1022, handleErrorWith can be defined like I suggested if we want it to introduce a scope and hence, release resources acquired during that stream. Critically, there's no resource leak there at all, just a mismatched expectation as to scoping.

As a thought experiment, what would happen if ++ on Stream introduced a new scope? Would we be able to keep the old definition of flatMap then?

It is not unfortunatelly helping (the issue with handleErrorWith). I will bring you example of resource leak thats because of the mismatched scoping later. See the issue is that with interrupt we need to interrupt flatMaps that are defined in program AFTER the scope was defined, however stays within the scope of execution. i.e.

Stream(1).flatMap { i => xxx}.interruptWhen(...).scope // that will work Stream(1).interruptWhen(...).scope.flatMap { i => xxx} // thats the problem

Re ++'s scope. That was one of the earliest attempts. If we introduce scope with every ++, performance sucks.

Yes I understand performance will be unacceptable, but my point is that all this adhoc machinery is really just trying to make up for lack of scopes on each append in a use case specific way. I'm really not convinced we should be doing this.

Not sure I have good answer to that. If we won't we couldn't have interruption done properly (not just ++, but also error propagation) and that is cutting fs2 as option to define more complex programs. If we introduce scope to ++ we will sacrifice perfromance in place where we don't want to.

I'd like to get some thoughts from @pchiusano -- maybe he has some ideas.

mpilquist · 2017-12-19T15:32:41Z

core/shared/src/main/scala/fs2/Stream.scala

+              // in that case, in certain situations, the flatMap followed by `handleErrorWith` may
+              // without this change postpone closure of resources until the handler in handleErrorWith is interpreted.
+
+              def fby( s1: Either[O, Stream[F, O]], s2: => Stream[F,O2]): Stream[F, O2] = {


I'm not convinced we should ever be searching the tail for more work -- seems like this is patching for something else that's wrong.

yes, what wrong is that we skip all cleanups on error. Thats it. And in that case we rely on the cleanups to be run by Algebra.runFold.

If we would signal interrupt by exception, that's likely only way. We could perhaps decide it not to signal by exception but somehow build it in the FreeC itself, but I do not have a good solution for that either.

pchlupacek · 2017-12-20T09:53:10Z

@mpilquist I have couple of more ideas how to move further here. We certainly need to fix #1022, and then it seems it would perhaps start to work as it is.

pchlupacek · 2017-12-20T18:36:24Z

@mpilquist this is next iteration.
I have introduced here the onInterrupt to clearly define the demarcation point of the interruption.
Removed specific flatMap implementation.
Although all tests pass, I am not still 100% convinced that this is right approach. For example :

val s1 = Stream(1,0,2)
val s2 = Stream(3,4,5)
 (s1.interruptWhen(x) onInterrupt s2).flatMap { x => 
   Stream.eval(IO { x }) 
 }

will still fail if, the flatMap is evaluated during interruption

What do you think?

pchlupacek · 2017-12-20T18:42:04Z

@mpilquist hm, surprisingly the above example won't fail even when being interrupted in s1. Was my bad. So this seems to be pretty working well. Would be nice if we could bring some type of stream that will actually fail.

mpilquist · 2017-12-20T18:43:16Z

@pchlupacek This is looking really good, thanks for all the word you've done on this!

Can we move Interrupted to the top-level package since it's observable by clients now?

pchlupacek · 2017-12-20T18:43:50Z

yes and thanks for support on this one

pchlupacek · 2017-12-20T18:57:33Z

@mpilquist unfortunatelly this is still not good enough

(Stream(1,0,2).covary[IO].interruptWhen(interrupt.get.attempt).onInterrupt(Stream(3,4,5)))
    .flatMap { i => Stream.eval(promise.get.map { _ => i }) }
      .collect { case x => x }

will fail with Interrupted cause. If we remove the collect things works as expected.

I may need to look further how the pulls are altering the scope hierarchy, as this seems to be related to that.

mpilquist · 2017-12-20T19:03:32Z

The collect example should have the same scope structure as:

(Stream(1,0,2).covary[IO].interruptWhen(interrupt.get.attempt).onInterrupt(Stream(3,4,5)))
    .flatMap { i => Stream.eval(promise.get.map { _ => i }) }
      .scope

pchlupacek · 2017-12-25T09:05:31Z

@mpilquist , @SystemFw, pushed another iteration.
Good news :

onInterrupt is no longer necessary
When recovering from failure scopes are not closed and resources not released #1022 is resolved (and all other interruption issues too)
performance is fine, almost no impact (see bellow)
added many more tests for interruption (see specifically, no 15 and 16)

Bad news :

The def s: Stream[Pure,O] = Stream(o).flatMap { _ => s } cannot be interrupted, due to specific implementation in flatMap. See tests https://github.com/functional-streams-for-scala/fs2/pull/1019/files#diff-3744f1b1a64b6a321a31bf4386b77291R185 and proposed solution https://github.com/functional-streams-for-scala/fs2/pull/1019/files#diff-3744f1b1a64b6a321a31bf4386b77291R185 that unfortunately SoE.
The infinite streams (i.e. def s: Stream[Pure,O] = Stream(o) ++ s ) are interruptible, but requires a limit of asHandlers to be processed before no more stream elements are consulted. Limit is now scope based and may be configurable. I was not able to figure out something smarter.
The solution of When recovering from failure scopes are not closed and resources not released #1022 introduced noticeable overhead on handleErrorWith. That I think is acceptable, but worth of mentioning

Moving forward:

I really need to resolve that first issue with SoE an single element flatMap. This is show stopper, and I was not able to figure solution to that

I am sort of out of options how to improve this any further. So I'll appreciate any ideas guys. My guts are that to make this more cleaner, we will need to heavily alter algebra, and to be honest I even don't know how :-( atm.

The whole complexity comes from the Algebra.uncons that essentially causes to put cleanups to tail position of the returned stream. This causes all that mess that need to be solved with this PR. I was also thinking to bring uncons in Algebra, but hence it is sort of uncons and F ~> F2 I am not really sure if that would be possible at all .

Guys, i think we have to give this a priority, hence first user examples are showing that may be related to interruption (#1030) for example

…-for-scala/fs2 into topic/interrupt-scope

pchlupacek · 2018-01-03T07:23:22Z

@mpilquist @SystemFw after playing with this little bit more, I run out of ideas on more proper way how to implement interruption than it is currently done.
My guess is this shall work ok in all cases I have been thinking of, but I really couldn't find a way how to counter uncons behaviour any better, and uncons gives that much flexibility, so we don't want to get back from it I bet.

So If you agree, I would like you to do final review of this, and then we could merge.

If you are unhappy with current interrupt implementation, I bet the only way around it is to say that only arbitrary Throwable is propagated correctly (there we don't have to look up for the correct point to recover from) and leave interrupt in old implementation with all the drawbacks it may bring.

mpilquist · 2018-01-03T12:56:03Z

@pchlupacek A final review sounds good. I'll do that this morning and then we can merge after. Then hopefully get a milestone release out today.

mpilquist · 2018-01-03T13:01:15Z

core/jvm/src/test/scala/fs2/Pipe2Spec.scala

+    }
+
+    "interrupt (3)" in {
+      // tests the interruption of the constant stream


Super minor style nitpick -- these comments can be either put in to the test name -- e.g. intterupt (3) - interruption of constant stream or wrapped in an info("interruption of constant stream")

not sure about info syntax, can you just write quick example pls?

"interrupt (3)" in { info("tests interruption of the constant stream") ... }

It just prints the info string to the test output in test reports. In this case, it's probably better in the test name itself.

I have recently experimented with that, however info() causes to print out the text at every Gen case, so perhaps 100x or so for every test.
I am not sure about moving test cases to the names, not sure what that will do with tools that using names of tests to filter execution of only one single test case.

Ouch. Moving text description to test name works fine -- e.g., testOnly *MergeJoinSpec -- -z interrupt will run all the tests that include the word interrupt in their name

mpilquist · 2018-01-03T13:03:07Z

core/shared/src/main/scala/fs2/internal/CompileScope.scala

+  def acquireResource[R](fr: F[R], release: R => F[Unit]): F[Either[Throwable, (R, Token)]] = {
+    val resource = Resource.create
+    F.flatMap(register(resource)) { mayAcquire =>
+      if (!mayAcquire) F.raiseError(Interrupted(id, 0)) // todo: not sure if we shall signal by interrupted anymore


Oh I see -- we used Interrupted to signal this condition before but now Interrupted has meaning that's specific to interruptWhen. How about a new internal exception type here, AcquireAfterScopeClosed or something?

kk makes sense

SystemFw · 2018-01-03T13:03:22Z

core/shared/src/main/scala/fs2/Interrupted.scala

+
+/**
+  * Signals interruption of the evaluation. Contains id of last scope that shall be interrupted and
+  * any children of that scope.


perhaps we can reword a bit?
"Contains id of last scope that shall be interrupted along with its children"

Assuming that the meaning is "this will interrupt the scope with id scopeId and also interrupt all the children of said scope"

mpilquist · 2018-01-03T13:04:45Z

core/shared/src/main/scala/fs2/Interrupted.scala

+  *                 In each recursive iteration, this will increment by 1 up to limit defined in current scope,
+  *                 After which this will Interrupt stream w/o searching further for any cleanups.
+  */
+final case class Interrupted(scopeId: Token, loop: Int) extends Throwable {


Hm, the internal Token type leaks to public API here. Can we make scopeId and loop private vals?

I can make them private, however I was sort of thinking ot make the whole Interrupt() back to private again. I cannot imagine case where user will somehow catch/use the Interrupt. DO you think I should make it back private?

Okay yeah, that's fine with me.

mpilquist · 2018-01-03T13:05:35Z

core/shared/src/main/scala/fs2/Stream.scala


    /**
     * Run the supplied effectful action at the end of this stream, regardless of how the stream terminates.
     */
    def onFinalize(f: F[Unit])(implicit F: Applicative[F]): Stream[F,O] =
      Stream.bracket(F.pure(()))(_ => self, _ => f)

+


Extra whitespace line

mpilquist · 2018-01-03T13:05:41Z

core/shared/src/main/scala/fs2/Stream.scala

    /** Deprecated alias for `compile.last`. */
+


Extra whitespace line

mpilquist · 2018-01-03T13:07:17Z

core/shared/src/main/scala/fs2/internal/CompileScope.scala

+            , promise = Promise.unsafeCreate[F, Throwable](effect, ec)
+            , ref = Ref.unsafeCreate[F, (Option[Throwable], Boolean)]((None, false))
+            , interruptScopeId = newScopeId
+            , maxInterruptDepth = 256 // todo: want we somehow configure this ?


Can we write a separate issue for configuring this value along with maxSteps and chunkSize in Algebra?

are we ok with the side effects here (creating promise, ref, token)?
It's guarded by flatMap, but it's threading the line (however, it might be hard to write it otherwise)

@SystemFw they are guarded by modify2, I think we are on the safe side here. The scope allocation is pretty costly now. It is also quite often, i.e. every repeatPull (that is used by map for example) closes and opens scope. So I would like to leave it as it is.

mpilquist · 2018-01-03T13:21:46Z

core/shared/src/main/scala/fs2/internal/CompileScope.scala

@@ -178,6 +226,11 @@ private[internal] final class CompileScope[F[_]] private (val id: Token, private
    go(self, Catenable.empty)
  }

+  /** yields to true, if this scope has ancestor with given scope Id **/
+  def hasAncestor(scopeId: Token): F[Boolean] = {
+    F.map(ancestors) { c => Catenable.instance.exists(c)(_.id == scopeId) }


Note that if you import cats.implicits._, you can write this line as ancestors.map(_.exists(_.id == scopeId))

There's quite a lot of F.map and F.flatMap, but imho it hurts readability a bit due to all the parenthesis, so unless there's a specific reason not to, definitely 👍 to use cats syntax

Are we sure there is no extra allocation coming from the cats syntax implicit?

No, there's an allocation involved. I'm fine with the policy of fs2.internal not using cats syntax.

mpilquist · 2018-01-03T13:22:24Z

core/shared/src/main/scala/fs2/internal/CompileScope.scala

+  // See docs on [[Scope#interrupt]]
+  def interrupt(cause: Either[Throwable, Unit]): F[Unit] = {
+    interruptible match {
+      case None => F.raiseError(new Throwable("Scope#interrupt called for Scope that cannot be interrupted"))


s/Throwable/IllegalStateException/

mpilquist · 2018-01-03T13:26:13Z

core/shared/src/main/scala/fs2/internal/CompileScope.scala

+            // for every asynchronous child scope of the stream, so this assumption shall be completely safe.
+            F.flatMap(iCtx.promise.cancellableGet) { case (get, cancel) =>
+              F.flatMap(fs2.async.race(get, F.attempt(f))(iCtx.effect, iCtx.ec)) {
+                case Right(result) => F.map(cancel)(_ => result)


With cats syntax this becomes cancel.as(result)

was trying to avoid cats.syntax in whole CompileScope. Syntax is zero cost right?

Generally no -- some cats syntax is zero cost but anything provided via simulacrum is not. I'm fine with not using it in CompileScope to avoid allocations.

SystemFw · 2018-01-03T13:27:17Z

core/shared/src/main/scala/fs2/internal/CompileScope.scala

+
+  /**
+    * When the stream is evaluated, there may be `Eval` that needs to be cancelled early, when asynchronous interruption
+    * is taking the place.


s/taking the place/taking place

SystemFw · 2018-01-03T13:27:35Z

core/shared/src/main/scala/fs2/internal/CompileScope.scala

+  /**
+    * When the stream is evaluated, there may be `Eval` that needs to be cancelled early, when asynchronous interruption
+    * is taking the place.
+    * This allows to augment eval so whenever this scope is interupted it will return on left the reason of interruption.


s/interupted/interrupted

mpilquist · 2018-01-03T13:29:06Z

core/shared/src/main/scala/fs2/internal/Algebra.scala

 private[fs2] sealed trait Algebra[F[_],O,R]

+// Shared algebra between fold/uncons
+private[fs2] sealed trait AlgebraShared[F[_], O, R] extends Algebra[F, O, R]


I'd like a better name here - also, let's move this trait inside Algebra companion object.

How about Algebra.Effectful? Because the only constructors which aren't subtypes of it are Output, Run, and Uncons which by themselves are all non-effectful (stepping through an uncons may involve evaluating some nested effectful nodes of course)

Yes. I am still playing with Scope to have specific implementation for Uncons/Fold to resolve the interrupt. But yes lets name it effectfull

mpilquist · 2018-01-03T13:32:23Z

core/shared/src/main/scala/fs2/internal/Algebra.scala

+    , chunkSize: Int
+    , maxSteps: Long
+  )(implicit F: Sync[F]): F[(CompileScope[F], Option[(Segment[O,Unit], FreeC[Algebra[F,O,?],Unit])])] = {
+    F.delay(s.viewL.get) flatMap {


Do we need to wrap s.viewL.get in a delay constructor?

Yes, there are situation that will SoE without it. I think one of them was transform stack safety

mpilquist · 2018-01-03T13:33:22Z

core/shared/src/main/scala/fs2/internal/Algebra.scala

+    F.delay(s.viewL.get) flatMap {
+      case done: FreeC.Pure[Algebra[F,O,?], Unit] => F.pure((scope, None))
+      case failed: FreeC.Fail[Algebra[F,O,?], Unit] => F.raiseError(failed.error)
+      case bound: FreeC.Bind[Algebra[F,O,?], _, Unit] =>


This lost a cleanup change that replaced the existential with a type variable, which then reduced need for casting below. I can make this change again after we merge though.

no prob I put the existential there

mpilquist · 2018-01-03T13:35:32Z

core/shared/src/main/scala/fs2/internal/Algebra.scala

+                  compileFoldLoop(scope, output.values.fold(acc)(g).force.run._2, g, f(Right(())))
+                }
+                catch {
+                  case err: Throwable => compileFoldLoop(scope, acc, g, f(Left(err)))


Replace case err: Throwable with a case NonFatal(err) throughout here.

mpilquist · 2018-01-03T13:38:59Z

core/shared/src/main/scala/fs2/internal/Algebra.scala

-      case failed: FreeC.Fail[Algebra[F,O,?], _] => F.raiseError(failed.error)
+              case run: Algebra.Run[F, O, r] =>
+                try {
+                  val (r, b) = run.values.fold(acc)(g).force.run


Seems like we need to use chunkSize and maxSteps here to limit this call to run?

run shall always terminate after the fixes should it? Except when you define one that never terminates and as pure. If that would be the case, then, perhaps I will revert to code that stepped through uncons, but I don't think so it is necessary.

Stream.segment(Segment.constant(0)).covary[IO].to(Sink.showLinesStdOut[IO,Int]).compile.drain <-- I'd expect this to print stuff forever but as defined now, I don't think it will print anything.

The above example works fine with this branch (because to uses evalMap which uses flatMap which calls uncons. The result is the full infinite segment, which is then lazily folded an element at a time.

I think you can ignore the need for chunkSize and maxSteps here until/if I come up with an example that fails.

I think this shall work always, hence the uncons is implemented with splitAt, honoring the chunkSize and maxSteps, however the runFold is implemented in fold.

Also on the ChunkSize, wouldn't that mean here that chunk is always like 256 items? Just thinking loudly what that does with Stream[F, Byte]....

mpilquist · 2018-01-03T13:40:09Z

core/shared/src/main/scala/fs2/internal/Algebra.scala

@@ -193,6 +228,7 @@ private[fs2] object Algebra {
        case o: Output[F,O2] => Output[G,O2](o.values)
        case Run(values) => Run[G,O2,X](values)
        case Eval(value) => Eval[G,O2,X](u(value))
+        case un:Uncons[F,x,O2] => Uncons[G,x,O2](FreeC.suspend(un.s.translate(algFtoG)), un.chunkSize, un.maxSteps).asInstanceOf[Algebra[G,O2,X]]


Why is suspend needed here?

I had problems with SoE and transform. Perhaps with delaying viewL, this could be removed too.

mpilquist · 2018-01-03T17:48:12Z

Looks good to me. Ready for merge?

pchlupacek · 2018-01-03T17:50:24Z

Just pushed the changes.
I am still not giving up on resolving Interrupt in other way. However I believe we shall be in better shape that w/o this in 0.10, so I don't want to delay it.
I am hoping to come with solution that will resolve interruption solely in compilation, but still not sure of all the aspects how to implement this.
Would be excellent to do M release now. I will work then immediately on improving the concurrent primitives (i.e. pause, merge etc) that may benefit from the new interruption mechanics.

mpilquist · 2018-01-03T17:51:03Z

Sounds like a good plan to me. I'll merge and then release M11.

pchlupacek added 3 commits December 15, 2017 15:31

Implemented interruption via Scope#Interrupt

4f9661a

Fixed concurrently to not propagate Interrupt outside of its scope.

e09177b

+ simplified concurrently definition with new Promise

Added more test cases to test flatMapped hung streams to be interrupted

a9967d9

mpilquist reviewed Dec 15, 2017

View reviewed changes

mpilquist approved these changes Dec 15, 2017

View reviewed changes

Added resume interruption test

a85bead

+ modified Stream#append to resume from intettruption + modified Stream#flatMap to not use Stream#append, but own interruptible version of append + added flatMap2 to FreeC

Added more tests and corrected scope closure in flatMap

8757ea2

mpilquist reviewed Dec 19, 2017

View reviewed changes

Revereted flatMap implementation, introduced onInterrupt

750d062

Moved Interrupted to fs2._ root

c206629

pchlupacek added 2 commits December 24, 2017 06:48

Introduce id of the scope of the intterruption

3cc0adb

Append based interruption

a886ee6

Introduced Algebra.Uncons

2ab62d8

Merge branch 'topic/interrupt-scope' of github.com:functional-streams…

6f64b36

…-for-scala/fs2 into topic/interrupt-scope

mpilquist reviewed Jan 3, 2018

View reviewed changes

SystemFw reviewed Jan 3, 2018

View reviewed changes

mpilquist reviewed Jan 3, 2018

View reviewed changes

core/shared/src/main/scala/fs2/Stream.scala Outdated

/** Deprecated alias for `compile.last`. */

Copy link

Member

mpilquist Jan 3, 2018

Choose a reason for hiding this comment

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

Extra whitespace line

mpilquist reviewed Jan 3, 2018

View reviewed changes

SystemFw reviewed Jan 3, 2018

View reviewed changes

mpilquist reviewed Jan 3, 2018

View reviewed changes

Addresses review comments

71ee481

SystemFw approved these changes Jan 3, 2018

View reviewed changes

mpilquist merged commit 17f672a into series/0.10 Jan 3, 2018

This was referenced Jan 3, 2018

Nonemiting Pure Stream cannot be interrupted (stay to run in background) #1010

Closed

When recovering from failure scopes are not closed and resources not released #1022

Closed

pchlupacek deleted the topic/interrupt-scope branch August 21, 2018 10:10

Topic/interrupt scope #1019

Topic/interrupt scope #1019

Conversation

pchlupacek commented Dec 15, 2017

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

mpilquist left a comment

Choose a reason for hiding this comment

pchlupacek commented Dec 16, 2017

pchlupacek commented Dec 16, 2017

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

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

pchlupacek commented Dec 20, 2017

pchlupacek commented Dec 20, 2017

pchlupacek commented Dec 20, 2017 • edited Loading

mpilquist commented Dec 20, 2017

pchlupacek commented Dec 20, 2017

pchlupacek commented Dec 20, 2017

mpilquist commented Dec 20, 2017

pchlupacek commented Dec 25, 2017 • edited Loading

pchlupacek commented Jan 3, 2018

mpilquist commented Jan 3, 2018

Choose a reason for hiding this comment

pchlupacek Jan 3, 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

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

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

SystemFw Jan 3, 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pchlupacek commented Dec 20, 2017 •

edited

Loading

pchlupacek commented Dec 25, 2017 •

edited

Loading

pchlupacek Jan 3, 2018 •

edited

Loading

SystemFw Jan 3, 2018 •

edited

Loading