core: Release lock before closing shared resource

If a resource has dependencies that also use SharedResourceHolder, and
its close() blocks waiting for processing on another thread, then the
threads could become deadlocked on the SharedResourceHolder lock. Our
answer should be "don't block," but it's also good to avoid calling
arbitrary code with a lock held.

create() is still called with the lock held, but that seems less likely
to do work on another thread, and it is harder to avoid the lock.
close() is very easy to call without the lock.

See d50098f and b/458736211

Author: ejona86

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

@@ -134,18 +134,16 @@ synchronized <T> T releaseInternal(final Resource<T> resource, final T instance)
         public void run() {
           synchronized (SharedResourceHolder.this) {
             // Refcount may have gone up since the task was scheduled. Re-check it.
-            if (cached.refcount == 0) {
-              try {
-                resource.close(instance);
-              } finally {
-                instances.remove(resource);
-                if (instances.isEmpty()) {
-                  destroyer.shutdown();
-                  destroyer = null;
-                }
-              }
+            if (cached.refcount != 0) {
+              return;
+            }
+            instances.remove(resource);
+            if (instances.isEmpty()) {
+              destroyer.shutdown();
+              destroyer = null;
             }
           }
+          resource.close(instance);
         }
       }), DESTROY_DELAY_SECONDS, TimeUnit.SECONDS);
     }

core/src/test/java/io/grpc/internal/SharedResourceHolderTest.java

@@ -30,7 +30,9 @@
 
 import io.grpc.internal.SharedResourceHolder.Resource;
 import java.util.LinkedList;
+import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.Delayed;
+import java.util.concurrent.FutureTask;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
@@ -201,6 +203,46 @@ public void close(ResourceInstance instance) {
     assertNotSame(instance, holder.getInternal(resource));
   }
 
+  @Test(timeout = 5000)
+  public void closeRunsConcurrently() throws Exception {
+    CyclicBarrier barrier = new CyclicBarrier(2);
+    class SlowResource implements Resource<ResourceInstance> {
+      @Override
+      public ResourceInstance create() {
+        return new ResourceInstance();
+      }
+
+      @Override
+      public void close(ResourceInstance instance) {
+        instance.closed = true;
+        try {
+          barrier.await();
+          barrier.await();
+        } catch (Exception ex) {
+          throw new AssertionError(ex);
+        }
+      }
+    }
+
+    Resource<ResourceInstance> resource = new SlowResource();
+    ResourceInstance instance = holder.getInternal(resource);
+    holder.releaseInternal(resource, instance);
+    MockScheduledFuture<?> scheduledDestroyTask = scheduledDestroyTasks.poll();
+    FutureTask<Void> runTask = new FutureTask<>(scheduledDestroyTask::runTask, null);
+    Thread t = new Thread(runTask);
+    t.start();
+
+    barrier.await(); // Ensure the other thread has blocked
+    assertTrue(instance.closed);
+    instance = holder.getInternal(resource);
+    assertFalse(instance.closed);
+    holder.releaseInternal(resource, instance);
+
+    barrier.await(); // Resume the other thread
+    t.join();
+    runTask.get(); // Check for exception
+  }
+
   private class MockExecutorFactory implements
       SharedResourceHolder.ScheduledExecutorFactory {
     @Override