Skip to content

xds: Fix TSAN data race on ClientCall cancellation in ExternalProcessorClientInterceptor #13022

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
kannanjgithub wants to merge 3 commits into
base: master
Choose a base branch
from
Open

xds: Fix TSAN data race on ClientCall cancellation in ExternalProcessorClientInterceptor #13022

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
aa34ab1
xds: Fix TSAN data race on ClientCall cancellation in ExternalProcess…
kannanjgithub Aug 31, 2026
dbf3c4e
Merge remote-tracking branch 'upstream/master' into tsan-data-error-2
kannanjgithub Sep 1, 2026
900b166
Merge error fix
kannanjgithub Sep 1, 2026
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
74 changes: 41 additions & 33 deletions xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,8 +236,18 @@ public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
MethodDescriptor<InputStream, InputStream> rawMethod =
(MethodDescriptor<InputStream, InputStream>) (MethodDescriptor<?, ?>) method;
ClientCall<InputStream, InputStream> rawCall =
(ClientCall<InputStream, InputStream>) (ClientCall<?, ?>)
next.newCall(method, callOptions);
new SimpleForwardingClientCall<InputStream, InputStream>(
(ClientCall<InputStream, InputStream>) (ClientCall<?, ?>)
next.newCall(method, callOptions)) {
private final AtomicBoolean cancelled = new AtomicBoolean(false);

@Override
public void cancel(@Nullable String message, @Nullable Throwable cause) {
if (cancelled.compareAndSet(false, true)) {
super.cancel(message, cause);
}
}
};

// Create a local subclass instance to buffer outbound actions
DataPlaneDelayedCall<InputStream, InputStream> delayedCall =
Expand Down Expand Up @@ -362,7 +372,6 @@ private static class DataPlaneClientCall
final AtomicBoolean isProcessingTrailers = new AtomicBoolean(false);
final AtomicBoolean pendingHalfClose = new AtomicBoolean(false);
final AtomicBoolean bodyMessageSentToExtProc = new AtomicBoolean(false);
private final AtomicBoolean downstreamCancelled = new AtomicBoolean(false);

