implements new version of WindowTimeout #2822

OlegDokuka · 2021-10-21T15:28:08Z

Signed-off-by: Oleh Dokuka [email protected]

simonbasle

high level review

simonbasle · 2021-10-22T13:48:13Z

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

 	}

 	@Override
 	public Object scanUnsafe(Attr key) {
-		if (key == Attr.RUN_ON) return timer;
-		if (key == Attr.RUN_STYLE) return Attr.RunStyle.ASYNC;
+		if (key == Attr.RUN_ON) {


nit: please revert to single lines in that method

simonbasle · 2021-10-22T13:48:38Z

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

-	final Scheduler      timer;
-
-	FluxWindowTimeout(Flux<T> source, int maxSize, long timespan, TimeUnit unit, Scheduler timer) {
+	final int       maxSize;


nit: please add your name above in the @author list

simonbasle · 2021-10-22T13:52:59Z

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

@@ -41,7 +41,7 @@
 	public void windowWithTimeoutAccumulateOnSize() {
 		StepVerifier.withVirtualTime(() -> Flux.range(1, 6)
 						   .delayElements(Duration.ofMillis(300))
-						   .windowTimeout(5, Duration.ofMillis(2000))
+						   .windowTimeout(5, Duration.ofMillis(2000), true)


the test coverage of FluxWindowTimeoutTest wasn't great in the first place (65% of lines in 3.3.x, 62% in this PR), but I think we can improve it?

Either:

turn these tests that use fair == trueinto parameterized ones (⚠️ this means forward merging will need to turn them into @ParameterizedTestWithName in main)

duplicate the tests, eg. in a @Nested...

wait, this is targeting main 🤔 so yeah, direct use of ParameterizedTestWithName then

simonbasle · 2021-10-22T14:13:57Z

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

@@ -9788,6 +9788,11 @@ public final void subscribe(Subscriber<? super T> actual) {
 		return windowTimeout(maxSize, maxTime , Schedulers.parallel());
 	}

+	public final Flux<Flux<T>> windowTimeout(int maxSize, Duration maxTime,
+			boolean fairBackpressure) {
+		return windowTimeout(maxSize, maxTime , Schedulers.parallel(), true);


fairBackpressure parameter isn't used here, hardcoded to true

simonbasle · 2021-10-22T14:36:58Z

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

@@ -88,13 +856,13 @@ public Object scanUnsafe(Attr key) {
 		volatile boolean done;
 		volatile boolean cancelled;

-		volatile long requested;
+		volatile     long                                            requested;


nit: avoid applying reformatting to the whole file, only edited lines

simonbasle · 2021-10-22T14:37:36Z

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

@@ -165,7 +165,7 @@ public void dispose() {
 		};

 		StepVerifier.create(Flux.range(1, 3).hide()
-		                        .windowTimeout(10, Duration.ofMillis(500), testScheduler))
+		                        .windowTimeout(10, Duration.ofMillis(500), testScheduler, false))


I guess there is a reason this particular test only works with old mode, in which case the parameterized comment above wouldn't apply.

simonbasle · 2021-10-22T14:38:34Z

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

@@ -230,7 +230,7 @@ public void dispose() {

 	@Test
 	public void scanOperator() {
-		FluxWindowTimeout<Integer> test = new FluxWindowTimeout<>(Flux.just(1), 123, 100, TimeUnit.MILLISECONDS, Schedulers.immediate());
+		FluxWindowTimeout<Integer> test = new FluxWindowTimeout<>(Flux.just(1), 123, 100, TimeUnit.MILLISECONDS, Schedulers.immediate(), true);


scan tests of WindowTimeoutBackpressureSubscriber and InnerWindow would need to be added

simonbasle · 2021-10-22T14:40:17Z

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

+			}
+		}
+
+		static final long FINALIZED_STATE          =


let's at least put constants at the top of the class. static methods usually go there too (although here there are a lot of them, so I'm less sure)

simonbasle · 2021-10-22T14:53:36Z

To try and give a little detail on the implementation and tradeoffs:

requesting N to main flux is interpreted as "requesting N windows", so 0 demand would mean 0 window opening
this version with backpressure requests maxSize to the source whenever a window is subscribed to (inner)
if a previous window was closed due to timeout, it means it has requested maxSize M to parent already, but might only have received M1 elements. So we're already expecting M2 = M - M1 elements to come naturally
in that case new window will only request M1 from the source (since we have pending M2, M2 + M1 = M = maxSize 👍)

The other big tradeoff is that this variant will actually prolong the life of the active window if it "closes" due to timeout (ie. there is still pending demand to the source) AND there is currently 0 request for a new window. So it doesn't really pause the upstream when there is 0 demand.

There is a fundamental question here: what does it actually mean to say "there is 0 demand for a new window"? windows are by nature tied to the timing of pushes from the source.

Possible alternative interpretations that I can think of:

never request more that 1 from the source at a time (maybe making it possible to actually pause the upstream)
open a new window right away which enqueues elements, and see if it will be requested (through main) and drained (through inner): doesn't pause the upstream, but correctly cuts the old window at the moment of the timeout at least
???

joshfree · 2021-11-11T16:45:36Z

Tagging @JonathanGiles @anuchandy

anuchandy · 2022-01-22T00:57:39Z

Hi @simonbasle, I pulled the draft and tried using the new operator with the below code.

To give you a summary - The code has a producer, producing one event in every 250 ms once the backpressure request is received. It means it produces 4 events per second. The code has a consumer using the new operator to retrieve events from this producer with a max wait time of 1 second with a max size of 10.

I was expecting the consumer with the new operator to receive a window with four events every 1 second, but it is not.

We can see backpressure request going to the producer, and it is indeed producing events, but for some reason it never comes out of the new operator

@Test
public void windowTimeoutWithBackPressureFromCore() throws InterruptedException {
	// -- The Event Producer
	//    The producer emitting requested events to downstream but with a delay of 250ms between each emission.
	//
	final int eventProduceDelayInMillis = 250;
	Flux<String> eventProducer = Flux.create(sink -> {
		sink.onRequest(request -> {
			if (request != Long.MAX_VALUE) {
				System.out.println("Backpressure Request(" + request + ")");
				LongStream.range(0, request)
						.mapToObj(String::valueOf)
						.forEach(message -> {
							try {
								TimeUnit.MILLISECONDS.sleep(eventProduceDelayInMillis);
							} catch (InterruptedException e) {
								e.printStackTrace();
							}
							System.out.println("Producing:" + message);
							sink.next(message);
						});
			} else {
				sink.error(new RuntimeException("No_Backpressure unsupported"));
			}
		});
	});

	// -- The Event Consumer
	//    The consumer using windowTimeout that batches maximum 10 events with a max wait time of 1 second.
	//    Given the Event producer produces at most 4 events per second (due to 250 ms delay between events),
	//    the consumer should receive 3-4 events.
	//
	final int eventConsumeDelayInMillis = 0;
	final Scheduler scheduler = Schedulers.newBoundedElastic(10, 10000, "queued-tasks");
	final AtomicBoolean hasError = new AtomicBoolean(false);
	final Semaphore isCompleted = new Semaphore(1);
	isCompleted.acquire();

	Disposable subscription = eventProducer.windowTimeout(10, Duration.ofSeconds(1), true)
			.concatMap(Flux::collectList)
			.publishOn(scheduler)
			.subscribe(eventBatch -> {
				for (String event : eventBatch) {
					System.out.println("Consuming: " + event);
					try {
						TimeUnit.MILLISECONDS.sleep(eventConsumeDelayInMillis);
					} catch (InterruptedException e) {
						System.err.println("Could not sleep for delay. Error: " + e);
					}
				}
				System.out.println("Completed batch.");
			}, error -> {
				System.err.println("Error: " + error);
				hasError.set(true);
				isCompleted.release();
			}, () -> {
				System.out.println("Completed.");
				isCompleted.release();
			});

	System.out.println("Running test...");
	final Duration TIME_TO_PUBLISH_EVENTS = Duration.ofMinutes(1);
	try {
		assertFalse(isCompleted.tryAcquire(TIME_TO_PUBLISH_EVENTS.toMinutes(), TimeUnit.MINUTES),
				"Should have been false because it would not error.");

		assertFalse(hasError.get(), "Should not have received an error.");
	} finally {
		subscription.dispose();
	}

	System.out.println("Completed test.");
}

Output


Backpressure Request(10)
Producing:0
Producing:1
Producing:2
Backpressure Request(3)
Producing:3
Producing:0
Producing:4
Producing:1
Producing:5
Producing:2
Consuming: 0
Consuming: 1
Consuming: 2
Completed batch.
Producing:6
Backpressure Request(7)
Producing:7
Producing:0
Producing:8
Producing:1
Producing:9
Running test...
Producing:2
Producing:3
Producing:4
Producing:5
Producing:6
Backpressure Request(10)
Producing:0
Producing:1
Producing:2
Producing:3
Producing:4
Producing:5
Producing:6
Producing:7
Producing:8
Producing:9
Backpressure Request(10)
Producing:0
Producing:1
Producing:2
Producing:3
Producing:4
Producing:5
Producing:6
Producing:7
Producing:8
Producing:9
Backpressure Request(10)
Producing:0
Producing:1
Producing:2
Producing:3
Producing:4
Producing:5
Producing:6
Producing:7
Producing:8
Producing:9
Backpressure Request(10)
Producing:0
Producing:1
Producing:2
Producing:3
Producing:4
Producing:5
Producing:6
Producing:7
Producing:8
Producing:9
Backpressure Request(10)
Producing:0
Producing:1
Producing:2
Producing:3
Producing:4
Producing:5
Producing:6
Producing:7
Producing:8
Producing:9
Backpressure Request(10)
Producing:0
Producing:1
Producing:2

Signed-off-by: Oleh Dokuka <[email protected]>

OlegDokuka · 2022-01-31T23:05:30Z

@anuchandy updated implementation a bit. Also, added your tests which demonstrate that the new impl is working as expected (probably).

Also, just FYI - operators like publishOn have a prefetch buffer which may store some amount of collected elements. Also, I added subscribeOn to decouple the requester thread from the producer thread. Impl is still in progress and requires some polishing/stress tests ( but at least it may give more ideas about the new behaviors )

JonathanGiles · 2022-01-31T23:07:12Z

Thank you @OlegDokuka - we will get to testing this straight away.

anuchandy · 2022-02-01T06:23:34Z

Hi @OlegDokuka, thanks of the updated impl and cleaning up the test. I’ll continue testing the impl.

Just curious, the stress test you mentioned - is it a part of reactor-core public repo or is it kind of validations done internally?

OlegDokuka · 2022-02-01T10:35:48Z

@anuchandy all existing stress tests could be found there -> https://github.com/reactor/reactor-core/tree/main/reactor-core/src/jcstress/java/reactor/core/publisher

anuchandy · 2022-02-01T18:44:59Z

Thanks, @OlegDokuka. As part of validations, I'm trying to understand the various backpressure involved, IIUC, considering the following code, there are two categories of backpressure -

Flux<Event> source = ..
Flux<Flux<Event>> sourceWindowed = source.windowTimeout(500, Duration.ofSeconds(3), ..);

Backpressure to source for events.
Backpressure to sourceWindowed for windows.

If we chain an operator with prefetch buffer (e.g., size 50) to sourceWindowed - does that mean

It results in requesting 50 "windows" to sourceWindowed
which internally requests 50 * 500 "events" to the source
with the timeout constrain of 3 sec applied to each of the 50 windows

Is this understanding correct?

Is there a setting that caps the number of concurrent "windows" in the above case? (e.g., capped by the threads in the scheduler)

OlegDokuka · 2022-02-03T12:54:36Z

@anuchandy

Backpressure to source for events.

there is a backpressure equal to the max elements in the window. So

Backpressure to sourceWindowed for windows.

it depends on the downstream. For example using concatMap(fn, 0) with zero prefetch setup, the demand going to be equal to 1. Once downstream prefetched a value from sourceWindowed the upstream going to receive up to maxWindowSize (which is in example 500. you may wonder why up to 500 - because if the previous window did not fulfill completely, then the next window will not request 500 again, but rather the diff between what received and what is remaining to be sent from the source).

It results in requesting 50 "windows" to sourceWindowed

yes, if downstream prefetches 50, then it will be stored in the requested value.

which internally requests 50 * 500 "events" to the source

nope, only 500 can be requested at a time.

with the timeout constrain of 3 sec applied to each of the 50 windows

since only a single window can be opened at a time. There is not possible intersections between windows in that operator

Hope, that explains everything,
Oleh

anuchandy · 2022-02-03T18:39:10Z

Thank you, @OlegDokuka, for the answers. Using the diff for the demand makes sense; we don't want to overread from the main source; combining that with the prolonged window (if no demand for new window) collecting any events after the timeout seems to ensure no event loss.

I am continuing the test/profiling. Please tag me if you amend new commit(s), that are ready for validation.

OlegDokuka · 2022-02-03T18:42:39Z

@anuchandy working on more fixes... will ping you

anuchandy · 2022-02-25T18:32:52Z

Hi @OlegDokuka, Hope you're doing great; I just want to check; how is the implementation going?

JonathanGiles · 2022-04-07T09:27:59Z

Any status update here @simonbasle and @OlegDokuka ? Thanks!!

OlegDokuka · 2022-04-07T09:52:01Z

@JonathanGiles it is coming. Stay tuned.

JonathanGiles · 2022-04-07T09:53:17Z

Thanks for the update, much appreciated. We are very eager to see this improvement!

Signed-off-by: Oleh Dokuka <[email protected]>

anuchandy · 2022-04-22T22:13:20Z

thanks @OlegDokuka, just tag me when the next draft is ready, so that I can pull it to test via our use cases

anuchandy · 2022-06-10T19:55:53Z

Hi @OlegDokuka, I hope you're doing good. Just checking, how is the work progressing? P lease let me know once you have the next draft to test.

OlegDokuka · 2022-06-10T20:06:59Z

@anuchandy this PR is superseed by the #3054 which is about to be merged and going to be released next Tuesday. Stay tuned

OlegDokuka · 2022-06-13T12:09:40Z

close since is superseded by #3054

OlegDokuka requested a review from a team as a code owner October 21, 2021 15:28

simonbasle mentioned this pull request Oct 22, 2021

WindowTimeout backpressure #1099

Closed

simonbasle suggested changes Oct 22, 2021

View reviewed changes

anuchandy mentioned this pull request Jan 22, 2022

Track windowTimeout progress with reactor-team / investigate workaround for OverflowException Azure/azure-sdk-for-java#23950

Closed

OlegDokuka marked this pull request as draft January 25, 2022 14:07

simonbasle added this to the 3.4.x Backlog milestone Jan 25, 2022

Oleh Dokuka added 2 commits February 1, 2022 01:02

implements new version of WindowTimeout

27c5a5e

Signed-off-by: Oleh Dokuka <[email protected]>

adds more tests and fixes impl

2ea6bca

Signed-off-by: Oleh Dokuka <[email protected]>

OlegDokuka force-pushed the enhancement/window-timeout-new branch from c69ca37 to 2ea6bca Compare January 31, 2022 23:02

Fix license headers year

cebab6f

simonbasle mentioned this pull request Feb 14, 2022

windowTimeout operator causing flux to not process all events #2920

Open

Oleh Dokuka added 4 commits April 14, 2022 13:05

wip

028226c

Signed-off-by: Oleh Dokuka <[email protected]>

wip

13d11a4

Signed-off-by: Oleh Dokuka <[email protected]>

wip

05c4bf1

Signed-off-by: Oleh Dokuka <[email protected]>

wip

da76514

Signed-off-by: Oleh Dokuka <[email protected]>

OlegDokuka closed this Jun 13, 2022

chemicL deleted the enhancement/window-timeout-new branch April 11, 2024 12:45

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

implements new version of WindowTimeout #2822

implements new version of WindowTimeout #2822

OlegDokuka commented Oct 21, 2021

simonbasle left a comment

simonbasle Oct 22, 2021

simonbasle Oct 22, 2021

simonbasle Oct 22, 2021 •

edited

Loading

simonbasle Oct 22, 2021

simonbasle Oct 22, 2021

simonbasle Oct 22, 2021

simonbasle Oct 22, 2021

simonbasle Oct 22, 2021

simonbasle Oct 22, 2021

simonbasle commented Oct 22, 2021 •

edited

Loading

joshfree commented Nov 11, 2021

anuchandy commented Jan 22, 2022 •

edited

Loading

OlegDokuka commented Jan 31, 2022 •

edited

Loading

JonathanGiles commented Jan 31, 2022

anuchandy commented Feb 1, 2022

OlegDokuka commented Feb 1, 2022

anuchandy commented Feb 1, 2022

OlegDokuka commented Feb 3, 2022 •

edited

Loading

anuchandy commented Feb 3, 2022

OlegDokuka commented Feb 3, 2022

anuchandy commented Feb 25, 2022

JonathanGiles commented Apr 7, 2022

OlegDokuka commented Apr 7, 2022

JonathanGiles commented Apr 7, 2022

anuchandy commented Apr 22, 2022

anuchandy commented Jun 10, 2022

OlegDokuka commented Jun 10, 2022 •

edited

Loading

OlegDokuka commented Jun 13, 2022

implements new version of WindowTimeout #2822

implements new version of WindowTimeout #2822

Conversation

OlegDokuka commented Oct 21, 2021

simonbasle left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

simonbasle Oct 22, 2021 • 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

simonbasle commented Oct 22, 2021 • edited Loading

joshfree commented Nov 11, 2021

anuchandy commented Jan 22, 2022 • edited Loading

OlegDokuka commented Jan 31, 2022 • edited Loading

JonathanGiles commented Jan 31, 2022

anuchandy commented Feb 1, 2022

OlegDokuka commented Feb 1, 2022

anuchandy commented Feb 1, 2022

OlegDokuka commented Feb 3, 2022 • edited Loading

anuchandy commented Feb 3, 2022

OlegDokuka commented Feb 3, 2022

anuchandy commented Feb 25, 2022

JonathanGiles commented Apr 7, 2022

OlegDokuka commented Apr 7, 2022

JonathanGiles commented Apr 7, 2022

anuchandy commented Apr 22, 2022

anuchandy commented Jun 10, 2022

OlegDokuka commented Jun 10, 2022 • edited Loading

OlegDokuka commented Jun 13, 2022

simonbasle Oct 22, 2021 •

edited

Loading

simonbasle commented Oct 22, 2021 •

edited

Loading

anuchandy commented Jan 22, 2022 •

edited

Loading

OlegDokuka commented Jan 31, 2022 •

edited

Loading

OlegDokuka commented Feb 3, 2022 •

edited

Loading

OlegDokuka commented Jun 10, 2022 •

edited

Loading