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 f855c89e51d..8ee85d500f2 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 RequestChannel(int backpressureThreshold) { * queue size exceeds the backpressure threshold, ALL channels associated with this * RequestChannel will be paused to prevent further memory growth. * - *

OPTIMIZATION: Only check backpressure if not already active (avoid redundant checks). + *

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 void putShutdownRequest() { 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 void unregisterChannel(Channel channel) { } } - /** - * 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. - * - *

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. - * - *

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 @@ private void pauseAllChannelsIfNeeded() { } } - /** - * 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. - * - *

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 d71b01665b8..b433d008249 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.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 @@ void testBackpressureActivationAndDeactivation() throws Exception { 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 @@ void testBackpressureActivationAndDeactivation() throws Exception { 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 @@ void testBackpressureActivationAndDeactivation() throws Exception { 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 resumeTask = + new FutureTask<>( + () -> { + channel.pollRequest(0); + return null; + }); + Thread resumeThread = new Thread(resumeTask); + FutureTask 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 pauseTask = + new FutureTask<>( + () -> { + channel.putRequest(createTestRequest(1)); + return null; + }); + Thread pauseThread = new Thread(pauseTask); + FutureTask 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 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 @@ boolean isAutoRead() { 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 ChannelConfig setMessageSizeEstimator(MessageSizeEstimator estimator) { /** 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 @@ protected void run() { @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(); }