Topic/interrupt scope

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

@@ -2,6 +2,7 @@ package fs2
 
 import scala.concurrent.duration._
 import cats.effect.IO
+import fs2.async.Promise
 
 class ConcurrentlySpec extends Fs2Spec {
 
@@ -13,7 +14,7 @@ class ConcurrentlySpec extends Fs2Spec {
 
     "when background stream fails, overall stream fails" in forAll { (s: PureStream[Int], f: Failure) =>
       val prg = Scheduler[IO](1).flatMap(scheduler => (scheduler.sleep_[IO](25.millis) ++ s.get).concurrently(f.get))
-      val throws = f.get.drain.compile.drain.attempt.unsafeRunSync.isLeft
+      val throws = f.get.compile.drain.attempt.unsafeRunSync.isLeft
       if (throws) an[Err.type] should be thrownBy runLog(prg)
       else runLog(prg)
     }
@@ -33,5 +34,17 @@ class ConcurrentlySpec extends Fs2Spec {
       runLog(prg)
       bgDone shouldBe true
     }
+
+    "when background stream fails, primary stream fails even when hung" in forAll { (s: PureStream[Int], f: Failure) =>
+      val promise = Promise.unsafeCreate[IO, Unit]
+      val prg = Scheduler[IO](1).flatMap{  scheduler =>
+        (scheduler.sleep_[IO](25.millis) ++ (Stream(1) ++ s.get)).concurrently(f.get)
+        .flatMap { i => Stream.eval(promise.get).map { _ => i } }
+      }
+
+      val throws = f.get.compile.drain.attempt.unsafeRunSync.isLeft
+      if (throws) an[Err.type] should be thrownBy runLog(prg)
+      else runLog(prg)
+    }
   }
 }

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

@@ -1,5 +1,6 @@
 package fs2
 
+
 import scala.concurrent.duration._
 import cats.effect.IO
 import cats.implicits._
