core: Release lock before closing shared resource

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