BoundedSourceQueue API #29770

seglo · 2020-10-21T21:42:40Z

Based on @jrudolph's FastDroppingQueue implementation in #29574

A new queue implementation that drops new elements immediately when the buffer is full. Does not use async callbacks like Source.queue with OverflowStrategy.dropNew, which can still result in OOM errors (#25798).

Other "drop" scenarios (dropTail, dropHead) may be supported by this implementation if their use can be justified.

akka-ci · 2020-10-21T21:58:57Z

Test FAILed.

akka-ci · 2020-10-21T21:59:11Z

Test FAILed.

akka-stream/src/main/scala/akka/stream/scaladsl/Source.scala

akka-ci · 2020-10-22T22:23:41Z

Test FAILed.

akka-ci · 2020-10-22T22:43:40Z

Test FAILed.

akka-ci · 2020-10-23T02:42:26Z

Test FAILed.

raboof

Looking good overall!

akka-docs/src/main/paradox/stream/operators/Source/queue.md

akka-stream-tests/src/test/scala/akka/stream/scaladsl/QueueSourceSpec.scala

akka-docs/src/main/paradox/stream/operators/Source/queue.md

akka-docs/src/test/java/jdocs/stream/IntegrationDocTest.java

akka-docs/src/test/scala/docs/stream/IntegrationDocSpec.scala

akka-docs/src/main/paradox/stream/operators/Source/queue.md

patriknw · 2020-10-26T13:06:27Z

akka-stream/src/main/scala/akka/stream/javadsl/Source.scala

-    new Source(
-      scaladsl.Source.queue[T](bufferSize, overflowStrategy, maxConcurrentOffers).mapMaterializedValue(_.asJava))
+    overflowStrategy match {
+      case _: OverflowStrategies.DropNew => queue[T](bufferSize).mapMaterializedValue(_.asJavaSourceQueue())


Are we sure that we want to change the queue implementation for DropNew? It's probably much better, but the change in behavior could impact users. I think we should consider deprecating this version of Source.queue instead and promote the new method instead. The deprecation could come in a separate PR (possibly later release) since we probably want to provide a good alternative replacement for at least Backpressure strategy.

I'll revert this change.

In the future I suggest we deprecate these two overloads altogether (with a backpressure and "fail" strategy alternative). We can't deprecate OverflowStrategies.DropNew because it's used in Source.actorRef as well.

and we might not need a fail strategy because that can be a decision for the caller (and implemented together with a KillSwitch)

akka-ci · 2020-11-02T15:06:42Z

Test PASSed.

johanandren

Looking good, some small things noted.

akka-docs/src/main/paradox/stream/operators/Source/queue.md

akka-docs/src/test/scala/docs/stream/IntegrationDocSpec.scala

akka-stream-tests/src/test/scala/akka/stream/scaladsl/BoundedQueueSourceSpec.scala

akka-stream/src/main/scala/akka/stream/BoundedQueueSource.scala

akka-stream/src/main/scala/akka/stream/QueueOfferResult.scala

akka-stream/src/main/scala/akka/stream/impl/BoundedQueueSource.scala

jrudolph

Thanks for doing this work, @seglo! I mostly looked at the documentation. I guess my biggest non-cosmetic suggestion is to rename BoundedQueueSource to BoundedSourceQueue...

akka-docs/src/main/paradox/stream/operators/index.md

jrudolph · 2020-11-05T13:09:44Z

akka-docs/src/main/paradox/stream/operators/Source/queue.md

@@ -1,15 +1,35 @@
 # Source.queue

-Materialize a `SourceQueue` onto which elements can be pushed for emitting from the source. 
+Materialize a `SourceQueue` or `BoundedSourceQueue` onto which elements can be pushed synchronously or asynchronously for emitting from the source.


Here's a suggestion of giving an overview about the options right away (I found the distinction between synchronous and asynchronous not really descriptive as a user coming to this page):

Suggested change

Materialize a `SourceQueue` or `BoundedSourceQueue` onto which elements can be pushed synchronously or asynchronously for emitting from the source.

Materialize a `BoundedSourceQueue` or `SourceQueue` onto which elements can be pushed for emitting from the source.

The `BoundedSourceQueue` is an optimized variant of the `SourceQueue` with `OverflowStrategy.dropNew`. The `BoundedSourceQueue` will give immediate, synchronous feedback whether an element was accepted or not and is therefore recommended for situations where overload and dropping elements is expected and needs to be handled quickly.

In contrast, the `SourceQueue` offers more variety of `OverflowStrategies` but feedback is only asynchronously provided through a @scala[`Future`]@java[`CompletionStage`] value. In cases where elements need to be discarded quickly at times of overload to avoid out-of-memory situations, delivering feedback asynchronously can itself become a problem. This happens if elements come in faster than the feedback can be delivered in which case the feedback mechanism itself is part of the reason that an out-of-memory situation arises.

In summary, prefer `BoundedSourceQueue` over `SourceQueue` with `OverflowStrategy.dropNew` especially in high-load scenarios. Use `SourceQueue` if you need one of the other `OverflowStrategies`.

I initially had more content above the fold like you suggested, but a docs post-processing step didn't like it. I'll take another look to see if I can make that work.

It looks like a rule was explicitly created to keep the description one line (to keep it short).

https://github.com/akka/akka/blob/master/project/StreamOperatorsIndexGenerator.scala#L244-L248

I like your description though, so I'll try to make it work with the rule intact.

akka-docs/src/main/paradox/stream/operators/Source/queue.md

jrudolph · 2020-11-05T13:26:54Z

akka-stream/src/main/scala/akka/stream/BoundedQueueSource.scala

+/**
+ * A queue of the given size that gives immediate feedback whether an element could be enqueued or not.
+ */
+trait BoundedQueueSource[T] {


Why not BoundedSourceQueue? After all, it's not a source but a queue and that would also be consistent with the existing naming? (I just realized that I confused those also above in my docs suggestions).

Btw. I think it would be fine if a BoundedQueueSource(Stage) would provide a BoundedSourceQueue. After all these are two opposite ends of looking at the component.

I renamed BoundedQueueSource to BoundedSourceQueue. I changed the stage too just to make it consistent.

akka-stream/src/main/scala/akka/stream/javadsl/Source.scala

jrudolph · 2020-11-05T13:40:19Z

LMK if this is acceptable @patriknw, @jrudolph.

Of course, thanks a lot for bringing that over the finish line.

seglo · 2020-11-05T17:57:25Z

I guess my biggest non-cosmetic suggestion is to rename BoundedQueueSource to BoundedSourceQueue

I originally used that convention, but I changed it to what it is now for some reason. I'll make it BoundedSourceQueue.

👍

seglo · 2020-11-05T18:56:47Z

@johanandren @jrudolph Thanks for the thorough review. I accommodated almost all of your feedback. I'll be away next week, but feel free to push commits if necessary to have this completed this sprint.

akka-ci · 2020-11-05T19:04:19Z

Test FAILed.

akka-ci · 2020-11-05T19:57:17Z

Test PASSed.

johanandren · 2020-11-06T07:59:10Z

Given that you are away next week, and @jrudolph is away today, let's merge this and follow up with fixes if J has any more feedback when back next week.

seglo · 2020-11-06T13:27:35Z

Sounds good. Thanks @johanandren 👍

seglo force-pushed the seglo/faster-drop-new-queue branch from 4cac2d6 to 0cfda82 Compare October 21, 2020 21:48

akka-ci added the validating PR is currently being validated by Jenkins label Oct 21, 2020

seglo force-pushed the seglo/faster-drop-new-queue branch from 0cfda82 to e746bc5 Compare October 21, 2020 21:50

akka-ci added needs-attention Indicates a PR validation failure (set by CI infrastructure) and removed validating PR is currently being validated by Jenkins labels Oct 21, 2020

patriknw reviewed Oct 22, 2020

View reviewed changes

akka-stream/src/main/scala/akka/stream/scaladsl/Source.scala Outdated Show resolved Hide resolved

seglo changed the title ~~DroppingQueueSource API~~ BoundedQueueSource API Oct 22, 2020

seglo force-pushed the seglo/faster-drop-new-queue branch from 3a48a4a to 66da0be Compare October 22, 2020 22:24

seglo force-pushed the seglo/faster-drop-new-queue branch from 66da0be to 44aead2 Compare October 23, 2020 01:50

raboof reviewed Oct 23, 2020

View reviewed changes

akka-docs/src/main/paradox/stream/operators/Source/queue.md Outdated Show resolved Hide resolved

akka-stream-tests/src/test/scala/akka/stream/scaladsl/QueueSourceSpec.scala Outdated Show resolved Hide resolved

patriknw reviewed Oct 26, 2020

View reviewed changes

ennru mentioned this pull request Oct 27, 2020

2020-10-26 Akka team plan akka/akka-meta#144

Closed

6 tasks

akka-ci added the validating PR is currently being validated by Jenkins label Oct 28, 2020

akka-ci removed the validating PR is currently being validated by Jenkins label Nov 2, 2020

johanandren reviewed Nov 5, 2020

View reviewed changes

jrudolph reviewed Nov 5, 2020

View reviewed changes

seglo force-pushed the seglo/faster-drop-new-queue branch from 52d28f1 to e2d1484 Compare November 5, 2020 18:51

seglo changed the title ~~BoundedQueueSource API~~ BoundedSourceQueue API Nov 5, 2020

akka-ci added validating PR is currently being validated by Jenkins and removed tested PR that was successfully built and tested by Jenkins labels Nov 5, 2020

akka-ci added needs-attention Indicates a PR validation failure (set by CI infrastructure) and removed validating PR is currently being validated by Jenkins labels Nov 5, 2020

BoundedQueueSource API

779161d

seglo force-pushed the seglo/faster-drop-new-queue branch from e2d1484 to 779161d Compare November 5, 2020 19:06

akka-ci added validating PR is currently being validated by Jenkins tested PR that was successfully built and tested by Jenkins and removed needs-attention Indicates a PR validation failure (set by CI infrastructure) validating PR is currently being validated by Jenkins labels Nov 5, 2020

johanandren approved these changes Nov 6, 2020

View reviewed changes

johanandren merged commit 2b8d0b2 into akka:master Nov 6, 2020

johanandren added this to the 2.6.11 milestone Nov 6, 2020

This was referenced Nov 9, 2020

A fast dropping queue implementation #29574

Closed

Reimplement Source.queue to remove reliance on async callbacks #29558

Closed

patriknw mentioned this pull request Nov 9, 2020

Replace old Source.queue #29801

Open

fabiotudone-da mentioned this pull request Nov 23, 2021

Make the InstrumentedSource.queue use the BoundedSourceQueue [KVL-1177] digital-asset/daml#11807

Merged

7 tasks

BoundedSourceQueue API #29770

BoundedSourceQueue API #29770

seglo commented Oct 21, 2020

akka-ci commented Oct 21, 2020

akka-ci commented Oct 21, 2020

akka-ci commented Oct 22, 2020

akka-ci commented Oct 22, 2020

akka-ci commented Oct 23, 2020

raboof left a comment

patriknw Oct 26, 2020

seglo Oct 28, 2020

patriknw Oct 28, 2020

akka-ci commented Nov 2, 2020

johanandren left a comment

jrudolph left a comment

jrudolph Nov 5, 2020

seglo Nov 5, 2020

seglo Nov 5, 2020

jrudolph Nov 5, 2020

jrudolph Nov 5, 2020

seglo Nov 5, 2020

jrudolph commented Nov 5, 2020

seglo commented Nov 5, 2020

seglo commented Nov 5, 2020

akka-ci commented Nov 5, 2020

akka-ci commented Nov 5, 2020

johanandren commented Nov 6, 2020

seglo commented Nov 6, 2020

-Materialize a `SourceQueue` or `BoundedSourceQueue` onto which elements can be pushed synchronously or asynchronously for emitting from the source.
+Materialize a `BoundedSourceQueue` or `SourceQueue` onto which elements can be pushed for emitting from the source.
+The `BoundedSourceQueue` is an optimized variant of the `SourceQueue` with `OverflowStrategy.dropNew`. The `BoundedSourceQueue` will give immediate, synchronous feedback whether an element was accepted or not and is therefore recommended for situations where overload and dropping elements is expected and needs to be handled quickly.
+In contrast, the `SourceQueue` offers more variety of `OverflowStrategies` but feedback is only asynchronously provided through a @scala[`Future`]@java[`CompletionStage`] value. In cases where elements need to be discarded quickly at times of overload to avoid out-of-memory situations, delivering feedback asynchronously can itself become a problem. This happens if elements come in faster than the feedback can be delivered in which case the feedback mechanism itself is part of the reason that an out-of-memory situation arises.
+In summary, prefer `BoundedSourceQueue` over `SourceQueue` with `OverflowStrategy.dropNew` especially in high-load scenarios. Use `SourceQueue` if you need one of the other `OverflowStrategies`.

BoundedSourceQueue API #29770

BoundedSourceQueue API #29770

Conversation

seglo commented Oct 21, 2020

akka-ci commented Oct 21, 2020

akka-ci commented Oct 21, 2020

akka-ci commented Oct 22, 2020

akka-ci commented Oct 22, 2020

akka-ci commented Oct 23, 2020

raboof left a 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

akka-ci commented Nov 2, 2020

johanandren left a comment

Choose a reason for hiding this comment

jrudolph left a 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

jrudolph commented Nov 5, 2020

seglo commented Nov 5, 2020

seglo commented Nov 5, 2020

akka-ci commented Nov 5, 2020

akka-ci commented Nov 5, 2020

johanandren commented Nov 6, 2020

seglo commented Nov 6, 2020