xds: Prevent concurrent cancellations of data plane call in ExternalProcessorClientInterceptor (grpc#12996)

If the ext-proc stream terminates with a non-OK status, the interceptor
cancels the downstream data plane call (on the executor thread).
Concurrently, the application thread may call cancel() on the returned
proxy call (e.g. for user cancellation or cleanup).

Since ClientCallImpl does not synchronize its cancelCalled field,
concurrent cancellations from these two threads resulted in a TSAN data
race (that was discussed in
grpc#12975).

This commit introduces an AtomicBoolean downstreamCancelled in
DataPlaneClientCall to guard all downstream cancellations, ensuring that
only the first cancellation is forwarded to delayedCall/super.cancel().
This deduplicates and serializes cancellation handling from both the
application and the interceptor threads, preventing the data race.

Author: kannanjgithub

xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java

@@ -315,6 +315,7 @@ 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,
@@ -403,7 +404,7 @@ private boolean validateCompressionSupport(BodyResponse bodyResponse) {
           }
           activateCall();
           markExtProcStreamFailed(extProcStreamState);
-          delayedCall.cancel("gRPC message compression not supported in ext_proc", ex);
+          cancelDownstream("gRPC message compression not supported in ext_proc", ex);
           closeExtProcStream();
           return false;
         }
@@ -590,7 +591,7 @@ public void onError(Throwable t) {
               handleFailOpen(wrappedListener);
             } else {
               String message = "External processor stream failed";
-              delayedCall.cancel(message, t);
+              cancelDownstream(message, t);
               wrappedListener.proceedWithClose();
             }
           }
@@ -712,7 +713,7 @@ private void internalOnError(Throwable t) {
           handleFailOpen(wrappedListener);
         } else {
           String message = "External processor stream failed";
-          delayedCall.cancel(message, t);
+          cancelDownstream(message, t);
           wrappedListener.proceedWithClose();
         }
       }
@@ -899,6 +900,12 @@ public void halfClose() {
           .build());
     }
 
+    private void cancelDownstream(@Nullable String message, @Nullable Throwable cause) {
+      if (downstreamCancelled.compareAndSet(false, true)) {
+        delayedCall.cancel(message, cause);
+      }
+    }
+
     @Override
     public void cancel(@Nullable String message, @Nullable Throwable cause) {
       synchronized (streamLock) {
@@ -910,7 +917,7 @@ public void cancel(@Nullable String message, @Nullable Throwable cause) {
                   .asRuntimeException());
         }
       }
-      super.cancel(message, cause);
+      cancelDownstream(message, cause);
     }
 
     private void handleRequestBodyResponse(BodyResponse bodyResponse) {