core: prevent data race in ManagedChannelOrphanWrapper (#6854)

Data race was detected internally when
ManagedChannelOrphanWrapper.shutdown() was called concurrently:

WARNING: ThreadSanitizer: data race (pid=5009)
  Write of size 8 at 0x7fd2f7f37530 by thread T49:
    #0 java.lang.ref.Reference.clear()V Reference.java:265
    #1 io.grpc.internal.ManagedChannelOrphanWrapper$ManagedChannelReference.clearInternal()V ManagedChannelOrphanWrapper.java:118
    #2 io.grpc.internal.ManagedChannelOrphanWrapper$ManagedChannelReference.clear()V ManagedChannelOrphanWrapper.java:110
    #3 io.grpc.internal.ManagedChannelOrphanWrapper.shutdown()Lio/grpc/ManagedChannel; ManagedChannelOrphanWrapper.java:58
    (stacktrace redacted)

  Previous write of size 8 at 0x7fd2f7f37530 by thread T45 (mutexes: write M267260296638793720, write M267541771615505864, write M267823246592216728, write M267260296898451984, write M267541771875162784, write M267823246851873416):
    #0 java.lang.ref.Reference.clear()V Reference.java:265
    #1 io.grpc.internal.ManagedChannelOrphanWrapper$ManagedChannelReference.clearInternal()V ManagedChannelOrphanWrapper.java:118
    #2 io.grpc.internal.ManagedChannelOrphanWrapper$ManagedChannelReference.clear()V ManagedChannelOrphanWrapper.java:110
    #3 io.grpc.internal.ManagedChannelOrphanWrapper.shutdown()Lio/grpc/ManagedChannel; ManagedChannelOrphanWrapper.java:58
    (stacktrace redacted)

Author: zhangkun83

core/src/main/java/io/grpc/internal/ManagedChannelOrphanWrapper.java

@@ -24,6 +24,7 @@
 import java.lang.ref.WeakReference;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Level;
 import java.util.logging.LogRecord;
 import java.util.logging.Logger;
@@ -54,15 +55,13 @@ final class ManagedChannelOrphanWrapper extends ForwardingManagedChannel {
 
   @Override
   public ManagedChannel shutdown() {
-    phantom.shutdown = true;
-    phantom.clear();
+    phantom.clearSafely();
     return super.shutdown();
   }
 
   @Override
   public ManagedChannel shutdownNow() {
-    phantom.shutdown = true;
-    phantom.clear();
+    phantom.clearSafely();
     return super.shutdownNow();
   }
 
@@ -81,7 +80,7 @@ static final class ManagedChannelReference extends WeakReference<ManagedChannelO
 
     private final String channelStr;
     private final Reference<RuntimeException> allocationSite;
-    private volatile boolean shutdown;
+    private final AtomicBoolean shutdown = new AtomicBoolean();
 
     ManagedChannelReference(
         ManagedChannelOrphanWrapper orphanable,
@@ -113,6 +112,15 @@ public void clear() {
       cleanQueue(refqueue);
     }
 
+    /**
+     * Safe to call concurrently.
+     */
+    private void clearSafely() {
+      if (!shutdown.getAndSet(true)) {
+        clear();
+      }
+    }
+
     // avoid reentrancy
     private void clearInternal() {
       super.clear();
@@ -135,7 +143,7 @@ static int cleanQueue(ReferenceQueue<ManagedChannelOrphanWrapper> refqueue) {
       while ((ref = (ManagedChannelReference) refqueue.poll()) != null) {
         RuntimeException maybeAllocationSite = ref.allocationSite.get();
         ref.clearInternal(); // technically the reference is gone already.
-        if (!ref.shutdown) {
+        if (!ref.shutdown.get()) {
           orphanedChannels++;
           Level level = Level.SEVERE;
           if (logger.isLoggable(level)) {