protected DataPlaneClientCall(
DataPlaneDelayedCall<InputStream, InputStream> delayedCall,
Expand Down Expand Up @@ -443,13 +452,14 @@ private boolean validateCompressionSupport(BodyResponse bodyResponse) {
.withDescription("gRPC message compression not supported in ext_proc")
.asRuntimeException();
synchronized (streamLock) {
if (!extProcStreamState.get().isCompleted()
&& extProcClientCallRequestObserver != null) {
extProcClientCallRequestObserver.onError(ex);
if (markExtProcStreamFailed(extProcStreamState)) {
if (extProcClientCallRequestObserver != null) {
extProcClientCallRequestObserver.onError(ex);
extProcClientCallRequestObserver = null;
}
}
}
activateCall();
markExtProcStreamFailed(extProcStreamState);
cancelDownstream("gRPC message compression not supported in ext_proc", ex);
closeExtProcStream();
return false;
Expand All @@ -463,7 +473,7 @@ public void start(Listener<InputStream> responseListener, Metadata headers) {
this.callContext = Context.current();
clientHeadersStartNanos = System.nanoTime();
this.requestHeaders = headers;
this.wrappedListener = new DataPlaneListener(responseListener, rawCall, this);
this.wrappedListener = new DataPlaneListener(responseListener, this);

// DelayedClientCall.start will buffer the listener and headers until setCall is called.
super.start(wrappedListener, headers);
Expand Down Expand Up @@ -664,6 +674,9 @@ public void onError(Throwable t) {
@Override
public void onCompleted() {
if (markExtProcStreamCompleted(extProcStreamState)) {
synchronized (streamLock) {
extProcClientCallRequestObserver = null;
}
handleFailOpen(wrappedListener);
}
}
Expand Down Expand Up @@ -693,7 +706,7 @@ public void onCompleted() {

private void sendToExtProc(ProcessingRequest request) {
synchronized (streamLock) {
if (extProcStreamState.get().isCompleted()) {
if (extProcStreamState.get().isCompleted() || extProcClientCallRequestObserver == null) {
return;
}

Expand Down Expand Up @@ -830,6 +843,7 @@ private void closeExtProcStream() {
if (markExtProcStreamCompleted(extProcStreamState)) {
if (extProcClientCallRequestObserver != null) {
extProcClientCallRequestObserver.onCompleted();
extProcClientCallRequestObserver = null;
}
}
}
Expand All @@ -839,11 +853,7 @@ private void internalOnError(Throwable t) {
if (markExtProcStreamFailed(extProcStreamState)) {
synchronized (streamLock) {
if (extProcClientCallRequestObserver != null) {
try {
extProcClientCallRequestObserver.onError(t);
} catch (Throwable ignored) {
// Ignore exceptions during cancel/onError propagation
}
extProcClientCallRequestObserver.onError(t);
extProcClientCallRequestObserver = null;
}
}
Expand Down Expand Up @@ -999,7 +1009,7 @@ public void sendMessage(InputStream message) {
ByteString copiedBody = ByteString.readFrom(message);
pendingDrainingMessages.add(new KnownLengthInputStream(copiedBody));
} catch (IOException e) {
rawCall.cancel("Failed to copy outbound message for buffering", e);
cancelDownstream("Failed to copy outbound message for buffering", e);
}
return;
}
Expand Down Expand Up @@ -1029,7 +1039,7 @@ public void sendMessage(InputStream message) {
}
}
} catch (IOException e) {
rawCall.cancel("Failed to serialize message for External Processor", e);
cancelDownstream("Failed to serialize message for External Processor", e);
}
}
}
Expand Down Expand Up @@ -1131,21 +1141,22 @@ public void halfClose() {
}
}

private void cancelDownstream(@Nullable String message, @Nullable Throwable cause) {
if (downstreamCancelled.compareAndSet(false, true)) {
delayedCall.cancel(message, cause);
}
void cancelDownstream(@Nullable String message, @Nullable Throwable cause) {
delayedCall.cancel(message, cause);
}

@Override
public void cancel(@Nullable String message, @Nullable Throwable cause) {
synchronized (streamLock) {
if (!extProcStreamState.get().isCompleted() && extProcClientCallRequestObserver != null) {
extProcClientCallRequestObserver.onError(
Status.CANCELLED
.withDescription(message)
.withCause(cause)
.asRuntimeException());
if (markExtProcStreamFailed(extProcStreamState)) {
if (extProcClientCallRequestObserver != null) {
extProcClientCallRequestObserver.onError(
Status.CANCELLED
.withDescription(message)
.withCause(cause)
.asRuntimeException());
extProcClientCallRequestObserver = null;
}
}
}
cancelDownstream(message, cause);
Expand Down Expand Up @@ -1318,7 +1329,7 @@ private void handleImmediateResponse(ImmediateResponse immediate, DataPlaneListe
// If sent in response to any other event, it will cause the data plane RPC to
// immediately fail with the specified status as if it were an out-of-band
// cancellation.
rawCall.cancel(status.getDescription(), null);
cancelDownstream(status.getDescription(), null);
listener.unblockAfterStreamComplete();
}
closeExtProcStream();
Expand Down Expand Up @@ -1442,7 +1453,6 @@ AtomicBoolean getIsProcessingTrailers() {
}

private static class DataPlaneListener extends SimpleForwardingClientCallListener<InputStream> {
private final ClientCall<?, ?> rawCall;
private final DataPlaneClientCall dataPlaneClientCall;
// Path 3: Upstream response bodies queued because upstream to sidestream window not available,
// response headers not cleared by ext_proc or ext_proc stream draining
Expand All @@ -1457,10 +1467,8 @@ private static class DataPlaneListener extends SimpleForwardingClientCallListene

protected DataPlaneListener(
ClientCall.Listener<InputStream> delegate,
ClientCall<?, ?> rawCall,
DataPlaneClientCall dataPlaneClientCall) {
super(delegate);
this.rawCall = rawCall;
this.dataPlaneClientCall = dataPlaneClientCall;
}

Expand Down Expand Up @@ -1538,7 +1546,7 @@ public void onMessage(InputStream message) {
ByteString copiedBody = ByteString.readFrom(message);
savedMessages.add(new KnownLengthInputStream(copiedBody));
} catch (IOException e) {
rawCall.cancel("Failed to copy inbound message for buffering", e);
dataPlaneClientCall.cancelDownstream("Failed to copy inbound message for buffering", e);
}
return;
}
Expand Down Expand Up @@ -1574,7 +1582,7 @@ public void onMessage(InputStream message) {
dataPlaneClientCall.drainPendingRequests();
}
} catch (IOException e) {
rawCall.cancel("Failed to read server response", e);
dataPlaneClientCall.cancelDownstream("Failed to read server response", e);
}
}
}
Expand All @@ -1592,7 +1600,7 @@ void drainSavedMessages() {
sendResponseBodyToExtProc(bodyByteString, false);
dataPlaneClientCall.bodyMessageSentToExtProc.set(true);
} catch (IOException e) {
rawCall.cancel("Failed to read buffered response body", e);
dataPlaneClientCall.cancelDownstream("Failed to read buffered response body", e);
}
}
}
Expand Down
Loading