-
Notifications
You must be signed in to change notification settings - Fork 602
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Fixing interruption behaviour #3183
Changes from 10 commits
b720a5b
b03e1f8
4955b51
b785e49
8182c8e
fdc2d16
d0f0a3a
ff688be
c73ab66
b64eb5a
8d57590
9ae18d7
89889fa
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -23,7 +23,7 @@ package fs2 | |
|
||
import cats.effect.kernel.Deferred | ||
import cats.effect.kernel.Ref | ||
import cats.effect.std.{Semaphore, Queue} | ||
import cats.effect.std.{Queue, Semaphore} | ||
import cats.effect.testkit.TestControl | ||
import cats.effect.{IO, SyncIO} | ||
import cats.syntax.all._ | ||
|
@@ -34,6 +34,7 @@ import org.scalacheck.Prop.forAll | |
|
||
import scala.concurrent.duration._ | ||
import scala.concurrent.TimeoutException | ||
import scala.util.control.NoStackTrace | ||
|
||
class StreamCombinatorsSuite extends Fs2Suite { | ||
override def munitIOTimeout = 1.minute | ||
|
@@ -834,6 +835,72 @@ class StreamCombinatorsSuite extends Fs2Suite { | |
) | ||
.assertEquals(0.millis) | ||
} | ||
|
||
test("upstream failures are propagated downstream") { | ||
TestControl.executeEmbed { | ||
case object SevenNotAllowed extends NoStackTrace | ||
|
||
val source = Stream | ||
.iterate(0)(_ + 1) | ||
.covary[IO] | ||
.evalTap(n => IO.raiseError(SevenNotAllowed).whenA(n == 7)) | ||
|
||
val downstream = source.groupWithin(100, 2.seconds) | ||
|
||
downstream.intercept[SevenNotAllowed.type] | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Should we make an assertion here about what the downstream has / has not received before the error? |
||
} | ||
} | ||
|
||
test( | ||
"upstream interruption causes immediate downstream termination with all elements being emitted" | ||
) { | ||
|
||
val sourceTimeout = 5.5.seconds | ||
val downstreamTimeout = sourceTimeout + 2.seconds | ||
|
||
TestControl | ||
.executeEmbed { | ||
val source: Stream[IO, Int] = | ||
Stream | ||
.iterate(0)(_ + 1) | ||
.covary[IO] | ||
.meteredStartImmediately(1.second) | ||
.interruptAfter(sourceTimeout) | ||
|
||
// large chunkSize and timeout (no emissions expected in the window | ||
// specified, unless source ends, due to interruption or | ||
// natural termination (i.e runs out of elements) | ||
val downstream: Stream[IO, Chunk[Int]] = | ||
source.groupWithin(Int.MaxValue, 1.day) | ||
|
||
downstream.compile.lastOrError | ||
.timeout(downstreamTimeout) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why is this timeout necessary? Since its an There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. it's because if the test fails we get a slightly better error message
otherwise we get this value on the diff which looks a bit random
but I'm happy to remove it There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Got it, that is a nicer error :) thanks! |
||
.map(_.toList) | ||
.timed | ||
} | ||
.assertEquals( | ||
// downstream ended immediately (i.e timeLapsed = sourceTimeout) | ||
// emitting whatever was accumulated at the time of interruption | ||
(sourceTimeout, List(0, 1, 2, 3, 4, 5)) | ||
) | ||
} | ||
|
||
test("stress test: all elements are processed") { | ||
val rangeLength = 10000 | ||
|
||
Stream | ||
.eval(Ref.of[IO, Int](0)) | ||
.flatMap { counter => | ||
Stream | ||
.range(0, rangeLength) | ||
.covary[IO] | ||
.groupWithin(4096, 100.micros) | ||
.evalTap(ch => counter.update(_ + ch.size)) *> Stream.eval(counter.get) | ||
} | ||
.compile | ||
.lastOrError | ||
.assertEquals(rangeLength) | ||
} | ||
} | ||
|
||
property("head")(forAll((s: Stream[Pure, Int]) => assertEquals(s.head.toList, s.toList.take(1)))) | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Sorry, dumb question: why is
Int.MaxValue
a "magic number" in this context? I would have thought it's effectively maxing out the semaphore, but if it needs+ outputLong
to work then I feel like it must have more significance?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Legit question to be fair. Had to think about it again.
Interruption of the upstream fiber (i.e. Outcome.Cancelled) is handled downstream by doing nothing (permits are never released)
So by increasing the supply to
Int.MaxValue
we are just evening out the negative balance (Int.MaxValue
is to account for the worst case scenario: at most thechunkSize
parameter will be equal toInt.MaxValue
)Now after getting past the "checkpoint" above we are acquiring
outputLong
permits againSo in order to get past this point we need to release an additional
outputLong
permits and that allows the stream to be unblockedEDIT
uhm well actually I've just tested it, it is not handled with
Outcome.Cancelled
...There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks for that explanation!
So could we just use
chunkSize
here, instead ofInt.MaxValue
?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@armanbilge apologies I was wrong, that's not what's happening here. I'm just doing some tests to figure out why we need the additional
outputLong
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Btw, if these implementations details are no longer relevant after your rewrite in the other PR, then let's not get too hung up on this one :)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ok I think I've figured it out (might be useful for the other implementation actually)
basically the problem is that we need enough supply to cover 2 iterations of the race loop. So if we only increase it by
Int.MaxValue
the following will happenif instead we increase it by
Int.MaxValue + outputLong
outputLong
So since the chunkSize can be as high as
Int.MaxValue
then the minimum supply to unblock the semaphore should beInt.MaxValue + outputLong
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Key word being "can". Wouldn't
chunkSize + outputLong
be sufficient?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
yeah that should work. The test still passes, I'll change it to
outputLong * 2
since chunkSize == outputLong