Provide blockhound integrations

.github/workflows/actions_build.yml

@@ -45,6 +45,8 @@ jobs:
           - java: 19
             on: self-hosted
             snapshot: true
+            # blockhound makes the build run about 10 minutes slower
+            blockhound: true
 
     steps:
       - uses: actions/checkout@v2
@@ -86,6 +88,7 @@ jobs:
           ${{ (matrix.on == 'self-hosted') && '--max-workers=8' || '--max-workers=2' }} --parallel \
           ${{ matrix.coverage && '-Pcoverage' || '' }} \
           ${{ matrix.leak && '-Pleak' || '' }} \
+          ${{ matrix.blockhound && '-Pblockhound' || '' }} \
           -PnoLint \
           -PflakyTests=false \
           -PbuildJdkVersion=${{ env.BUILD_JDK_VERSION }} \
@@ -133,17 +136,23 @@ jobs:
         if: ${{ matrix.coverage }}
         uses: codecov/codecov-action@v1
 
+      - name: Fail the run if any threads were blocked
+        if: ${{ matrix.blockhound }}
+        run: "if [[ -z `find . -name 'blockhound.log' -size +0` ]]; then exit 0; else exit 1; fi"
+        shell: bash
+
       - name: Collect the test reports
         if: failure()
-        run: find . '(' -name 'java_pid*.hprof' -or -name 'hs_err_*.log' -or -path '*/build/reports/tests' -or -path '*/build/test-results' -or -path '*/javadoc.options' ')' -exec tar rf "reports-JVM-${{ matrix.java }}.tar" {} ';'
+        run: |
+          find . '(' -name 'java_pid*.hprof' -or -name 'hs_err_*.log' -or -path '*/build/reports/tests' -or -path '*/build/test-results' -or -path '*/javadoc.options' -or -name 'blockhound.log' ')' -exec tar rf "reports-JVM-${{ matrix.on }}-${{ matrix.java }}.tar" {} ';'
         shell: bash
 
       - name: Upload the artifacts
         if: failure()
         uses: actions/upload-artifact@v2
         with:
-          name: reports-JVM-${{ matrix.java }}
-          path: reports-JVM-${{ matrix.java }}.tar
+          name: reports-JVM-${{ matrix.on }}-${{ matrix.java }}
+          path: reports-JVM-${{ matrix.on }}-${{ matrix.java }}.tar
           retention-days: 3
 
   lint:

brave/src/main/java/com/linecorp/armeria/common/brave/BraveBlockHoundIntegration.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2022 LINE Corporation
+ *
+ * LINE Corporation licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.linecorp.armeria.common.brave;
+
+import com.linecorp.armeria.common.annotation.UnstableApi;
+
+import reactor.blockhound.BlockHound.Builder;
+import reactor.blockhound.integration.BlockHoundIntegration;
+
+/**
+ * A {@link BlockHoundIntegration} for the brave module.
+ */
+@UnstableApi
+public final class BraveBlockHoundIntegration implements BlockHoundIntegration {
+
+    @Override
+    public void applyTo(Builder builder) {
+        builder.allowBlockingCallsInside("zipkin2.reporter.AsyncReporter$BoundedAsyncReporter", "report");
+    }
+}

brave/src/main/resources/META-INF/services/reactor.blockhound.integration.BlockHoundIntegration

@@ -0,0 +1 @@
+com.linecorp.armeria.common.brave.BraveBlockHoundIntegration

brave/src/test/java/com/linecorp/armeria/it/brave/BraveIntegrationTest.java

@@ -65,8 +65,8 @@
 import com.linecorp.armeria.common.brave.HelloService.AsyncIface;
 import com.linecorp.armeria.common.brave.RequestContextCurrentTraceContext;
 import com.linecorp.armeria.common.thrift.ThriftFuture;
-import com.linecorp.armeria.common.util.Exceptions;
 import com.linecorp.armeria.common.util.ThreadFactories;
+import com.linecorp.armeria.internal.testing.BlockingUtils;
 import com.linecorp.armeria.server.AbstractHttpService;
 import com.linecorp.armeria.server.HttpService;
 import com.linecorp.armeria.server.ServerBuilder;
@@ -569,17 +569,13 @@ private static class SpanHandlerImpl extends SpanHandler {
 
         @Override
         public boolean end(TraceContext context, MutableSpan span, Cause cause) {
-            return spans.add(span);
+            return BlockingUtils.blockingRun(() -> spans.add(span));
         }
 
         MutableSpan[] take(int numSpans) {
             final List<MutableSpan> taken = new ArrayList<>();
             while (taken.size() < numSpans) {
-                try {
-                    taken.add(spans.poll(30, TimeUnit.SECONDS));
-                } catch (InterruptedException e) {
-                    return Exceptions.throwUnsafely(e);
-                }
+                BlockingUtils.blockingRun(() -> taken.add(spans.poll(30, TimeUnit.SECONDS)));
             }
 
             // Reverse the collected spans to sort the spans by request time.

build.gradle

@@ -78,10 +78,17 @@ allprojects {
             systemProperty "java.security.manager", "allow"
         }
 
+        // required by blockhound for jvm 13+. See https://github.com/reactor/BlockHound/issues/33.
+        if (rootProject.ext.testJavaVersion >= 13) {
+            jvmArgs "-XX:+AllowRedefinitionToAddDeleteMethods"
+        }
+
         // Use verbose exception/response reporting for easier debugging.
         systemProperty 'com.linecorp.armeria.verboseExceptions', 'true'
         systemProperty 'com.linecorp.armeria.verboseResponses', 'true'
 
+        systemProperty 'com.linecorp.armeria.blockhound.reportFile', "${project.buildDir}/blockhound.log"
+
         // Pass special system property to tell our tests that we are measuring coverage.
         if (project.hasFlags('coverage')) {
             systemProperty 'com.linecorp.armeria.testing.coverage', 'true'
@@ -165,6 +172,9 @@ configure(projectsWithFlags('java')) {
         // Reflections
         implementation libs.reflections
 
+        // Blockhound
+        optionalImplementation libs.blockhound
+
         // Test-time dependencies
         testImplementation libs.guava.testlib
         testImplementation libs.junit4
@@ -183,6 +193,11 @@ configure(projectsWithFlags('java')) {
         testImplementation libs.apache.httpclient5
         testImplementation libs.hamcrest
         testImplementation libs.hamcrest.library
+        testRuntimeOnly libs.kotlin.coroutines.debug
+
+        if (rootProject.hasProperty('blockhound')) {
+            testRuntimeOnly libs.blockhound.junit.platform
+        }
     }
 
     // Configure the default DuplicatesStrategy for such as:

core/src/main/java/com/linecorp/armeria/client/AbstractEventLoopState.java

@@ -21,6 +21,8 @@
 
 import com.google.common.annotations.VisibleForTesting;
 
+import com.linecorp.armeria.internal.common.util.ReentrantShortLock;
+
 import io.netty.channel.EventLoop;
 
 abstract class AbstractEventLoopState {
@@ -35,7 +37,7 @@ static AbstractEventLoopState of(List<EventLoop> eventLoops, int maxNumEventLoop
         return new HeapBasedEventLoopState(eventLoops, maxNumEventLoops, scheduler);
     }
 
-    private final ReentrantLock lock = new ReentrantLock();
+    private final ReentrantLock lock = new ReentrantShortLock();
     private final List<EventLoop> eventLoops;
     private final DefaultEventLoopScheduler scheduler;
 

core/src/main/java/com/linecorp/armeria/client/DefaultEventLoopScheduler.java

@@ -41,6 +41,7 @@
 import com.linecorp.armeria.common.SessionProtocol;
 import com.linecorp.armeria.common.annotation.Nullable;
 import com.linecorp.armeria.common.util.ReleasableHolder;
+import com.linecorp.armeria.internal.common.util.ReentrantShortLock;
 
 import io.netty.channel.EventLoop;
 import io.netty.channel.EventLoopGroup;
@@ -58,7 +59,7 @@ final class DefaultEventLoopScheduler implements EventLoopScheduler {
 
     static final int DEFAULT_MAX_NUM_EVENT_LOOPS = 1;
 
-    private final ReentrantLock lock = new ReentrantLock();
+    private final ReentrantLock lock = new ReentrantShortLock();
 
     private final List<EventLoop> eventLoops;
 

core/src/main/java/com/linecorp/armeria/client/cookie/DefaultCookieJar.java

@@ -34,6 +34,7 @@
 import com.linecorp.armeria.common.Cookies;
 import com.linecorp.armeria.common.Scheme;
 import com.linecorp.armeria.common.SessionProtocol;
+import com.linecorp.armeria.internal.common.util.ReentrantShortLock;
 
 import io.netty.util.NetUtil;
 import it.unimi.dsi.fastutil.objects.Object2LongOpenHashMap;
@@ -60,7 +61,7 @@ final class DefaultCookieJar implements CookieJar {
         this.cookiePolicy = cookiePolicy;
         store = new Object2LongOpenHashMap<>();
         filter = new HashMap<>();
-        lock = new ReentrantLock();
+        lock = new ReentrantShortLock();
     }
 
     @Override

core/src/main/java/com/linecorp/armeria/client/endpoint/DynamicEndpointGroup.java

@@ -30,7 +30,6 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
 
 import com.google.common.base.MoreObjects;
 import com.google.common.base.MoreObjects.ToStringHelper;
@@ -45,6 +44,7 @@
 import com.linecorp.armeria.common.util.AsyncCloseableSupport;
 import com.linecorp.armeria.common.util.EventLoopCheckingFuture;
 import com.linecorp.armeria.common.util.ListenableAsyncCloseable;
+import com.linecorp.armeria.internal.common.util.ReentrantShortLock;
 
 /**
  * A dynamic {@link EndpointGroup}. The list of {@link Endpoint}s can be updated dynamically.
@@ -66,7 +66,7 @@ public static DynamicEndpointGroupBuilder builder() {
     private final EndpointSelectionStrategy selectionStrategy;
     private final AtomicReference<EndpointSelector> selector = new AtomicReference<>();
     private volatile List<Endpoint> endpoints = UNINITIALIZED_ENDPOINTS;
-    private final Lock endpointsLock = new ReentrantLock();
+    private final Lock endpointsLock = new ReentrantShortLock();
 
     private final CompletableFuture<List<Endpoint>> initialEndpointsFuture = new InitialEndpointsFuture();
     private final AsyncCloseableSupport closeable = AsyncCloseableSupport.of(this::closeAsync);

core/src/main/java/com/linecorp/armeria/client/endpoint/FileWatcherRegistry.java

@@ -38,6 +38,7 @@
 import com.google.common.base.MoreObjects;
 
 import com.linecorp.armeria.client.endpoint.FileWatcherRunnable.FileWatchEvent;
+import com.linecorp.armeria.internal.common.util.ReentrantShortLock;
 
 /**
  * A registry which wraps a {@link WatchService} and allows paths to be registered.
@@ -133,7 +134,7 @@ void close() throws IOException {
 
     private final Map<FileSystem, FileSystemWatchContext> fileSystemWatchServiceMap =
             new HashMap<>();
-    private final ReentrantLock lock = new ReentrantLock();
+    private final ReentrantLock lock = new ReentrantShortLock();
 
     /**
      * Registers a {@code filePath} and {@code callback} to the {@link WatchService}. When the

core/src/main/java/com/linecorp/armeria/client/endpoint/RestartableThread.java

@@ -22,13 +22,14 @@
 import java.util.function.Supplier;
 
 import com.linecorp.armeria.common.annotation.Nullable;
+import com.linecorp.armeria.internal.common.util.ReentrantShortLock;
 
 /**
  * A restartable thread utility class.
  */
 final class RestartableThread {
 
-    private final ReentrantLock lock = new ReentrantLock();
+    private final ReentrantLock lock = new ReentrantShortLock();
 
     @Nullable
     private Thread thread;

core/src/main/java/com/linecorp/armeria/client/endpoint/WeightedRandomDistributionEndpointSelector.java

@@ -27,6 +27,7 @@
 
 import com.linecorp.armeria.client.Endpoint;
 import com.linecorp.armeria.common.annotation.Nullable;
+import com.linecorp.armeria.internal.common.util.ReentrantShortLock;
 
 /**
  * This selector selects an {@link Endpoint} using random and the weight of the {@link Endpoint}. If there are
@@ -37,7 +38,7 @@
  */
 final class WeightedRandomDistributionEndpointSelector {
 
-    private final ReentrantLock lock = new ReentrantLock();
+    private final ReentrantLock lock = new ReentrantShortLock();
     private final List<Entry> allEntries;
     @GuardedBy("lock")
     private final List<Entry> currentEntries;

core/src/main/java/com/linecorp/armeria/client/endpoint/healthcheck/DefaultHealthCheckerContext.java

@@ -43,6 +43,7 @@
 import com.linecorp.armeria.common.annotation.Nullable;
 import com.linecorp.armeria.common.util.AsyncCloseable;
 import com.linecorp.armeria.common.util.EventLoopCheckingFuture;
+import com.linecorp.armeria.internal.common.util.ReentrantShortLock;
 
 import io.netty.channel.EventLoopGroup;
 import io.netty.util.concurrent.Future;
@@ -54,7 +55,7 @@ final class DefaultHealthCheckerContext
     private final Endpoint endpoint;
     private final SessionProtocol protocol;
     private final ClientOptions clientOptions;
-    private final ReentrantLock lock = new ReentrantLock();
+    private final ReentrantLock lock = new ReentrantShortLock();
 
     /**
      * Keeps the {@link Future}s which were scheduled via this {@link ScheduledExecutorService}.

core/src/main/java/com/linecorp/armeria/client/endpoint/healthcheck/HealthCheckedEndpointGroup.java

@@ -49,6 +49,7 @@
 import com.linecorp.armeria.common.annotation.Nullable;
 import com.linecorp.armeria.common.metric.MeterIdPrefix;
 import com.linecorp.armeria.common.util.AsyncCloseable;
+import com.linecorp.armeria.internal.common.util.ReentrantShortLock;
 
 import io.micrometer.core.instrument.binder.MeterBinder;
 
@@ -110,7 +111,7 @@ public static HealthCheckedEndpointGroupBuilder builder(EndpointGroup delegate,
     @VisibleForTesting
     final HealthCheckStrategy healthCheckStrategy;
 
-    private final ReentrantLock lock = new ReentrantLock();
+    private final ReentrantLock lock = new ReentrantShortLock();
     @GuardedBy("lock")
     private final Deque<HealthCheckContextGroup> contextGroupChain = new ArrayDeque<>(4);
 

core/src/main/java/com/linecorp/armeria/client/endpoint/healthcheck/HttpHealthChecker.java

@@ -49,6 +49,7 @@
 import com.linecorp.armeria.common.util.AsyncCloseable;
 import com.linecorp.armeria.common.util.AsyncCloseableSupport;
 import com.linecorp.armeria.common.util.TimeoutMode;
+import com.linecorp.armeria.internal.common.util.ReentrantShortLock;
 import com.linecorp.armeria.unsafe.PooledObjects;
 
 import io.netty.util.AsciiString;
@@ -60,7 +61,7 @@ final class HttpHealthChecker implements AsyncCloseable {
 
     private static final AsciiString ARMERIA_LPHC = HttpHeaderNames.of("armeria-lphc");
 
-    private final ReentrantLock lock = new ReentrantLock();
+    private final ReentrantLock lock = new ReentrantShortLock();
     private final HealthCheckerContext ctx;
     private final WebClient webClient;
     private final String authority;