adds on discard support for `FluxZip`/`MonoZip` #3209

OlegDokuka · 2022-10-03T10:10:06Z

This PR adds support for missing the onDiscard functionality to the FluxZip / MonoZip operators

Also, this PR includes fixes to fluky tests

Signed-off-by: Oleh Dokuka odokuka@vmware.com

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

reactor-core/src/main/java/reactor/core/publisher/FluxZip.java

chemicL · 2022-10-04T09:17:40Z

reactor-core/src/main/java/reactor/core/publisher/FluxZip.java

+
+		void drain(@Nullable ZipInner<T> callerInner, @Nullable Object dataSignal) {
 			if (WIP.getAndIncrement(this) != 0) {
+				if (callerInner != null) {


As onDiscard is a no-op when dataSignal == null, perhaps it's worth checking that it's non-null also in this condition to avoid the unnecessary checks that lead to the no-op?

This is already done inside onDiscard

That is what I am saying, but without the "already" part. My suggestion is to do the null check and fail early without executing the following 4 if statements + 1 method call, all leading to a no-op.

chemicL · 2022-10-04T09:30:18Z

reactor-core/src/main/java/reactor/core/publisher/FluxZip.java

+
+					if (queue != null) {
+						if (sourceMode == ASYNC) {
+							// delegates discarding to the queue holder to ensure there is no racing on draining from the SpScQueue


Are there actual implementations of this logic? How can the queue holder tell whether the clear came from a discard or another behaviour?

As of the fusion contract, the call for the queue.clear means - termination of the process. In the case of fusion - the queue comes from the upstream, so the upstream is a creator/owner of the queue, so it and only it can have full control of when values are offered and drained. See #2636 for more information

Thanks, that's what I was asking about. I checked a sample implementation and saw that the clear() gets overriden and handles the discarding.

chemicL · 2022-10-04T09:56:07Z

reactor-core/src/test/java/reactor/core/publisher/OnDiscardShouldNotLeakTest.java

+			DiscardScenario.allFluxSourceArray("zipScalar", 4,
+					sources -> {
+						Publisher<Tracked>[] sources1 =
+								Stream.concat(sources.stream(), Stream.of(Mono.just(Tracked.RELEASED))).toArray(Publisher[]::new);


Why is the scalar already RELEASED? Shouldn't it be an unreleased one, to be able to validate that it is properly consumed?

Scenarios are defined statically, so within a call, we can not get access to the allocator which holds all the references to Tracked objects. In this particular case, we initiate a specific path for FluxZip, which uses ZipSingleCoordinator instead of the default ZipCoordinator. Thus, it is not that important for now, although this can be improved in a separate PR so the rules API can get access to allocator upon the rule installation

I see. Now I understand it's not vital to validate the scalar value is released. Thanks.

chemicL · 2022-10-04T10:17:35Z

reactor-test/src/main/java/reactor/test/publisher/ColdTestPublisher.java


 	@Override
-	public ColdTestPublisher<T> next(@Nullable T t) {
+	public ColdTestPublisher<T> next(@Nullable T t, Consumer<T> onUndelivered) {


Is there no need to use the onUndelivered value like in DefaultTestPublisher?

I rolled back this API since it is a breaking one after all. In the case of ColdTestPublisher, all the values are cached in the internal store, so they can be consumed slightly latter

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

chemicL · 2022-10-06T12:19:44Z

reactor-core/src/main/java/reactor/core/publisher/MonoZip.java

-				for (ZipInner<R> ms : subscribers) {
-					ms.cancel();
-				}
+			if (DONE.getAndSet(this, Integer.MIN_VALUE) <=0) {


Suggested change

if (DONE.getAndSet(this, Integer.MIN_VALUE) <=0) {

if (DONE.getAndSet(this, Integer.MIN_VALUE) <= 0) {

chemicL · 2022-10-06T12:23:07Z

reactor-core/src/main/java/reactor/core/publisher/MonoZip.java


 		@Override
 		public void onError(Throwable t) {
+			if (value != null) {


When can this happen? Should this case not be reported somehow? Is it not a behaviour change?

generally, this should never happen as of Mono specification which guarantees only the following cases:

onNext + onComplete
onComplete
onError

The others are prohibited so onNext + onError was never expected although I added that as an extra check.

Will roll it back

chemicL · 2022-10-06T12:26:18Z

reactor-core/src/main/java/reactor/core/publisher/MonoZip.java

 			if (value == null) {
 				value = t;
-				parent.signal();
+				if (!parent.signal()) {


My understanding is that false is returned when this value does not finish the collection of the zipped items. Why then should the onDiscard be called?

false means the signal was not delivered which happens when the process was canceled or force terminated with the error or empty source. In that scenario value is discarded by a caller since the signal method has no information about the caller or signaled value

The way it is implemented right now seems different. Also, please have a look at failing tests from CI.
The condition for returning false is:

if (DONE.incrementAndGet(this) != n) { return false; }

And that is the same condition which signifies the zipping is done. I'm a bit confused.

chemicL · 2022-10-06T12:29:56Z

reactor-test/src/main/java/reactor/test/publisher/TestPublisher.java

-import java.util.Arrays;
 import java.util.EnumSet;
 import java.util.Objects;
+import java.util.function.Consumer;


As the API got reverted, the unused imports can be removed.

chemicL

I have one doubt that seems significant, about discarding items in MonoZip, but everything else looks good. Happy to approve once I understand the discarding in MonoZip.

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

simonbasle

a few comments on the tests, plus minor note on change from onNextDropped to onDiscard

reactor-core/src/main/java/reactor/core/publisher/FluxZip.java

reactor-test/src/main/java/reactor/test/publisher/ColdTestPublisher.java

reactor-core/src/test/java/reactor/core/publisher/MonoCacheInvalidateWhenTest.java

simonbasle · 2022-10-07T16:05:51Z

reactor-core/src/test/java/reactor/core/publisher/FluxFilterFuseableTest.java

 		                        .publishOn(Schedulers.newSingle("discardPollAsync"))
 		                        .filter(i -> { throw new IllegalStateException("boom"); })
 		                        .filter(i -> true)
+		                        .contextWrite(previous -> {


does this change really relates to this PR?

This fails on my machine all the time (m1 does goos job in detecting such things)

simonbasle · 2022-10-07T16:05:56Z

reactor-core/src/test/java/reactor/core/publisher/FluxFilterFuseableTest.java

+		CountDownLatch latch = new CountDownLatch(10);
 		StepVerifier.create(Flux.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10) //range uses tryOnNext, so let's use just instead
 		                        .publishOn(Schedulers.newSingle("discardPollAsync"), 1)
+				                .contextWrite(previous -> {


does this change really relates to this PR?

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

simonbasle

good from my perspective. @chemicL did you have any unresolved comment?

chemicL

LGTM. I had one comment about handling dataSignal == null case early in FluxZip::drain but now the logic is different. Anyways, it's not a blocker. Worth considering an optimisation to avoid a few branched executions just to hit a null check. That optimisation can be addressed in a more focused effort around #3227.

Signed-off-by: Oleh Dokuka <odokuka@vmware.com> Signed-off-by: OlegDokuka <odokuka@vmware.com>

reactorbot · 2022-10-24T10:42:24Z

@OlegDokuka this PR seems to have been merged on a maintenance branch, please ensure the change is merge-forwarded to intermediate maintenance branches and up to main 🙇

Signed-off-by: Oleh Dokuka <odokuka@vmware.com> Signed-off-by: OlegDokuka <odokuka@vmware.com>

This PR adds support for missing the `onDiscard` functionality to the `FluxZip` / `MonoZip` operators Also, this commit includes fixes to fluky tests related to the `onDiscard` functionality

adds on discard support for FluxZip

9c547e2

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

OlegDokuka added type/enhancement A general enhancement area/doOnDiscard This belongs to the doOnDiscard theme labels Oct 3, 2022

OlegDokuka added this to the 3.4.24 milestone Oct 3, 2022

OlegDokuka requested a review from a team as a code owner October 3, 2022 10:10

OlegDokuka changed the base branch from main to 3.4.x October 3, 2022 10:10

improves TestPublisher next to discard value if no active consumers

b628408

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

chemicL reviewed Oct 4, 2022

View reviewed changes

Oleh Dokuka added 3 commits October 4, 2022 16:02

adds MonoZip onDiscard support

c29794f

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

cleanups and spotless applied

5a759a4

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

rollbacks TestPublisher API changes

4c120fa

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

OlegDokuka requested a review from a team October 4, 2022 13:55

chemicL reviewed Oct 6, 2022

View reviewed changes

chemicL requested changes Oct 6, 2022

View reviewed changes

Oleh Dokuka added 6 commits October 7, 2022 01:23

improves Mono and Flux zip impl

76ccdbc

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

improves FluxPublishOn to ensure no leaks

3839c7e

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

improves fluky test

0648600

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

improves fluky test

db21d44

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

revert TestPublisher

7b25cb9

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

improves test

a9f5201

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

OlegDokuka changed the title ~~adds on discard support for FluxZip~~ adds on discard support for FluxZip/MonoZip. Improves FluxPublishOn onDiscard behavior Oct 7, 2022

reverts PublishOn

86c45a9

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

OlegDokuka changed the title ~~adds on discard support for FluxZip/MonoZip. Improves FluxPublishOn onDiscard behavior~~ adds on discard support for FluxZip/MonoZip Oct 7, 2022

simonbasle suggested changes Oct 7, 2022

View reviewed changes

Oleh Dokuka added 3 commits October 7, 2022 22:03

improves FluxZip

c37f47d

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

removes unnecessary changes to the file

8e23460

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

adjusts scan info

40f3cb6

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

OlegDokuka requested review from chemicL and simonbasle October 8, 2022 08:48

simonbasle modified the milestones: 3.4.24, 3.4.25 Oct 10, 2022

simonbasle approved these changes Oct 19, 2022

View reviewed changes

chemicL approved these changes Oct 20, 2022

View reviewed changes

improves onDiscard during wip

ffa9467

Signed-off-by: Oleh Dokuka <odokuka@vmware.com> Signed-off-by: OlegDokuka <odokuka@vmware.com>

OlegDokuka merged commit fe5a3b5 into 3.4.x Oct 24, 2022

OlegDokuka deleted the bugfix/3.4.x-zip-ondiscard branch October 24, 2022 10:42

OlegDokuka pushed a commit that referenced this pull request Oct 24, 2022

Merge #3209 into 3.5.0

e476286

Signed-off-by: Oleh Dokuka <odokuka@vmware.com> Signed-off-by: OlegDokuka <odokuka@vmware.com>

VictoryWangCN mentioned this pull request Mar 10, 2025

adds class check for discarded values rsocket/rsocket-java#1091

Merged

	if (DONE.getAndSet(this, Integer.MIN_VALUE) <=0) {
	if (DONE.getAndSet(this, Integer.MIN_VALUE) <= 0) {

adds on discard support for FluxZip/MonoZip #3209

adds on discard support for FluxZip/MonoZip #3209

Uh oh!

Conversation

OlegDokuka commented Oct 3, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

OlegDokuka Oct 5, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chemicL Oct 6, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chemicL left a comment

Choose a reason for hiding this comment

Uh oh!

simonbasle left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

simonbasle left a comment

Choose a reason for hiding this comment

Uh oh!

chemicL left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

reactorbot commented Oct 24, 2022

Uh oh!

adds on discard support for `FluxZip`/`MonoZip` #3209

adds on discard support for `FluxZip`/`MonoZip` #3209

OlegDokuka commented Oct 3, 2022 •

edited

Loading

OlegDokuka Oct 5, 2022 •

edited

Loading

chemicL Oct 6, 2022 •

edited

Loading

chemicL left a comment •

edited

Loading