You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by rc...@apache.org on 2022/09/19 02:58:30 UTC
[james-project] 05/07: JAMES-3816 IMAP throttler should not await nested tasks
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit da71b92cf4e2452c42596c695adcd1c99773b9fe
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Mon Sep 12 19:32:38 2022 +0700
JAMES-3816 IMAP throttler should not await nested tasks
---
.../james/imapserver/netty/ReactiveThrottler.java | 12 ++--
.../imapserver/netty/ReactiveThrottlerTest.java | 69 +++++++++++++++++-----
2 files changed, 60 insertions(+), 21 deletions(-)
diff --git a/server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/ReactiveThrottler.java b/server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/ReactiveThrottler.java
index 0d9028d556..2a53e4a58b 100644
--- a/server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/ReactiveThrottler.java
+++ b/server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/ReactiveThrottler.java
@@ -28,6 +28,7 @@ import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
+import reactor.core.scheduler.Schedulers;
public class ReactiveThrottler {
public static class RejectedException extends RuntimeException {
@@ -58,7 +59,7 @@ public class ReactiveThrottler {
if (requestNumber <= maxConcurrentRequests) {
// We have capacity for one more concurrent request
return Mono.from(task)
- .then(Mono.defer(this::onRequestDone));
+ .doFinally(any -> onRequestDone());
} else if (requestNumber <= maxQueueSize + maxConcurrentRequests) {
// Queue the request for later
Sinks.One<Void> one = Sinks.one();
@@ -77,13 +78,14 @@ public class ReactiveThrottler {
}
}
- private Mono<Void> onRequestDone() {
+ private void onRequestDone() {
concurrentRequests.decrementAndGet();
Publisher<Void> throttled = queue.poll();
if (throttled != null) {
- return Mono.from(throttled)
- .then(Mono.defer(this::onRequestDone));
+ Mono.from(throttled)
+ .doFinally(any -> onRequestDone())
+ .subscribeOn(Schedulers.parallel())
+ .subscribe();
}
- return Mono.empty();
}
}
diff --git a/server/protocols/protocols-imap4/src/test/java/org/apache/james/imapserver/netty/ReactiveThrottlerTest.java b/server/protocols/protocols-imap4/src/test/java/org/apache/james/imapserver/netty/ReactiveThrottlerTest.java
index e8d626c2dc..3b767f61cc 100644
--- a/server/protocols/protocols-imap4/src/test/java/org/apache/james/imapserver/netty/ReactiveThrottlerTest.java
+++ b/server/protocols/protocols-imap4/src/test/java/org/apache/james/imapserver/netty/ReactiveThrottlerTest.java
@@ -1,3 +1,22 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you under the Apache License, Version 2.0 (the *
+ * "License"); you may not use this file except in compliance *
+ * with the License. You may obtain a copy of the License at *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
+ * Unless required by applicable law or agreed to in writing, *
+ * software distributed under the License is distributed on an *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
+ * KIND, either express or implied. See the License for the *
+ * specific language governing permissions and limitations *
+ * under the License. *
+ ****************************************************************/
+
package org.apache.james.imapserver.netty;
import static org.assertj.core.api.Assertions.assertThat;
@@ -36,9 +55,9 @@ class ReactiveThrottlerTest {
// When I submit many tasks task
AtomicBoolean executed = new AtomicBoolean(false);
- Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(200)).then())).subscribeOn(Schedulers.elastic()).subscribe();
- Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(200)).then())).subscribeOn(Schedulers.elastic()).subscribe();
- Mono.from(testee.throttle(Mono.fromRunnable(() -> executed.getAndSet(true)))).subscribeOn(Schedulers.elastic()).subscribe();
+ Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(200)).then())).subscribeOn(Schedulers.boundedElastic()).subscribe();
+ Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(200)).then())).subscribeOn(Schedulers.boundedElastic()).subscribe();
+ Mono.from(testee.throttle(Mono.fromRunnable(() -> executed.getAndSet(true)))).subscribeOn(Schedulers.boundedElastic()).subscribe();
// Then that task is not executed straight away
assertThat(executed.get()).isFalse();
@@ -51,9 +70,9 @@ class ReactiveThrottlerTest {
// When I submit many tasks task
AtomicBoolean executed = new AtomicBoolean(false);
- Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(50)).then())).subscribeOn(Schedulers.elastic()).subscribe();
- Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(50)).then())).subscribeOn(Schedulers.elastic()).subscribe();
- Mono.from(testee.throttle(Mono.fromRunnable(() -> executed.getAndSet(true)))).subscribeOn(Schedulers.elastic()).subscribe();
+ Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(50)).then())).subscribeOn(Schedulers.boundedElastic()).subscribe();
+ Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(50)).then())).subscribeOn(Schedulers.boundedElastic()).subscribe();
+ Mono.from(testee.throttle(Mono.fromRunnable(() -> executed.getAndSet(true)))).subscribeOn(Schedulers.boundedElastic()).subscribe();
// Then that task is eventually executed
Awaitility.await().atMost(Duration.ofSeconds(10))
@@ -67,8 +86,8 @@ class ReactiveThrottlerTest {
// When I await a submitted task execution and it is queued
AtomicBoolean executed = new AtomicBoolean(false);
- Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(50)).then())).subscribeOn(Schedulers.elastic()).subscribe();
- Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(50)).then())).subscribeOn(Schedulers.elastic()).subscribe();
+ Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(50)).then())).subscribeOn(Schedulers.boundedElastic()).subscribe();
+ Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(50)).then())).subscribeOn(Schedulers.boundedElastic()).subscribe();
Mono.from(testee.throttle(Mono.fromRunnable(() -> executed.getAndSet(true)))).block();
// Then when done that task have been executed
@@ -82,10 +101,10 @@ class ReactiveThrottlerTest {
// When I submit too many tasks task
AtomicBoolean executed = new AtomicBoolean(false);
- Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(50)).then())).subscribeOn(Schedulers.elastic()).subscribe();
- Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(50)).then())).subscribeOn(Schedulers.elastic()).subscribe();
- Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(50)).then())).subscribeOn(Schedulers.elastic()).subscribe();
- Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(50)).then())).subscribeOn(Schedulers.elastic()).subscribe();
+ Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(50)).then())).subscribeOn(Schedulers.boundedElastic()).subscribe();
+ Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(50)).then())).subscribeOn(Schedulers.boundedElastic()).subscribe();
+ Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(50)).then())).subscribeOn(Schedulers.boundedElastic()).subscribe();
+ Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(50)).then())).subscribeOn(Schedulers.boundedElastic()).subscribe();
// Then extra tasks are rejected
assertThatThrownBy(() -> Mono.from(testee.throttle(Mono.fromRunnable(() -> executed.getAndSet(true)))).block())
@@ -94,6 +113,24 @@ class ReactiveThrottlerTest {
assertThat(executed.get()).isFalse();
}
+ @Test
+ void throttleShouldNotAwaitOtherTasks() throws Exception {
+ // Given a throttler
+ ReactiveThrottler testee = new ReactiveThrottler(new NoopGaugeRegistry(), 2, 2);
+
+ // When I submit a short and a long task
+ AtomicBoolean executed = new AtomicBoolean(false);
+ Mono.from(testee.throttle(Mono.delay(Duration.ofMillis(100)).then()))
+ .then(Mono.fromRunnable(() -> executed.getAndSet(true)))
+ .subscribeOn(Schedulers.boundedElastic())
+ .subscribe();
+ Mono.from(testee.throttle(Mono.delay(Duration.ofSeconds(2)).then())).subscribeOn(Schedulers.boundedElastic()).subscribe();
+
+ // Then extra tasks are rejected
+ Thread.sleep(200);
+ assertThat(executed.get()).isTrue();
+ }
+
@Test
void throttleShouldNotExceedItsConcurrency() {
// Given a throttler
@@ -111,10 +148,10 @@ class ReactiveThrottlerTest {
int i = concurrentTasks.getAndDecrement();
concurrentTasksCountSnapshots.add(i);
}));
- Mono.from(testee.throttle(operation)).subscribeOn(Schedulers.elastic()).subscribe();
- Mono.from(testee.throttle(operation)).subscribeOn(Schedulers.elastic()).subscribe();
- Mono.from(testee.throttle(operation)).subscribeOn(Schedulers.elastic()).subscribe();
- Mono.from(testee.throttle(operation)).subscribeOn(Schedulers.elastic()).subscribe();
+ Mono.from(testee.throttle(operation)).subscribeOn(Schedulers.boundedElastic()).subscribe();
+ Mono.from(testee.throttle(operation)).subscribeOn(Schedulers.boundedElastic()).subscribe();
+ Mono.from(testee.throttle(operation)).subscribeOn(Schedulers.boundedElastic()).subscribe();
+ Mono.from(testee.throttle(operation)).subscribeOn(Schedulers.boundedElastic()).subscribe();
// Then maximum parallelism is not exceeded
Awaitility.await().untilAsserted(() -> assertThat(concurrentTasksCountSnapshots.size()).isEqualTo(8));
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org