Skip to content

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

Open
kannanjgithub wants to merge 3 commits into
grpc:masterfrom
kannanjgithub:tsan-data-error-2
Open

xds: Fix TSAN data race on ClientCall cancellation in ExternalProcessorClientInterceptor#13022
kannanjgithub wants to merge 3 commits into
grpc:masterfrom
kannanjgithub:tsan-data-error-2

Conversation

@kannanjgithub

Copy link
Copy Markdown
Contributor

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.
  • In DataPlaneClientCall.cancel(), atomically transition extProcStreamState to FAILED, catch exceptions during onError(), and clear extProcClientCallRequestObserver.
  • In sendToExtProc(), return early if the ext-proc stream is already completed or the observer is null, and catch unexpected onNext() exceptions to trigger internalOnError() rather than letting exceptions escape into listener callbacks.
  • Safely complete and clear extProcClientCallRequestObserver in closeExtProcStream() and halfCloseExtProcStream().
  • Route all rawCall.cancel() calls in sendMessage(), handleImmediateResponse(), and DataPlaneListener through cancelDownstream().

Fixes the TSAN detected race condition below. The previous fix in commit cf92f2d introduced AtomicBoolean downstreamCancelled in cancelDownstream(). However, the TSAN data race still reproduced under postsubmit because DelayedListener exception path bypasses cancelDownstream(): When DelayedClientCall.start(wrappedListener, headers) is called, it registers an internal DelayedListener. If wrappedListener.onHeaders() throws any exception, DelayedListener.deliverHeaders() catches it and calls exceptionThrown(), which invokes delayedCall.cancel() directly. Once delayedCall has been activated (realCall != null), DelayedClientCall.cancel() directly calls realCall.cancel() on whatever thread triggered it without any de-duplication.

