From aa34ab1092f1ff01316393a6aab1e244389c27f7 Mon Sep 17 00:00:00 2001 From: Kannan J Date: Mon, 31 Aug 2026 09:44:50 +0000 Subject: [PATCH 1/2] xds: Fix TSAN data race on ClientCall cancellation in ExternalProcessorClientInterceptor Prevent concurrent cancellations of the underlying ClientCall in ExternalProcessorClientInterceptor: - Wrap rawCall with SimpleForwardingClientCall using an AtomicBoolean to ensure the underlying ClientCall.cancel() is executed at most once, even if invoked concurrently across threads or from DelayedListener. - Remove redundant downstreamCancelled AtomicBoolean from DataPlaneClientCall and simplify cancelDownstream() to directly delegate to delayedCall.cancel(), as DelayedClientCall internally synchronizes pending cancellations and the wrapped rawCall deduplicates active cancellations. - Remove the now-unused rawCall field and constructor parameter from DataPlaneListener. - In DataPlaneClientCall.cancel() and validateCompressionSupport(), atomically transition extProcStreamState to FAILED and clear extProcClientCallRequestObserver. - In sendToExtProc(), return early if the ext-proc stream is already completed or the observer is null. - Safely complete and clear extProcClientCallRequestObserver in closeExtProcStream(). - Route all rawCall.cancel() calls in sendMessage(), handleImmediateResponse(), and DataPlaneListener through cancelDownstream(). Jetski conversations: - 502db5da-b88e-474c-9c05-a14b441d3eac - 3318c948-f0f7-49ce-afe3-5670e30bf376 CONV=502db5da-b88e-474c-9c05-a14b441d3eac --- .../ExternalProcessorClientInterceptor.java | 72 ++++++++++--------- 1 file changed, 40 insertions(+), 32 deletions(-) diff --git a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java index 0bc79e5ec5d..4d5bfd73558 100644 --- a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java +++ b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java @@ -238,8 +238,18 @@ public ClientCall interceptCall( MethodDescriptor rawMethod = (MethodDescriptor) (MethodDescriptor) method; ClientCall rawCall = - (ClientCall) (ClientCall) - next.newCall(method, callOptions); + new SimpleForwardingClientCall( + (ClientCall) (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 delayedCall = @@ -315,7 +325,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 delayedCall, @@ -397,13 +406,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; @@ -419,7 +429,7 @@ public void start(Listener 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); @@ -600,6 +610,9 @@ public void onError(Throwable t) { @Override public void onCompleted() { if (markExtProcStreamCompleted(extProcStreamState)) { + synchronized (streamLock) { + extProcClientCallRequestObserver = null; + } handleFailOpen(wrappedListener); } } @@ -629,7 +642,7 @@ public void onCompleted() { private void sendToExtProc(ProcessingRequest request) { synchronized (streamLock) { - if (extProcStreamState.get().isCompleted()) { + if (extProcStreamState.get().isCompleted() || extProcClientCallRequestObserver == null) { return; } @@ -691,6 +704,7 @@ private void closeExtProcStream() { if (markExtProcStreamCompleted(extProcStreamState)) { if (extProcClientCallRequestObserver != null) { extProcClientCallRequestObserver.onCompleted(); + extProcClientCallRequestObserver = null; } } } @@ -700,11 +714,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; } } @@ -809,7 +819,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; } @@ -835,7 +845,7 @@ public void sendMessage(InputStream message) { super.sendMessage(new KnownLengthInputStream(bodyByteString)); } } catch (IOException e) { - rawCall.cancel("Failed to serialize message for External Processor", e); + cancelDownstream("Failed to serialize message for External Processor", e); } } @@ -900,21 +910,22 @@ public void halfClose() { .build()); } - 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); @@ -970,7 +981,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(); @@ -1058,7 +1069,6 @@ AtomicBoolean getIsProcessingTrailers() { } private static class DataPlaneListener extends SimpleForwardingClientCallListener { - private final ClientCall rawCall; private final DataPlaneClientCall dataPlaneClientCall; private final Queue savedMessages = new ConcurrentLinkedQueue<>(); private boolean inboundPassThrough = false; @@ -1071,10 +1081,8 @@ private static class DataPlaneListener extends SimpleForwardingClientCallListene protected DataPlaneListener( ClientCall.Listener delegate, - ClientCall rawCall, DataPlaneClientCall dataPlaneClientCall) { super(delegate); - this.rawCall = rawCall; this.dataPlaneClientCall = dataPlaneClientCall; } @@ -1153,7 +1161,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; } @@ -1184,7 +1192,7 @@ public void onMessage(InputStream message) { () -> delegate().onMessage(bodyByteString.newInput())); } } catch (IOException e) { - rawCall.cancel("Failed to read server response", e); + dataPlaneClientCall.cancelDownstream("Failed to read server response", e); } } From 900b166423f3d9c8e3c6dc64fcbd2b10fa3f96e1 Mon Sep 17 00:00:00 2001 From: Kannan J Date: Tue, 1 Sep 2026 10:03:11 +0000 Subject: [PATCH 2/2] Merge error fix --- .../java/io/grpc/xds/ExternalProcessorClientInterceptor.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java index 017b36863f4..f9352d51826 100644 --- a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java +++ b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java @@ -1600,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); } } }