From 93f10890287d793721c459c80fcb5df5d889e447 Mon Sep 17 00:00:00 2001 From: Andrew Azores Date: Thu, 7 Sep 2023 13:23:46 -0400 Subject: [PATCH] fix(startup): platform detection progresses with limited parallelism (#1658) * fix(threads): use independent pools where sensible rather than ForkJoinPool.commonPool() * fix(vertx): ensure minimum event loop pool size --- .../cryostat/discovery/DiscoveryModule.java | 2 + .../cryostat/discovery/DiscoveryStorage.java | 41 ++++++++++--------- .../java/io/cryostat/net/AgentClient.java | 19 ++++----- .../java/io/cryostat/net/NetworkModule.java | 18 +++++--- .../internal/DockerPlatformStrategy.java | 7 ++-- .../internal/PodmanPlatformStrategy.java | 10 ++--- .../cryostat/recordings/RecordingsModule.java | 6 +-- .../discovery/DiscoveryStorageTest.java | 2 + 8 files changed, 58 insertions(+), 47 deletions(-) diff --git a/src/main/java/io/cryostat/discovery/DiscoveryModule.java b/src/main/java/io/cryostat/discovery/DiscoveryModule.java index 4d70250f8a..830ffe1697 100644 --- a/src/main/java/io/cryostat/discovery/DiscoveryModule.java +++ b/src/main/java/io/cryostat/discovery/DiscoveryModule.java @@ -17,6 +17,7 @@ import java.time.Duration; import java.util.Set; +import java.util.concurrent.Executors; import javax.inject.Named; import javax.inject.Singleton; @@ -79,6 +80,7 @@ static DiscoveryStorage provideDiscoveryStorage( Logger logger) { return new DiscoveryStorage( deployer, + Executors.newCachedThreadPool(), pingPeriod, builtin, dao, diff --git a/src/main/java/io/cryostat/discovery/DiscoveryStorage.java b/src/main/java/io/cryostat/discovery/DiscoveryStorage.java index 2d480b21c8..315d6ea135 100644 --- a/src/main/java/io/cryostat/discovery/DiscoveryStorage.java +++ b/src/main/java/io/cryostat/discovery/DiscoveryStorage.java @@ -28,7 +28,7 @@ import java.util.Optional; import java.util.UUID; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.ExecutorService; import java.util.function.Predicate; import javax.script.ScriptException; @@ -69,6 +69,7 @@ public class DiscoveryStorage extends AbstractPlatformClientVerticle { public static final URI NO_CALLBACK = null; private final Duration pingPeriod; private final VerticleDeployer deployer; + private final ExecutorService executor; private final Lazy builtin; private final PluginInfoDao dao; private final Lazy jvmIdHelper; @@ -86,6 +87,7 @@ public class DiscoveryStorage extends AbstractPlatformClientVerticle { DiscoveryStorage( VerticleDeployer deployer, + ExecutorService executor, Duration pingPeriod, Lazy builtin, PluginInfoDao dao, @@ -96,6 +98,7 @@ public class DiscoveryStorage extends AbstractPlatformClientVerticle { WebClient http, Logger logger) { this.deployer = deployer; + this.executor = executor; this.pingPeriod = pingPeriod; this.builtin = builtin; this.dao = dao; @@ -183,26 +186,24 @@ public void start(Promise future) throws Exception { } private void testNonConnectedTargets(Predicate> predicate) { - ForkJoinPool.commonPool() - .execute( - () -> { - Map copy = new HashMap<>(nonConnectableTargets); - for (var entry : copy.entrySet()) { - try { - if (predicate.test(entry)) { - nonConnectableTargets.remove(entry.getKey()); - UUID id = entry.getValue(); - PluginInfo plugin = getById(id).orElseThrow(); - EnvironmentNode original = - gson.fromJson( - plugin.getSubtree(), EnvironmentNode.class); - update(id, original.getChildren()); - } - } catch (JsonSyntaxException e) { - throw new RuntimeException(e); - } + executor.execute( + () -> { + Map copy = new HashMap<>(nonConnectableTargets); + for (var entry : copy.entrySet()) { + try { + if (predicate.test(entry)) { + nonConnectableTargets.remove(entry.getKey()); + UUID id = entry.getValue(); + PluginInfo plugin = getById(id).orElseThrow(); + EnvironmentNode original = + gson.fromJson(plugin.getSubtree(), EnvironmentNode.class); + update(id, original.getChildren()); } - }); + } catch (JsonSyntaxException e) { + throw new RuntimeException(e); + } + } + }); } @Override diff --git a/src/main/java/io/cryostat/net/AgentClient.java b/src/main/java/io/cryostat/net/AgentClient.java index 84b81c6094..eacde81eed 100644 --- a/src/main/java/io/cryostat/net/AgentClient.java +++ b/src/main/java/io/cryostat/net/AgentClient.java @@ -23,7 +23,7 @@ import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; -import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.ExecutorService; import javax.management.ObjectName; import javax.script.ScriptException; @@ -49,7 +49,6 @@ import com.google.gson.Gson; import io.vertx.core.Future; -import io.vertx.core.Vertx; import io.vertx.core.http.HttpMethod; import io.vertx.core.json.JsonArray; import io.vertx.core.json.JsonObject; @@ -64,7 +63,7 @@ public class AgentClient { public static final String NULL_CREDENTIALS = "No credentials found for agent"; - private final Vertx vertx; + private final ExecutorService executor; private final Gson gson; private final long httpTimeout; private final WebClient webClient; @@ -73,14 +72,14 @@ public class AgentClient { private final Logger logger; AgentClient( - Vertx vertx, + ExecutorService executor, Gson gson, long httpTimeout, WebClient webClient, CredentialsManager credentialsManager, URI agentUri, Logger logger) { - this.vertx = vertx; + this.executor = executor; this.gson = gson; this.httpTimeout = httpTimeout; this.webClient = webClient; @@ -237,7 +236,7 @@ private Future> invoke(HttpMethod mtd, String path, BodyCode throw new RuntimeException(e); } }, - ForkJoinPool.commonPool()) + executor) .exceptionally( t -> { throw new RuntimeException(t); @@ -246,7 +245,7 @@ private Future> invoke(HttpMethod mtd, String path, BodyCode static class Factory { - private final Vertx vertx; + private final ExecutorService executor; private final Gson gson; private final long httpTimeout; private final WebClient webClient; @@ -254,13 +253,13 @@ static class Factory { private final Logger logger; Factory( - Vertx vertx, + ExecutorService executor, Gson gson, long httpTimeout, WebClient webClient, CredentialsManager credentialsManager, Logger logger) { - this.vertx = vertx; + this.executor = executor; this.gson = gson; this.httpTimeout = httpTimeout; this.webClient = webClient; @@ -270,7 +269,7 @@ static class Factory { AgentClient create(URI agentUri) { return new AgentClient( - vertx, gson, httpTimeout, webClient, credentialsManager, agentUri, logger); + executor, gson, httpTimeout, webClient, credentialsManager, agentUri, logger); } } diff --git a/src/main/java/io/cryostat/net/NetworkModule.java b/src/main/java/io/cryostat/net/NetworkModule.java index d7cffcc02d..cd4332b6ae 100644 --- a/src/main/java/io/cryostat/net/NetworkModule.java +++ b/src/main/java/io/cryostat/net/NetworkModule.java @@ -19,7 +19,7 @@ import java.net.UnknownHostException; import java.nio.file.Path; import java.time.Duration; -import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.Executors; import javax.inject.Named; import javax.inject.Singleton; @@ -104,14 +104,18 @@ static AgentConnection.Factory provideAgentConnectionFactory( @Provides @Singleton static AgentClient.Factory provideAgentClientFactory( - Vertx vertx, Gson gson, @Named(HttpModule.HTTP_REQUEST_TIMEOUT_SECONDS) long httpTimeout, WebClient webClient, CredentialsManager credentialsManager, Logger logger) { return new AgentClient.Factory( - vertx, gson, httpTimeout, webClient, credentialsManager, logger); + Executors.newCachedThreadPool(), + gson, + httpTimeout, + webClient, + credentialsManager, + logger); } @Provides @@ -127,7 +131,7 @@ static TargetConnectionManager provideTargetConnectionManager( connectionToolkit, agentConnectionFactory, storage, - ForkJoinPool.commonPool(), + Executors.newCachedThreadPool(), Scheduler.systemScheduler(), maxTargetTtl, maxTargetConnections, @@ -144,7 +148,11 @@ static JFRConnectionToolkit provideJFRConnectionToolkit( @Provides @Singleton static Vertx provideVertx() { - return Vertx.vertx(new VertxOptions().setPreferNativeTransport(true)); + VertxOptions defaults = new VertxOptions(); + return Vertx.vertx( + new VertxOptions() + .setPreferNativeTransport(true) + .setEventLoopPoolSize(defaults.getEventLoopPoolSize() + 4)); } @Provides diff --git a/src/main/java/io/cryostat/platform/internal/DockerPlatformStrategy.java b/src/main/java/io/cryostat/platform/internal/DockerPlatformStrategy.java index d98be1358c..d96f0a378f 100644 --- a/src/main/java/io/cryostat/platform/internal/DockerPlatformStrategy.java +++ b/src/main/java/io/cryostat/platform/internal/DockerPlatformStrategy.java @@ -18,7 +18,6 @@ import java.net.URI; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; -import java.util.concurrent.ForkJoinPool; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -94,8 +93,7 @@ public boolean isAvailable() { private boolean testDockerApi() { CompletableFuture result = new CompletableFuture<>(); URI requestPath = URI.create("http://d/v1.41/info"); - ForkJoinPool.commonPool() - .submit( + new Thread( () -> { webClient .get() @@ -117,7 +115,8 @@ private boolean testDockerApi() { } result.complete(true); }); - }); + }) + .start(); try { return result.get(2, TimeUnit.SECONDS); } catch (InterruptedException | TimeoutException | ExecutionException e) { diff --git a/src/main/java/io/cryostat/platform/internal/PodmanPlatformStrategy.java b/src/main/java/io/cryostat/platform/internal/PodmanPlatformStrategy.java index 2d2f16caac..c7f60d8a8c 100644 --- a/src/main/java/io/cryostat/platform/internal/PodmanPlatformStrategy.java +++ b/src/main/java/io/cryostat/platform/internal/PodmanPlatformStrategy.java @@ -18,7 +18,7 @@ import java.net.URI; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; -import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -94,8 +94,7 @@ public boolean isAvailable() { private boolean testPodmanApi() { CompletableFuture result = new CompletableFuture<>(); URI requestPath = URI.create("http://d/info"); - ForkJoinPool.commonPool() - .submit( + new Thread( () -> { webClient .get() @@ -117,7 +116,8 @@ private boolean testPodmanApi() { } result.complete(true); }); - }); + }) + .start(); try { return result.get(2, TimeUnit.SECONDS); } catch (InterruptedException | TimeoutException | ExecutionException e) { @@ -130,7 +130,7 @@ private boolean testPodmanApi() { public PodmanPlatformClient getPlatformClient() { logger.info("Selected {} Strategy", getClass().getSimpleName()); return new PodmanPlatformClient( - ForkJoinPool.commonPool(), + Executors.newSingleThreadExecutor(), webClient, vertx, getSocket(), diff --git a/src/main/java/io/cryostat/recordings/RecordingsModule.java b/src/main/java/io/cryostat/recordings/RecordingsModule.java index 3851c06678..9f1fcbb598 100644 --- a/src/main/java/io/cryostat/recordings/RecordingsModule.java +++ b/src/main/java/io/cryostat/recordings/RecordingsModule.java @@ -21,7 +21,7 @@ import java.nio.file.attribute.PosixFilePermission; import java.nio.file.attribute.PosixFilePermissions; import java.util.Set; -import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.Executors; import javax.inject.Named; import javax.inject.Provider; @@ -177,7 +177,7 @@ static RecordingMetadataManager provideRecordingMetadataManager( PosixFilePermission.OWNER_EXECUTE))); } return new RecordingMetadataManager( - ForkJoinPool.commonPool(), + Executors.newSingleThreadExecutor(), metadataDir, archivedRecordingsPath, connectionTimeoutSeconds, @@ -210,7 +210,7 @@ static JvmIdHelper provideJvmIdHelper( credentialsManager, storage, connectionTimeoutSeconds, - ForkJoinPool.commonPool(), + Executors.newCachedThreadPool(), Scheduler.systemScheduler(), base32, logger); diff --git a/src/test/java/io/cryostat/discovery/DiscoveryStorageTest.java b/src/test/java/io/cryostat/discovery/DiscoveryStorageTest.java index c64b417b7c..703b9f6c15 100644 --- a/src/test/java/io/cryostat/discovery/DiscoveryStorageTest.java +++ b/src/test/java/io/cryostat/discovery/DiscoveryStorageTest.java @@ -28,6 +28,7 @@ import javax.inject.Singleton; +import io.cryostat.DirectExecutorService; import io.cryostat.MainModule; import io.cryostat.MockVertx; import io.cryostat.VerticleDeployer; @@ -106,6 +107,7 @@ void setup() { this.storage = new DiscoveryStorage( deployer, + new DirectExecutorService(), Duration.ofMinutes(5), () -> builtin, dao,