From aa34ab1092f1ff01316393a6aab1e244389c27f7 Mon Sep 17 00:00:00 2001 From: Kannan J Date: Mon, 31 Aug 2026 09:44:50 +0000 Subject: [PATCH 1/3] 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/3] 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); } } } From a08bdb9a5ce5e40797d1c22acf4a492734f34f69 Mon Sep 17 00:00:00 2001 From: Kannan J Date: Tue, 8 Sep 2026 09:28:27 +0000 Subject: [PATCH 3/3] xds: Add unit test for concurrent cancellation in ExternalProcessorClientInterceptor Add unit test verifying that when a call is cancelled concurrently from multiple threads while a data plane response is arriving: - The underlying ClientCall.cancel() is executed exactly once, verifying deduplication and thread-safety of the wrapped rawCall. - The external processor stream observer receives onError at most once, and no requests are sent to ext-proc after cancellation. Jetski conversations: - 502db5da-b88e-474c-9c05-a14b441d3eac - 3318c948-f0f7-49ce-afe3-5670e30bf376 CONV=502db5da-b88e-474c-9c05-a14b441d3eac --- ...xternalProcessorClientInterceptorTest.java | 160 ++++++++++++++++++ 1 file changed, 160 insertions(+) diff --git a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java index 0701b670b5d..eeae69bbdb4 100644 --- a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java +++ b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java @@ -90,9 +90,11 @@ import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CyclicBarrier; import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -21129,6 +21131,164 @@ public void onClose(Status status, Metadata trailers) { channelManager.close(); } + @Test + public void whenCallCancelledConcurrently_underlyingCallCancelledExactlyOnceAndNoExtProcLeak() + throws Exception { + String uniqueExtProcServerName = InProcessServerBuilder.generateName(); + String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); + + ExternalProcessor proto = createBaseProto(uniqueExtProcServerName) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SEND) + .setResponseHeaderMode(ProcessingMode.HeaderSendMode.SEND) + .build()) + .build(); + ExternalProcessorFilterConfig filterConfig = + provider.parseFilterConfig(Any.pack(proto), filterContext).config; + + final AtomicInteger extProcErrorCount = new AtomicInteger(); + final AtomicInteger extProcRequestsAfterError = new AtomicInteger(); + final AtomicBoolean extProcHadError = new AtomicBoolean(false); + + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + public StreamObserver process( + StreamObserver responseObserver) { + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (extProcHadError.get()) { + extProcRequestsAfterError.incrementAndGet(); + } + if (request.hasRequestHeaders()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestHeaders(HeadersResponse.newBuilder().build()) + .build()); + } + } + + @Override + public void onError(Throwable t) { + extProcHadError.set(true); + extProcErrorCount.incrementAndGet(); + } + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + } + }; + + grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build()); + }); + + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + final CountDownLatch dataPlaneServerStartedLatch = new CountDownLatch(1); + final CountDownLatch releaseServerResponseLatch = new CountDownLatch(1); + + MutableHandlerRegistry dataPlaneRegistry = new MutableHandlerRegistry(); + dataPlaneRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_SAY_HELLO, ServerCalls.asyncUnaryCall((request, responseObserver) -> { + dataPlaneServerStartedLatch.countDown(); + try { + releaseServerResponseLatch.await(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + responseObserver.onNext("Hello " + request); + responseObserver.onCompleted(); + })) + .build()); + + grpcCleanup.register(InProcessServerBuilder.forName(uniqueDataPlaneServerName) + .fallbackHandlerRegistry(dataPlaneRegistry) + .directExecutor() + .build().start()); + + final AtomicInteger rawCallCancelCount = new AtomicInteger(); + ClientInterceptor countingInterceptor = new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new SimpleForwardingClientCall(next.newCall(method, callOptions)) { + @Override + public void cancel(String message, Throwable cause) { + rawCallCancelCount.incrementAndGet(); + super.cancel(message, cause); + } + }; + } + }; + + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueDataPlaneServerName) + .intercept(countingInterceptor) + .directExecutor() + .build()); + + ClientCall proxyCall = interceptCall( + interceptor, + METHOD_SAY_HELLO, + DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()), + dataPlaneChannel); + + proxyCall.start(new ClientCall.Listener() {}, new Metadata()); + + proxyCall.request(1); + proxyCall.sendMessage("ping"); + proxyCall.halfClose(); + + // Wait until the data plane call has started and is active + assertThat(dataPlaneServerStartedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + // Concurrently release server response and trigger multiple concurrent cancellations + int numCancelThreads = 8; + ExecutorService executor = Executors.newFixedThreadPool(numCancelThreads + 1); + CyclicBarrier barrier = new CyclicBarrier(numCancelThreads + 1); + List> futures = new ArrayList<>(); + + futures.add(executor.submit(() -> { + barrier.await(); + releaseServerResponseLatch.countDown(); + return null; + })); + + for (int i = 0; i < numCancelThreads; i++) { + final int threadId = i; + futures.add(executor.submit(() -> { + barrier.await(); + proxyCall.cancel("Cancel from thread " + threadId, null); + return null; + })); + } + + for (Future f : futures) { + f.get(5, TimeUnit.SECONDS); + } + + // Underlying rawCall was cancelled exactly once despite multiple concurrent cancellations + assertThat(rawCallCancelCount.get()).isEqualTo(1); + // Ext-proc received onError at most once + assertThat(extProcErrorCount.get()).isAtMost(1); + // Ext-proc never received any requests after being cancelled + assertThat(extProcRequestsAfterError.get()).isEqualTo(0); + + channelManager.close(); + shutdownAndAwaitTermination(executor); + } + private static List filterClientRequests(List requests) { List clientRequests = new ArrayList<>(); for (ProcessingRequest r : requests) {