@@ -150,17 +151,70 @@ class Pipe2Spec extends Fs2Spec {
 
     "interrupt (1)" in forAll { (s1: PureStream[Int]) =>
       val s = async.mutable.Semaphore[IO](0).unsafeRunSync()
-      val interrupt = Stream.emit(true) ++ Stream.eval_(s.increment)
+      val interrupt = mkScheduler.flatMap { _.sleep_[IO](50.millis) }.compile.drain.attempt
       // tests that termination is successful even if stream being interrupted is hung
       runLog { s1.get.covary[IO].evalMap(_ => s.decrement).interruptWhen(interrupt) } shouldBe Vector()
     }
 
     "interrupt (2)" in forAll { (s1: PureStream[Int]) =>
+      val s = async.mutable.Semaphore[IO](0).unsafeRunSync()
+      val interrupt = Stream.emit(true) ++ Stream.eval_(s.increment)
+      // tests that termination is successful even if stream being interrupted is hung
+      runLog { s1.get.covary[IO].evalMap(_ => s.decrement).interruptWhen(interrupt) } shouldBe Vector()
+    }
+
+    "interrupt (3)" in {
+      // tests the interruption of the constant stream
+      val interrupt = mkScheduler.flatMap { _.sleep_[IO](20.millis) }.compile.drain.attempt
+      Stream.constant(true).covary[IO].interruptWhen(interrupt).compile.drain.unsafeRunSync
+    }
+
+    "interrupt (4)" in {
+      // tests the interruption of the constant stream with flatMap combinator
+      val interrupt = mkScheduler.flatMap { _.sleep_[IO](20.millis) }.compile.drain.attempt
+      Stream.constant(true).covary[IO].interruptWhen(interrupt).flatMap { _ => Stream.emit(1) }.compile.drain.unsafeRunSync
+    }
+
+    "interrupt (5)" in {
+      // tests the interruption of the stream that recurses infinitelly
+      val interrupt = mkScheduler.flatMap { _.sleep_[IO](20.millis) }.compile.drain.attempt
+      def loop(i: Int): Stream[IO, Int] = Stream.emit(i).covary[IO].flatMap { i => Stream.emit(i) ++ loop(i+1) }
+      loop(0).interruptWhen(interrupt).compile.drain.unsafeRunSync
+    }
+
+    //todo: need to resolve SoE in flatMap
+    "interrupt (6)" in {
+      // tests the interruption of the stream that recurse infinitely and never emits
+      val interrupt = mkScheduler.flatMap { _.sleep_[IO](20.millis) }.compile.drain.attempt
+      def loop: Stream[IO, Int] = Stream.eval(IO{()}).flatMap { _ => loop }
+      loop.interruptWhen(interrupt).compile.drain.unsafeRunSync
+    }
+
+    "interrupt (7)" in {
+      // tests the interruption of the stream that recurse infinitely, is pure and never emits
+      val interrupt = mkScheduler.flatMap { _.sleep_[IO](20.millis) }.compile.drain.attempt
+      def loop: Stream[IO, Int] = Stream.emit(()).covary[IO].flatMap { _ => loop }
+      loop.interruptWhen(interrupt).compile.drain.unsafeRunSync
+    }
+
+    "interrupt (8)" in {
+      // tests the interruption of the stream that repeatedly evaluates
+      val interrupt = mkScheduler.flatMap { _.sleep_[IO](20.millis) }.compile.drain.attempt
+      Stream.repeatEval(IO{()}).interruptWhen(interrupt).compile.drain.unsafeRunSync
+    }
+
+    "interrupt (9)" in {
+      // tests the interruption of the constant drained stream
+      val interrupt = mkScheduler.flatMap { _.sleep_[IO](1.millis) }.compile.drain.attempt
+      Stream.constant(true).dropWhile(! _ ).covary[IO].interruptWhen(interrupt).compile.drain.unsafeRunSync
+    }
+
+    "interrupt (10)" in forAll { (s1: PureStream[Int]) =>
       // tests that termination is successful even if interruption stream is infinitely false
       runLog { s1.get.covary[IO].interruptWhen(Stream.constant(false)) } shouldBe runLog(s1.get)
     }
 
-    "interrupt (3)" in forAll { (s1: PureStream[Int]) =>
+    "interrupt (11)" in forAll { (s1: PureStream[Int]) =>
       val barrier = async.mutable.Semaphore[IO](0).unsafeRunSync()
       val enableInterrupt = async.mutable.Semaphore[IO](0).unsafeRunSync()
       val interruptedS1 = s1.get.covary[IO].evalMap { i =>
@@ -176,6 +230,72 @@ class Pipe2Spec extends Fs2Spec {
       assert(out.forall(i => i % 7 != 0))
     }
 
+    "interrupt (12)" in forAll { (s1: PureStream[Int]) =>
+      // tests interruption of stream that never terminates in flatMap
+      val s = async.mutable.Semaphore[IO](0).unsafeRunSync()
+      val interrupt = mkScheduler.flatMap { _.sleep_[IO](50.millis) }.compile.drain.attempt
+      // tests that termination is successful even when flatMapped stream is hung
+      runLog { s1.get.covary[IO].interruptWhen(interrupt).flatMap(_ => Stream.eval_(s.decrement)) } shouldBe Vector()
+    }
+
+    "interrupt (13)" in forAll { (s1: PureStream[Int], f: Failure) =>
+      // tests that failure from the interrupt signal stream will be propagated to main stream
+      // even when flatMap stream is hung
+
+      val s = async.mutable.Semaphore[IO](0).unsafeRunSync()
+      val interrupt = mkScheduler.flatMap { _.sleep_[IO](50.millis) ++ f.get.map { _ => false } }
+      val prg = (Stream(1) ++ s1.get).covary[IO].interruptWhen(interrupt).flatMap(_ => Stream.eval_(s.decrement))
+      val throws = f.get.compile.drain.attempt.unsafeRunSync.isLeft
+      if (throws) an[Err.type] should be thrownBy runLog(prg)
+      else runLog(prg)
+    }
+
+    "interrupt (14)" in  forAll { s1: PureStream[Int] =>
+      // tests that when interrupted, the interruption will resume with append.
+      val s = async.mutable.Semaphore[IO](0).unsafeRunSync()
+      val interrupt = mkScheduler.flatMap { _.sleep_[IO](50.millis) }.compile.drain.attempt
+      val prg = (
+        (s1.get.covary[IO].interruptWhen(interrupt).evalMap { _ => s.decrement map { _ => None } })
+         ++ (s1.get.map(Some(_)))
+      ).collect { case Some(v) => v }
+
+      runLog(prg) shouldBe runLog(s1.get)
+    }
+
+    "interrupt (15)" in forAll { s1: PureStream[Int] =>
+      // tests that interruption works even when flatMap is followed by `collect`
+      // also tests scenario when interrupted stream is followed by other stream and both have map fusion defined
+      val s = async.mutable.Semaphore[IO](0).unsafeRunSync()
+      val interrupt = mkScheduler.flatMap { _.sleep_[IO](20.millis) }.compile.drain.attempt
+      val prg =
+        (s1.get.covary[IO].interruptWhen(interrupt).map { i => None } ++ s1.get.map(Some(_)))
+        .flatMap {
+          case None => Stream.eval(s.decrement.map { _ => None })
+          case Some(i) => Stream.emit(Some(i))
+        }
+        .collect { case Some(i) => i }
+
+      runLog(prg) shouldBe runLog(s1.get)
+
+    }
+
+    "nested-interrupt (1)" in forAll { s1: PureStream[Int] =>
+      val s = async.mutable.Semaphore[IO](0).unsafeRunSync()
+      val interrupt: IO[Either[Throwable, Unit]] = mkScheduler.flatMap { _.sleep_[IO](20.millis) }.compile.drain.attempt
+      val neverInterrupt = IO.async[Unit] { _ => () }.attempt
+
+      val prg =
+        (s1.get.covary[IO].interruptWhen(interrupt).map(_ => None) ++ s1.get.map(Some(_))).interruptWhen(neverInterrupt)
+        .flatMap {
+          case None => Stream.eval(s.decrement.map { _ => None })
+          case Some(i) => Stream.emit(Some(i))
+        }
+        .collect { case Some(i) => i }
+
+      runLog(prg) shouldBe runLog(s1.get)
+    }
+
+
     "pause" in {
       forAll { (s1: PureStream[Int]) =>
         val pausedStream = Stream.eval(async.signalOf[IO,Boolean](false)).flatMap { pause =>

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

@@ -31,12 +31,24 @@ class StreamPerformanceSpec extends Fs2Spec {
       }
     }}
 
+    "left-associated eval() ++ flatMap 1" - { Ns.foreach { N =>
+      N.toString in {
+        runLog((1 until N).map(emit).foldLeft(emit(0).covary[IO])((acc,a) => acc flatMap { _ => eval(IO {()}) flatMap { _ =>  a }})) shouldBe Vector(N-1)
+      }
+    }}
+
     "right-associated flatMap 1" - { Ns.foreach { N =>
       N.toString in {
         runLog((1 until N).map(emit).reverse.foldLeft(emit(0))((acc,a) => a flatMap { _ => acc })) shouldBe Vector(0)
       }
     }}
 
+    "right-associated eval() ++ flatMap 1" - { Ns.foreach { N =>
+      N.toString in {
+        runLog((1 until N).map(emit).reverse.foldLeft(emit(0).covary[IO])((acc,a) => a flatMap { _ => eval(IO {()}) flatMap { _=> acc } })) shouldBe Vector(0)
+      }
+    }}
+
     "left-associated flatMap 2" - { Ns.foreach { N =>
       N.toString in {
         runLog((1 until N).map(emit).foldLeft(emit(0) ++ emit(1) ++ emit(2))(

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

@@ -0,0 +1,18 @@
+package fs2
+
+import fs2.internal.Token
+
+/**
+  * Signals interruption of the evaluation.
+  *
+  * @param scopeId  Id of the scope that shall be the last interrupted scope by this signal
+  * @param loop     In case of infinite recursion this prevents interpreter to search for `CloseScope` indefinitely.
+  *                 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(private[fs2] val scopeId: Token, private[fs2] val loop: Int) extends Throwable {
+  override def fillInStackTrace = this
+
+  override def toString = s"Interrupted($scopeId, $loop)"
+}
+

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

@@ -26,4 +26,26 @@ abstract class Scope[F[_]] {
    * successfully leased.
    */
   def lease: F[Option[Lease[F]]]
+
+  /**
+    * Interrupts evaluation of the current scope. Only scopes previously indicated wih Stream.interruptScope may be interrupted.
+    * For other scopes this will fail.
+    *
+    * Interruption is final and may take two forms:
+    *
+    * When invoked on right side, that will interrupt only current scope evaluation, and will resume when control is given
+    * to next scope.
+    *
+    * When invoked on left side, then this will inject given throwable like it will be caused by stream evaluation,
+    * and then, without any error handling the whole stream will fail with supplied throwable.
+    *
+    */
+  def interrupt(cause: Either[Throwable, Unit]): F[Unit]
+
+  /**
+    * Yields to true, if the scope is interrupted.
+    * Note that when scope is interrupted with error, this yields to false
+    * @return
+    */
+  def isInterrupted: F[Boolean]
 }

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

@@ -1034,6 +1034,21 @@ object Segment {
       result.get
     }
 
+    /**
+      * Like `run` but allows to run `f` for each `O`.
+      * as they are processed when running this segment.
+      * Allows to perfrom efficient accumulation of `O` while running the stream.
+      */
+    final def runForEach(f: O => Unit): R = {
+      def chunk(ch: Chunk[O]): Unit = ch.foreach(f)
+      var result: Option[R] = None
+      val trampoline = new Trampoline
+      val step = self.stage(Depth(0), trampoline.defer, f, chunk, r => { result = Some(r); throw Done }).value
+      try while (true) stepAll(step, trampoline)
+      catch { case Done => }
+      result.get
+    }
+
     /**
      * Splits this segment at the specified index by simultaneously taking and dropping.
      *