Timed Pull #2062

SystemFw · 2020-10-02T22:32:45Z

Timed Pulls aim to cover a gap in our current api: aggregation with timeouts.

Examples:

The current implementation of groupWithin , which is very complex
groupWeighedWithin, which was asked in Add groupedWeightedWithin #2060
In the past I've needed groupAdjacentWithin for some Kafka code
Things like posting a metric if the stream is not "advancing" in time

Since the demise of AsyncPull, uncons cannot be timed out, and the alternative approach needed to implement this set of combinators is complex: you need queues, a mechanism of resettable timeout, a mechanism to avoid losing messages, etc.

Timed Pull brings a much simpler, Pull-like api to the problem, but without requiring any "kernel mode" operations.

This is still very WIP, but I'd like to get some feedback about the api before I go implement the various combinators and relative tests and docs.

mpilquist · 2020-10-03T12:37:24Z

core/shared/src/main/scala/fs2/concurrent/TimedPull.scala

+    def startTimer(t: FiniteDuration): Pull[F, INothing, Unit]
+  }
+  object TimedPull {
+    def go[F[_]: Temporal, A, B](pull: TimedPull[F, A] => Pull[F, B, Unit]): Pipe[F, A, B] = { source =>


Instead of this signature, what do you think of:

// On ToPull[F[_], O] def timed[A](pull: TimedPull[F, O] => Pull[F, B, Unit])(implicit t: Temporal[F]): Pull[F, B, Unit]

Usage like:

s.pull.timed { tp => ... }.stream

I was thinking of doing both, kinda like Broadcast does it.
I prefer yours for normal usage tbh, my only "issue" with it is that if you go to look at TimedPull, there's nothing there showing you how to use it (and you need to refer to ToPull)

The alternative is to put TimedPull inside Stream, which is how StepLeg does it, and have the signature you propose exclusively. I don't mind that.

Yeah I'm going with your suggestion for sure, since it allows reusing things like repeatPull

mpilquist

Looks pretty cool. One question - how much of the pull creation API would need to be duplicated on TimedPull -- e.g., how do you get a unconsN and unconsLimit that's time aware?

SystemFw · 2020-10-03T12:41:58Z

how do you get a unconsN and unconsLimit that's time aware?

That's a great question, one I do not have a great answer to yet. My instinct is to be light on duplication, but perhaps the two you have mentioned should be there.

My instinct is to be light on duplication,

Or rather, to be guided by the timed combinators we want to add, rather than starting out by duplicating big swathes of the Pull api

SystemFw · 2020-10-03T12:52:23Z

e.g., how do you get a unconsN and unconsLimit that's time aware?

Btw, if Pull was left abstract (more like an interface than a sealed trait), there is a scary/intriguing possibility of transforming existing Pulls into timed versions, but it would need to be able to reinterpret Pull.pure as well. Not quite sure whether that's a fantastic or terrible idea :)

mpilquist · 2020-10-03T13:12:39Z

We need explicit signaling of the timeout starts though, right? E.g. startTimeout then uncons 2 chunks versus timeout uncons timeout uncons? Or would the transformed pulls just add a timeout to each uncons?

SystemFw · 2020-10-03T13:37:18Z

Or would the transformed pulls just add a timeout to each uncons?

I mean, it's a pretty far out idea, but I was thinking each uncons would handle a potential a timeout and each "return" would start/reset one. You need to know which Pull you're reinterpreting though, cause for some "return" is pure, and for others is output. But overall I'd like to get this api (the manual one) in regardless, given how hard it is to implement the equivalent functionality right now, and then think longer about Pull introspection

Daenyth · 2020-10-03T13:40:37Z

core/shared/src/main/scala/fs2/concurrent/TimedPull.scala

+  trait TimedPull[F[_], A] {
+    type Timeout
+    def uncons: Pull[F, INothing, Option[(Either[Timeout, Chunk[A]], TimedPull[F, A])]]
+    def startTimer(t: FiniteDuration): Pull[F, INothing, Unit]
+  }


This all makes sense to me, it's a nice mirror of the Pull api 👍

Daenyth · 2020-10-03T13:44:06Z

core/shared/src/main/scala/fs2/concurrent/TimedPull.scala

+
+  // if we express timedUncons and startTimer as `F`, that guarantee is hard to achieve
+  // cause they could be concurrent. We could express both as Pulls, forcing a sync loop.
+  // need to add a note that the `unconsTimed` `Pull`, unlike `uncons`, is not idempotent


Can you elaborate at this? For me it's not obvious what the cause or implication of this is

sure, I need some time to formulate a response, it's a bit subtle (which is why I noted it down first thing).
As a start, try and get familiar with the stale timeout check in the old groupWithin

ah wait, do you mean the idempotency comment? That's not true anymore, it is idempotent now, since it's expressed in terms of uncons1. I thought you meant why using Pull in TimedPull rather than F[Unit] and F[Option[Either[Timeout, Chunk[A]]]], that's the subtle bit, but it's settled now

I was referring to the idempotency comment, yes

Right, so I knew I wanted to have Pull to forbid concurrency in the api (that's the thing I thought you were asking), and at first I was thinking about having an explicit queue and do uncons = Pull.eval(dequeue1), which would mean however that uncons >> uncons would get different elements, whereas the normal Pull.uncons is idempotent. I thought it didn't matter too much anyway since it's generally used linearly, but I was happier when I found out I could avoid the explicit queue altogether and rely on merge only, instead.

Daenyth · 2020-10-10T20:31:13Z

core/shared/src/test/scala/fs2/StreamCombinatorsSuite.scala

@@ -706,6 +706,9 @@ class StreamCombinatorsSuite extends Fs2Suite {
  }

  group("groupWithin") {
+    // TODO more tests:
+    // with no waiting, it should be equal to chunkN


core/shared/src/main/scala/fs2/concurrent/TimedPull.scala

SystemFw · 2020-10-20T15:43:07Z

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

+              .dropWhile(_.id == initial.id)
+              .switchMap { timeout =>
+                Stream.sleep(timeout.d).as(timeout.id)
+              }


I have a question on this @mpilquist @djspiewak . I think only data can give a definite answer but I just want to gauge what you think first. Look at the diff for the implementation of timeouts.
The problem is implementing resettable timeouts.
The first one only uses one extra fiber, but it only works if calls to resetTimer never try to reset a timeout which hasn't triggered yet with a shorter one.
The second is fully general, but it switches fiber at every timeout reset.

My question is, do you think the first (based on your sensibility) would offer a significant advantage in perf and/or memory utilisation? If so, do you think it's worth trying to introduce it as a "fast-path" for when resetTimer is called in the right conditions (and only fallback to switchMap if not), or is that too much trouble for little gain?

Interesting question. Pondering…

One thought here is that fibers are extremely cheap. The context shift around start doesn't even involve memory barriers by default, since it just plops the new fiber onto the same worker thread, so you can pretty safely spam that function. The allocation itself creates some memory pressure, but the resident size of IOFiber is not enormous. It's basically analogous to the magnitude of allocations you need to perform if you interpret .pull.uncons1 on a small (~16) chunk, but with fewer objects. Specifically, five objects get allocated in addition to the IOFiber when you call start, two of which are small arrays. The object header of IOFiber itself is the largest memory footprint in the whole menagerie.

Tldr, I suspect that vomiting out fibers is going to be fine. It's not something I would want to be doing in a hot path, but I also wouldn't want to be hitting Stream chunk boundaries within a hot path either (for the same reason), and it's around the same order of magnitude cost.

Cool. Just note that switchMap is a lot more than just creating a Fiber (queues, semaphores etc are involved), but I think the memory pressure concern would be the same, and sounds like it shouldn't be too much of an issue

mpilquist · 2020-10-22T12:05:32Z

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

+  * See below for detailed descriptions of `timeout` and `uncons`, and
+  * look at the [[ToPull.timed]] scaladoc for an example of usage.
+  */
+  trait TimedPull[F[_], O] {


Thoughts on defining this as Pull.Timed instead of Stream.TimedPull?

oh interesting. It is closer to ToPull in spirit, but I do like Pull.Timed and it's probably more discoverable, so if you're definitely 👍 on Pull.Timed, I'll move it there :)

SystemFw · 2020-10-23T01:58:45Z

Great, there is a failure in MemoryLeakSpec for groupWithin. Although I have a vague memory that you've seen it recently (but before this change) @mpilquist ?

SystemFw · 2020-10-23T02:07:14Z

oh yeah #2056 (comment) , I'll restart

SystemFw · 2020-10-23T07:53:09Z

I ran the groupWithin 2 memory leak test locally a couple of times and it passes with no problems (manual inspection with visualvm agrees)

SystemFw requested a review from mpilquist October 2, 2020 22:32

mpilquist reviewed Oct 3, 2020

View reviewed changes

Daenyth reviewed Oct 3, 2020

View reviewed changes

SystemFw added 9 commits October 9, 2020 12:52

Gather ideas for generalised groupWithin

cae950a

Add key implementation notes for TimedPull

c171a75

Sketch out TimedPull without explicit queues

d2dcf76

Add roughly working prototype of groupWithin based on TimedPull

a9b2632

Cleanup implementation of TimedPull

8cc3109

Make Timeout existential in TimedPull

87cedfd

Remove stale code

15ded9b

Add TODOs

6cfb7a4

Start the initial timer in groupWithin with TimedPull

9ac3843

SystemFw force-pushed the timed-pull branch from 846b6e7 to 9ac3843 Compare October 9, 2020 13:17

Add test for equivalence between groupWithin and chunkN

9936e76

Daenyth reviewed Oct 10, 2020

View reviewed changes

SystemFw added 4 commits October 11, 2020 01:02

Sketch strategy for testing stale timeouts

2c99bbd

Identity groupWithin corner case on first chunk

904fd93

Detail groupWithin bug

5cfa6cf

Add test case for chunk manipulation in groupWithin

e6de3b7

SystemFw commented Oct 11, 2020

View reviewed changes

core/shared/src/main/scala/fs2/concurrent/TimedPull.scala Outdated Show resolved Hide resolved

SystemFw added 5 commits October 12, 2020 00:39

Fix compile error in dotty

3ef1dca

Better logic to filter stale timeouts

7b20b07

Clean up logic for emitting non empty chunks only

b420b74

Prototype groupWithin semantics for timeout with empty accumulator

eb3449d

Remove stale code

365fc6d

SystemFw added 2 commits October 20, 2020 15:36

Specify the ability to shorten timeouts

f1c5cb3

Redesign timeouts in TimedPull to allow shortening of timeouts

f0b89dc

SystemFw commented Oct 20, 2020

View reviewed changes

SystemFw added 9 commits October 20, 2020 23:15

Eliminate custom Timeout class in TimedPull

7044f85

Move definition of TimedPull into timed method

0d5dd13

Remove standalone TimedPull

f44178d

Rename TimedPull.startTimer to timeout

c95e4ef

Refactor TimedPullSuite

ae13e33

Add ability to cancel a timeout without having to start another one

b21ad2a

Add TimedPull example

e7e93a8

Add scaladoc for pull.timed

4fa9e24

Add scaladoc for TimedPull

c00946a

zakolenko mentioned this pull request Oct 21, 2020

generalize Stream.groupWithin #2098

Closed

mpilquist reviewed Oct 22, 2020

View reviewed changes

SystemFw and others added 7 commits October 22, 2020 21:03

TimedPull --> Pull.Timed

70bcf9b

Fix scaladoc links

ab6bf2b

Complete scaladoc for Pull.Timed

2eab6b2

Revamp scaladoc for groupWithin

8ae0947

Remove playground

b3c8203

Formatting

92c3767

Merge branch 'develop' into timed-pull

5249926

SystemFw marked this pull request as ready for review October 23, 2020 01:27

Formatting

7802b8b

Fix exhaustivity warning

099a13e

Ignore the secondary groupWithin leak test

e59a02c

mpilquist merged commit e355955 into typelevel:develop Oct 23, 2020

diesalbla mentioned this pull request Mar 15, 2021

Faster groupWithin implementation #2327

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Timed Pull #2062

Timed Pull #2062

SystemFw commented Oct 2, 2020

mpilquist Oct 3, 2020 •

edited

Loading

SystemFw Oct 3, 2020

SystemFw Oct 4, 2020

SystemFw Oct 5, 2020

mpilquist left a comment

SystemFw commented Oct 3, 2020 •

edited

Loading

SystemFw commented Oct 3, 2020

mpilquist commented Oct 3, 2020

SystemFw commented Oct 3, 2020 •

edited

Loading

Daenyth Oct 3, 2020

Daenyth Oct 3, 2020

SystemFw Oct 3, 2020 •

edited

Loading

SystemFw Oct 3, 2020

Daenyth Oct 3, 2020

SystemFw Oct 3, 2020 •

edited

Loading

Daenyth Oct 10, 2020

SystemFw Oct 20, 2020 •

edited

Loading

djspiewak Oct 20, 2020

SystemFw Oct 20, 2020

mpilquist Oct 22, 2020

SystemFw Oct 22, 2020

SystemFw commented Oct 23, 2020

SystemFw commented Oct 23, 2020

SystemFw commented Oct 23, 2020

Timed Pull #2062

Timed Pull #2062

Conversation

SystemFw commented Oct 2, 2020

mpilquist Oct 3, 2020 • 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

mpilquist left a comment

Choose a reason for hiding this comment

SystemFw commented Oct 3, 2020 • edited Loading

SystemFw commented Oct 3, 2020

mpilquist commented Oct 3, 2020

SystemFw commented Oct 3, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SystemFw Oct 3, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SystemFw Oct 3, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SystemFw Oct 20, 2020 • 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

SystemFw commented Oct 23, 2020

SystemFw commented Oct 23, 2020

SystemFw commented Oct 23, 2020

mpilquist Oct 3, 2020 •

edited

Loading

SystemFw commented Oct 3, 2020 •

edited

Loading

SystemFw commented Oct 3, 2020 •

edited

Loading

SystemFw Oct 3, 2020 •

edited

Loading

SystemFw Oct 3, 2020 •

edited

Loading

SystemFw Oct 20, 2020 •

edited

Loading