You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ya...@apache.org on 2026/08/20 01:48:43 UTC
(fluss) branch main updated: [rpc] Fix RequestChannel backpressure race
This is an automated email from the ASF dual-hosted git repository.
platinumhamburg pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 0bd725c05 [rpc] Fix RequestChannel backpressure race
0bd725c05 is described below
commit 0bd725c055fce2cf9139f6745bec2f9ca7c16d1f
Author: Xiaobing Fang <bi...@qq.com>
AuthorDate: Thu Aug 20 09:48:37 2026 +0800
[rpc] Fix RequestChannel backpressure race
- Reconcile queue thresholds and backpressure state atomically.
- Add deterministic race and zero-threshold regression coverage.
---
.../fluss/rpc/netty/server/RequestChannel.java | 167 ++++++++------------
.../fluss/rpc/protocol/RequestChannelTest.java | 171 ++++++++++++++++++---
2 files changed, 212 insertions(+), 126 deletions(-)
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestChannel.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestChannel.java
index f855c89e5..8ee85d500 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestChannel.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestChannel.java
@@ -93,8 +93,8 @@ public class RequestChannel {
public RequestChannel(int backpressureThreshold) {
this.requestQueue = new LinkedBlockingQueue<>();
- this.backpressureThreshold = backpressureThreshold;
- this.resumeThreshold = backpressureThreshold / 2;
+ this.backpressureThreshold = Math.max(1, backpressureThreshold);
+ this.resumeThreshold = this.backpressureThreshold / 2;
}
/**
@@ -105,17 +105,12 @@ public class RequestChannel {
* queue size exceeds the backpressure threshold, ALL channels associated with this
* RequestChannel will be paused to prevent further memory growth.
*
- * <p>OPTIMIZATION: Only check backpressure if not already active (avoid redundant checks).
+ * <p>The common path uses lock-free state and queue-size reads. The lock is acquired only when
+ * a backpressure transition may be required.
*/
public void putRequest(RpcRequest request) {
requestQueue.add(request);
-
- // CRITICAL OPTIMIZATION: Skip check if already in backpressure state.
- // This avoids lock contention on every putRequest() call when system is under pressure.
- // The volatile read is very cheap compared to lock acquisition.
- if (!isBackpressureActive) {
- pauseAllChannelsIfNeeded();
- }
+ reconcileBackpressure();
}
/**
@@ -137,8 +132,8 @@ public class RequestChannel {
public RpcRequest pollRequest(long timeoutMs) {
try {
RpcRequest request = requestQueue.poll(timeoutMs, TimeUnit.MILLISECONDS);
- if (isBackpressureActive) {
- tryResumeChannels();
+ if (request != null) {
+ reconcileBackpressure();
}
return request;
} catch (InterruptedException e) {
@@ -194,66 +189,30 @@ public class RequestChannel {
}
}
- /**
- * Check if the queue size has exceeded the backpressure threshold. When true, channel reads
- * should be paused to prevent memory exhaustion.
- */
- private boolean shouldApplyBackpressure() {
- return requestQueue.size() >= backpressureThreshold;
- }
-
- /**
- * Check if the queue size has dropped below the resume threshold. When true, paused channels
- * can be resumed to accept new requests.
- */
- private boolean shouldResumeChannels() {
- return requestQueue.size() <= resumeThreshold;
- }
-
- /**
- * Pauses ALL channels associated with this RequestChannel if the queue size exceeds the
- * backpressure threshold. This ensures that when the queue is full, all channels stop sending
- * requests to prevent memory exhaustion.
- *
- * <p>Uses a lock to protect the entire operation (state check + state change + task submission)
- * as an atomic unit. This prevents race conditions with resume operations and channel
- * registrations.
- *
- * <p>TODO: In the future, consider pausing only a subset of channels instead of all channels to
- * reduce the impact on upstream traffic. A selective pause strategy could minimize disruption
- * to the overall system while still providing effective backpressure control.
- */
- private void pauseAllChannelsIfNeeded() {
- if (!shouldApplyBackpressure()) {
+ /** Reconciles the backpressure state with the current queue size. */
+ private void reconcileBackpressure() {
+ int queueSize = requestQueue.size();
+ boolean backpressureActive = isBackpressureActive;
+ if (backpressureActive ? queueSize > resumeThreshold : queueSize < backpressureThreshold) {
return;
}
- // Lock protects: state check + state change + task submission as atomic operation
backpressureLock.lock();
try {
- // Check if already in backpressure state
- if (isBackpressureActive) {
- return; // Already paused, nothing to do
- }
-
- // Activate backpressure and pause all channels
- isBackpressureActive = true;
-
- for (Channel channel : associatedChannels) {
- if (channel.isActive()) {
- // Submit to the channel's EventLoop to ensure thread safety
- channel.eventLoop()
- .execute(
- () -> {
- if (channel.isActive() && channel.config().isAutoRead()) {
- channel.config().setAutoRead(false);
- LOG.warn(
- "Queue size ({}) exceeded backpressure threshold ({}), paused channel: {}",
- requestsCount(),
- backpressureThreshold,
- channel.remoteAddress());
- }
- });
+ while (true) {
+ queueSize = requestQueue.size();
+ if (isBackpressureActive) {
+ if (queueSize > resumeThreshold) {
+ return;
+ }
+ isBackpressureActive = false;
+ resumeAllChannels(queueSize);
+ } else {
+ if (queueSize < backpressureThreshold) {
+ return;
+ }
+ isBackpressureActive = true;
+ pauseAllChannels(queueSize);
}
}
} finally {
@@ -261,49 +220,43 @@ public class RequestChannel {
}
}
- /**
- * Attempts to resume all associated channels if the queue size has dropped below the resume
- * threshold. This method is called automatically after a request is dequeued.
- *
- * <p>Uses a lock to protect the entire operation (state check + state change + task submission)
- * as an atomic unit. This prevents race conditions with pause operations and channel
- * registrations.
- */
- private void tryResumeChannels() {
- if (!shouldResumeChannels()) {
- return;
- }
-
- // Lock protects: state check + state change + task submission as atomic operation
- backpressureLock.lock();
- try {
- // Check if backpressure is not active
- if (!isBackpressureActive) {
- return; // Already resumed, nothing to do
+ private void pauseAllChannels(int queueSize) {
+ for (Channel channel : associatedChannels) {
+ if (channel.isActive()) {
+ // Submit to the channel's EventLoop to ensure thread safety
+ channel.eventLoop()
+ .execute(
+ () -> {
+ if (channel.isActive() && channel.config().isAutoRead()) {
+ channel.config().setAutoRead(false);
+ LOG.warn(
+ "Queue size ({}) reached backpressure threshold ({}), paused channel: {}",
+ queueSize,
+ backpressureThreshold,
+ channel.remoteAddress());
+ }
+ });
}
+ }
+ }
- // Deactivate backpressure and resume all channels
- isBackpressureActive = false;
-
- for (Channel channel : associatedChannels) {
- if (channel.isActive()) {
- // Submit resume task to the channel's EventLoop to ensure thread safety
- channel.eventLoop()
- .execute(
- () -> {
- if (channel.isActive() && !channel.config().isAutoRead()) {
- channel.config().setAutoRead(true);
- LOG.info(
- "Queue size ({}) dropped below resume threshold ({}), resumed channel: {}",
- requestsCount(),
- resumeThreshold,
- channel.remoteAddress());
- }
- });
- }
+ private void resumeAllChannels(int queueSize) {
+ for (Channel channel : associatedChannels) {
+ if (channel.isActive()) {
+ // Submit resume task to the channel's EventLoop to ensure thread safety
+ channel.eventLoop()
+ .execute(
+ () -> {
+ if (channel.isActive() && !channel.config().isAutoRead()) {
+ channel.config().setAutoRead(true);
+ LOG.info(
+ "Queue size ({}) reached resume threshold ({}), resumed channel: {}",
+ queueSize,
+ resumeThreshold,
+ channel.remoteAddress());
+ }
+ });
}
- } finally {
- backpressureLock.unlock();
}
}
}
diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/RequestChannelTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/RequestChannelTest.java
index d71b01665..b433d0082 100644
--- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/RequestChannelTest.java
+++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/RequestChannelTest.java
@@ -52,6 +52,8 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -148,10 +150,6 @@ public class RequestChannelTest {
channel.putRequest(createTestRequest(i));
}
- // Wait for backpressure to be applied (channel operations are async via eventLoop)
- // The backpressure should be triggered when we add the threshold-th request
- testChannel.waitForAutoReadChange(false, 2, TimeUnit.SECONDS);
-
// Verify backpressure is active: autoRead should be false
assertThat(testChannel.isAutoRead()).isFalse();
assertThat(channel.requestsCount()).isGreaterThanOrEqualTo(backpressureThreshold);
@@ -165,9 +163,6 @@ public class RequestChannelTest {
assertThat(request).isNotNull();
}
- // Wait for backpressure to be released
- testChannel.waitForAutoReadChange(true, 2, TimeUnit.SECONDS);
-
// Verify backpressure is released: autoRead should be true again
assertThat(testChannel.isAutoRead()).isTrue();
assertThat(channel.requestsCount()).isLessThanOrEqualTo(resumeThreshold);
@@ -176,6 +171,129 @@ public class RequestChannelTest {
channel.unregisterChannel(testChannel);
}
+ @Test
+ void testBackpressureStaysReleasedWhenQueueDrainsBeforePause() throws Exception {
+ RequestChannel channel = new RequestChannel(2);
+ TestChannel testChannel = new TestChannel();
+ channel.registerChannel(testChannel);
+ channel.putRequest(createTestRequest(0));
+ channel.putRequest(createTestRequest(1));
+ assertThat(testChannel.isAutoRead()).isFalse();
+
+ testChannel.eventLoop.blockNextTask();
+ FutureTask<Void> resumeTask =
+ new FutureTask<>(
+ () -> {
+ channel.pollRequest(0);
+ return null;
+ });
+ Thread resumeThread = new Thread(resumeTask);
+ FutureTask<Void> pauseTask =
+ new FutureTask<>(
+ () -> {
+ channel.putRequest(createTestRequest(2));
+ return null;
+ });
+ Thread pauseThread = new Thread(pauseTask);
+
+ try {
+ resumeThread.start();
+ testChannel.eventLoop.waitUntilTaskBlocked();
+ pauseThread.start();
+ waitForRequestCount(channel, 2);
+ waitUntilBlocked(pauseThread);
+ assertThat(channel.pollRequest(0)).isNotNull();
+ assertThat(channel.pollRequest(0)).isNotNull();
+ } finally {
+ testChannel.eventLoop.releaseBlockedTask();
+ }
+
+ resumeTask.get(2, TimeUnit.SECONDS);
+ pauseTask.get(2, TimeUnit.SECONDS);
+ assertThat(channel.requestsCount()).isZero();
+ assertThat(testChannel.isAutoRead()).isTrue();
+ }
+
+ @Test
+ void testBackpressureStaysActiveWhenQueueRefillsBeforeResume() throws Exception {
+ RequestChannel channel = new RequestChannel(2);
+ TestChannel testChannel = new TestChannel();
+ channel.registerChannel(testChannel);
+ channel.putRequest(createTestRequest(0));
+
+ testChannel.eventLoop.blockNextTask();
+ FutureTask<Void> pauseTask =
+ new FutureTask<>(
+ () -> {
+ channel.putRequest(createTestRequest(1));
+ return null;
+ });
+ Thread pauseThread = new Thread(pauseTask);
+ FutureTask<Void> resumeTask =
+ new FutureTask<>(
+ () -> {
+ channel.pollRequest(0);
+ return null;
+ });
+ Thread resumeThread = new Thread(resumeTask);
+
+ try {
+ pauseThread.start();
+ testChannel.eventLoop.waitUntilTaskBlocked();
+ resumeThread.start();
+ waitForRequestCount(channel, 1);
+ waitUntilBlocked(resumeThread);
+ channel.putRequest(createTestRequest(2));
+ } finally {
+ testChannel.eventLoop.releaseBlockedTask();
+ }
+
+ pauseTask.get(2, TimeUnit.SECONDS);
+ resumeTask.get(2, TimeUnit.SECONDS);
+ assertThat(channel.requestsCount()).isEqualTo(2);
+ assertThat(testChannel.isAutoRead()).isFalse();
+ }
+
+ @Test
+ void testZeroBackpressureThresholdDoesNotSpin() throws Exception {
+ RequestChannel channel = new RequestChannel(0);
+ channel.putRequest(createTestRequest(0));
+ FutureTask<RpcRequest> pollTask = new FutureTask<>(() -> channel.pollRequest(0));
+ Thread pollThread = new Thread(pollTask);
+ pollThread.setDaemon(true);
+
+ try {
+ pollThread.start();
+ pollThread.join(TimeUnit.SECONDS.toMillis(2));
+ assertThat(pollTask.isDone()).isTrue();
+ } finally {
+ if (!pollTask.isDone()) {
+ channel.putRequest(createTestRequest(1));
+ pollThread.join(TimeUnit.SECONDS.toMillis(2));
+ }
+ }
+
+ assertThat(pollTask.get(2, TimeUnit.SECONDS)).isNotNull();
+ }
+
+ private static void waitForRequestCount(RequestChannel channel, int expectedCount) {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2);
+ while (channel.requestsCount() != expectedCount && System.nanoTime() < deadline) {
+ Thread.yield();
+ }
+ assertThat(channel.requestsCount()).isEqualTo(expectedCount);
+ }
+
+ private static void waitUntilBlocked(Thread thread) {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2);
+ while (thread.getState() != Thread.State.WAITING
+ && thread.getState() != Thread.State.BLOCKED
+ && System.nanoTime() < deadline) {
+ Thread.yield();
+ }
+ assertThat(thread.getState()).isIn(Thread.State.WAITING, Thread.State.BLOCKED);
+ }
+
/** Helper method to create a test RpcRequest with a unique identifier. */
private RpcRequest createTestRequest(int id) {
return new FlussRequest(
@@ -231,17 +349,6 @@ public class RequestChannelTest {
return autoRead.get();
}
- void waitForAutoReadChange(boolean expectedValue, long timeout, TimeUnit unit)
- throws InterruptedException {
- long deadline = System.nanoTime() + unit.toNanos(timeout);
- while (autoRead.get() != expectedValue && System.nanoTime() < deadline) {
- Thread.sleep(10);
- }
- assertThat(autoRead.get())
- .as("AutoRead should be " + expectedValue + " within timeout")
- .isEqualTo(expectedValue);
- }
-
// Minimal implementation of other required methods
@Override
public ChannelPipeline pipeline() {
@@ -578,10 +685,26 @@ public class RequestChannelTest {
/** Test EventLoop that executes tasks immediately in the current thread for testing. */
private static class TestEventLoop extends SingleThreadEventExecutor implements EventLoop {
+ private final AtomicBoolean blockNextExecution = new AtomicBoolean();
+ private final CountDownLatch taskBlocked = new CountDownLatch(1);
+ private final CountDownLatch blockedTaskRelease = new CountDownLatch(1);
+
TestEventLoop() {
super(null, new DefaultThreadFactory("test"), false);
}
+ void blockNextTask() {
+ blockNextExecution.set(true);
+ }
+
+ void waitUntilTaskBlocked() throws InterruptedException {
+ assertThat(taskBlocked.await(2, TimeUnit.SECONDS)).isTrue();
+ }
+
+ void releaseBlockedTask() {
+ blockedTaskRelease.countDown();
+ }
+
@Override
protected void run() {
// No-op, tasks are executed synchronously
@@ -589,7 +712,17 @@ public class RequestChannelTest {
@Override
public void execute(Runnable task) {
- // Execute immediately in current thread for testing
+ if (blockNextExecution.compareAndSet(true, false)) {
+ taskBlocked.countDown();
+ try {
+ if (!blockedTaskRelease.await(2, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("Timed out waiting to release task");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ }
task.run();
}