Details
==================
WARNING: ThreadSanitizer: data race (pid=7774)
  Read of size 1 at 0x000089d8a086 by thread T113:
    #0 io.grpc.internal.ClientCallImpl.cancelInternal(Ljava/lang/String;Ljava/lang/Throwable;)V ClientCallImpl.java:464 
    #1 io.grpc.internal.ClientCallImpl.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V ClientCallImpl.java:455 
    #2 io.grpc.PartialForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V PartialForwardingClientCall.java:39 
    #3 io.grpc.ForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V ForwardingClientCall.java:22 
    #4 io.grpc.ForwardingClientCall$SimpleForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V ForwardingClientCall.java:44 
    #5 io.grpc.PartialForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V PartialForwardingClientCall.java:39 
    #6 io.grpc.ForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V ForwardingClientCall.java:22 
    #7 io.grpc.ForwardingClientCall$SimpleForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V ForwardingClientCall.java:44 
    #8 io.grpc.PartialForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V PartialForwardingClientCall.java:39 
    #9 io.grpc.ForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V ForwardingClientCall.java:22 
    #10 io.grpc.ForwardingClientCall$SimpleForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V ForwardingClientCall.java:44 
    #11 io.grpc.internal.DelayedClientCall$2.run()V DelayedClientCall.java:269 
    #12 io.grpc.internal.DelayedClientCall.delayOrExecute(Ljava/lang/Runnable;)V DelayedClientCall.java:292 
    #13 io.grpc.internal.DelayedClientCall.cancel(Lio/grpc/Status;Z)V DelayedClientCall.java:266 
    #14 io.grpc.internal.DelayedClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V DelayedClientCall.java:242 
    #15 io.grpc.xds.ExternalProcessorClientInterceptor$DataPlaneClientCall.internalOnError(Ljava/lang/Throwable;)V ExternalProcessorClientInterceptor.java:715 
    #16 io.grpc.xds.ExternalProcessorClientInterceptor$DataPlaneClientCall$1.onNext(Lio/envoyproxy/envoy/service/ext_proc/v3/ProcessingResponse;)V ExternalProcessorClientInterceptor.java:457 
    #17 io.grpc.xds.ExternalProcessorClientInterceptor$DataPlaneClientCall$1.onNext(Ljava/lang/Object;)V ExternalProcessorClientInterceptor.java:426 
    #18 io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onMessage(Ljava/lang/Object;)V ClientCalls.java:568 
    #19 io.grpc.ForwardingClientCallListener.onMessage(Ljava/lang/Object;)V ForwardingClientCallListener.java:33 
    #20 io.grpc.ForwardingClientCallListener.onMessage(Ljava/lang/Object;)V ForwardingClientCallListener.java:33 
    #21 io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1MessagesAvailable.runInternal()V ClientCallImpl.java:662 
    #22 io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1MessagesAvailable.runInContext()V ClientCallImpl.java:649 
    #23 io.grpc.internal.ContextRunnable.run()V ContextRunnable.java:37 
    #24 io.grpc.internal.SerializingExecutor.run()V SerializingExecutor.java:133 
    #25 io.grpc.internal.SerializingExecutor.run()V SerializingExecutor.java:133 
    #26 java.util.concurrent.ThreadPoolExecutor.runWorker(Ljava/util/concurrent/ThreadPoolExecutor$Worker;)V ThreadPoolExecutor.java:1090 
    #27 java.util.concurrent.ThreadPoolExecutor$Worker.run()V ThreadPoolExecutor.java:614 
    #28 java.lang.Thread.runWith(Ljava/lang/Object;Ljava/lang/Runnable;)V Thread.java:1560 
    #29 java.lang.Thread.run()V Thread.java:1547 
    #30 (Generated Stub) <null> 

  Previous write of size 1 at 0x000089d8a086 by thread T29:
    #0 io.grpc.internal.ClientCallImpl.cancelInternal(Ljava/lang/String;Ljava/lang/Throwable;)V ClientCallImpl.java:467 
    #1 io.grpc.internal.ClientCallImpl.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V ClientCallImpl.java:455 
    #2 io.grpc.PartialForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V PartialForwardingClientCall.java:39 
    #3 io.grpc.ForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V ForwardingClientCall.java:22 
    #4 io.grpc.ForwardingClientCall$SimpleForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V ForwardingClientCall.java:44 
    #5 io.grpc.PartialForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V PartialForwardingClientCall.java:39 
    #6 io.grpc.ForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V ForwardingClientCall.java:22 
    #7 io.grpc.ForwardingClientCall$SimpleForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V ForwardingClientCall.java:44 
    #8 io.grpc.PartialForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V PartialForwardingClientCall.java:39 
    #9 io.grpc.ForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V ForwardingClientCall.java:22 
    #10 io.grpc.ForwardingClientCall$SimpleForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V ForwardingClientCall.java:44 
    #11 io.grpc.internal.DelayedClientCall$2.run()V DelayedClientCall.java:269 
    #12 io.grpc.internal.DelayedClientCall.delayOrExecute(Ljava/lang/Runnable;)V DelayedClientCall.java:292 
    #13 io.grpc.internal.DelayedClientCall.cancel(Lio/grpc/Status;Z)V DelayedClientCall.java:266 
    #14 io.grpc.internal.DelayedClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V DelayedClientCall.java:242 
    #15 io.grpc.PartialForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V PartialForwardingClientCall.java:39 
    #16 io.grpc.ForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V ForwardingClientCall.java:22 
    #17 io.grpc.ForwardingClientCall$SimpleForwardingClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V ForwardingClientCall.java:44 
    #18 io.grpc.xds.ExternalProcessorClientInterceptor$DataPlaneClientCall.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V ExternalProcessorClientInterceptor.java:887 
    #19 io.grpc.xds.XdsNameResolver$RawMessageClientInterceptor$2.cancel(Ljava/lang/String;Ljava/lang/Throwable;)V XdsNameResolver.java:1182 
    #20 io.grpc.xds.ExternalProcessorClientInterceptorTest.clientInterceptor_contextPropagatedToStartCall()V ExternalProcessorClientInterceptorTest.java:13411 

…orClientInterceptor

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
# Conflicts:
#	xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant