Reimplemented Channel in terms of Queue
#2856
Closed
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
SystemFw
reviewed
Mar 26, 2022
SystemFw
reviewed
Mar 26, 2022
|
Working on this PR uncovered a bug in Cats Effect 3.4.0-RC1 (fixed in typelevel/cats-effect#3180), so this PR is based on a snapshot. If it's accepted, we'll do an RC2. |
armanbilge
reviewed
Sep 30, 2022
Comment on lines
-283
to
-288
| // allocate once | ||
| @inline private final def closed[A]: Either[Closed, A] = _closed | ||
| private[this] final val _closed: Either[Closed, Nothing] = Left(Closed) | ||
| private final val rightUnit: Either[Closed, Unit] = Right(()) | ||
| private final val rightTrue: Either[Closed, Boolean] = Right(true) | ||
| private final val rightFalse: Either[Closed, Boolean] = Right(false) |
There was a problem hiding this comment.
Can bring them back. I think I chopped them out a while ago.
Comment on lines
146
to
149
| closedR.complete(()).map { | ||
| case false => Left(Channel.Closed) | ||
| case true => Right(()) | ||
| } |
There was a problem hiding this comment.
Also was going to suggest you use the cached left, but I see they're gone 😅
Suggested change
| closedR.complete(()).map { | |
| case false => Left(Channel.Closed) | |
| case true => Right(()) | |
| } | |
| closedR.complete(()).ifF(Either.unit, Left(Channel.Closed)) |
Comment on lines
138
to
142
| def sendAll: Pipe[F, A, Nothing] = | ||
| _.evalMapChunk(send(_)) | ||
| .takeWhile(_.isRight) | ||
| .onComplete(Stream.exec(close.void)) | ||
| .drain |
Comment on lines
202
to
203
| // you can do this more efficiently, just proves a point | ||
| Stream.eval(takeN).repeat.takeWhile(!_.isEmpty).flatMap(Stream.chunk(_)) |
There was a problem hiding this comment.
Suggested change
| // you can do this more efficiently, just proves a point | |
| Stream.eval(takeN).repeat.takeWhile(!_.isEmpty).flatMap(Stream.chunk(_)) | |
| // you can do this more efficiently, just proves a point | |
| Stream.eval(takeN).repeat.takeWhile(!_.isEmpty).unchunks |
There was a problem hiding this comment.
It was one of my first PRs to FS2 :)
There was a problem hiding this comment.
I've honestly wanted that convenience function for so long. I spend my whole life doing it.
|
|
Wow. How many elements were you trying to stick into that queue?? |
Co-authored-by: Arman Bilge <armanbilge@gmail.com>
|
Probably this? fs2/core/shared/src/test/scala/fs2/concurrent/TopicSuite.scala Lines 46 to 48 in ab256c3 fs2/core/shared/src/main/scala/fs2/concurrent/Topic.scala Lines 160 to 162 in 9ac3e43 |
|
So it seems like anyone who was using |
Yes. They shouldn't do that. :-P I mean, we can special case large bounds, but genuinely this just isn't a case which should be supported. It only worked before by coincidence. |
This reverts commit 3ff9ee2.
|
This is being caused by typelevel/cats-effect#3187 |
|
Getting better! Not quite better across the board yet, but doing really well. Because of how things are optimized, I'm pretty sure that the new implementation gets better the faster the producers run relative to the consumer, though the numbers don't seem to intuitively bear that out. Still fiddling with things, and the upstream BeforeAfter |
|
…and a little better: |
SystemFw
reviewed
Oct 13, 2022
|
@SystemFw I think this is ready for another look. Your Depends on a hash snapshot for the time being, hence why this is marked as draft. |
|
Updated with benchmark results. The simple cases got faster with the consolidated
|
|
Fiddled around some more with the performance cases. I took a flier at porting one of the most dramatic performance deltas from the Cats Effect
I suspect this is still a worthwhile change for a few reasons (e.g. the GC tends to do a lot better with contiguous array structures rather than linked-list equivalents, and this is not well measured in microbenchmarks), and I have some ideas for how to improve it a bit further, but I think we can probably foreclose seeing the kinds of huge performance leaps in |
mm, I don't want to be extremely gatekeeper-y, but this is the type of worst case scenario I feared: changing something that works, complicating the implementation (double refs, complex sentinel logic, Btw, what do you mean by linked list here? If it the fact that the current impl is based on List? That was benchmarked against a couple of alternatives, but it doesn't seem fundamental one way or another |
|
That's the benchmark results of List vs Vector: #2751 (comment) In |
|
I did my final set of experiments! Tldr, I'm closing this PR. So my last set of experiments looked at GC pressure (using The results were the exact opposite. The implementation on this PR imposes about double the heap pressure, GC count, GC time, and survivor count. I genuinely have no intuition for why, but at this point I think we can safely conclude that the performance of this implementation is somewhere between "worse" and "much worse" than the performance of the old version. Full output attached. As an aside, I have separately verified that the performance improvements in CE's All in all, I think we can conclude a few things:
I'll open a new PR bringing over the tests from this PR. :-) |
djspiewak
added a commit
to djspiewak/fs2
that referenced
this pull request
Nov 11, 2022
mpilquist
added a commit
that referenced
this pull request
Nov 12, 2022
Cherry-picked additional tests from #2856
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Do not merge. Depends on a snapshot of Cats Effect 3.4 (well, technically could be implemented without the snapshot, but it's better to wait for. The snapshot in question includes the (still unmerged) improvements totryTakeN)Queueon anyAsync[F](typelevel/cats-effect#2885 and typelevel/cats-effect#2914), but critically does not include an implementation of typelevel/cats-effect#2890, which is unquestionably the main bottleneck here:Before:
After:
To me, the fact that this is even close to the hand-rolled version is pretty cool, and a slightly less naive
streamwould probably put it almost on-par even without the optimizedtryTakeN. Once we get the latter though this should be head and shoulders above the prior implementation.Closes #2852