Skip to content

[rpc] Fix RequestChannel backpressure race #4024

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
fxbing wants to merge 1 commit into
base: main
Choose a base branch
from
+212 −126
Open

[rpc] Fix RequestChannel backpressure race #4024

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
167 changes: 60 additions & 107 deletions fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestChannel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}

/**
Expand All @@ -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.
*
* <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();
}

/**
Expand All @@ -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) {
Expand Down Expand Up @@ -194,116 +189,74 @@ 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.
*
* <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 {
backpressureLock.unlock();
}
}

/**
* 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();
}
}
}
Loading
Loading