From 046a123d1aa7009500ab472e47122775853520f2 Mon Sep 17 00:00:00 2001 From: Paolo Di Tommaso Date: Sun, 17 Nov 2024 16:33:25 +0100 Subject: [PATCH 1/6] Migration to virtual trheads - phase 1 (#737) This commit introduces the use of BLOCKING tasks executors for controllers, PairingWebSocket and TowerConnector. When running over Java 21 (and later) the blocking tasks executor use virtual thread pool to carry out the tasks execution. Signed-off-by: Paolo Di Tommaso --- .../seqera/wave/controller/BuildController.groovy | 2 +- .../wave/controller/ContainerController.groovy | 6 +++--- .../wave/controller/InspectController.groovy | 2 +- .../wave/controller/MetricsController.groovy | 2 +- .../wave/controller/MirrorController.groovy | 2 +- .../controller/RegistryProxyController.groovy | 2 +- .../seqera/wave/controller/ScanController.groovy | 2 +- .../wave/controller/ServiceInfoController.groovy | 2 +- .../wave/controller/ValidateController.groovy | 2 +- .../seqera/wave/controller/ViewController.groovy | 2 +- .../io/seqera/wave/core/ContainerAugmenter.groovy | 3 +-- .../io/seqera/wave/http/HttpClientFactory.groovy | 15 +++++++++++---- .../data/stream/AbstractMessageStream.groovy | 4 ++++ .../io/seqera/wave/service/job/JobManager.groovy | 1 - .../pairing/socket/PairingWebSocket.groovy | 2 +- .../tower/client/connector/TowerConnector.groovy | 6 +++++- .../connector/WebSocketTowerConnector.groovy | 9 +-------- 17 files changed, 35 insertions(+), 29 deletions(-) diff --git a/src/main/groovy/io/seqera/wave/controller/BuildController.groovy b/src/main/groovy/io/seqera/wave/controller/BuildController.groovy index 06679fa4ae..f995490872 100644 --- a/src/main/groovy/io/seqera/wave/controller/BuildController.groovy +++ b/src/main/groovy/io/seqera/wave/controller/BuildController.groovy @@ -42,7 +42,7 @@ import jakarta.inject.Inject @Slf4j @CompileStatic @Controller("/") -@ExecuteOn(TaskExecutors.IO) +@ExecuteOn(TaskExecutors.BLOCKING) class BuildController { @Inject diff --git a/src/main/groovy/io/seqera/wave/controller/ContainerController.groovy b/src/main/groovy/io/seqera/wave/controller/ContainerController.groovy index 10f71ddb42..a65568c40d 100644 --- a/src/main/groovy/io/seqera/wave/controller/ContainerController.groovy +++ b/src/main/groovy/io/seqera/wave/controller/ContainerController.groovy @@ -103,7 +103,7 @@ import static java.util.concurrent.CompletableFuture.completedFuture @Slf4j @CompileStatic @Controller("/") -@ExecuteOn(TaskExecutors.IO) +@ExecuteOn(TaskExecutors.BLOCKING) class ContainerController { @Inject @@ -181,13 +181,13 @@ class ContainerController { @Deprecated @Post('/container-token') - @ExecuteOn(TaskExecutors.IO) + @ExecuteOn(TaskExecutors.BLOCKING) CompletableFuture> getToken(HttpRequest httpRequest, @Body SubmitContainerTokenRequest req) { return getContainerImpl(httpRequest, req, false) } @Post('/v1alpha2/container') - @ExecuteOn(TaskExecutors.IO) + @ExecuteOn(TaskExecutors.BLOCKING) CompletableFuture> getTokenV2(HttpRequest httpRequest, @Body SubmitContainerTokenRequest req) { return getContainerImpl(httpRequest, req, true) } diff --git a/src/main/groovy/io/seqera/wave/controller/InspectController.groovy b/src/main/groovy/io/seqera/wave/controller/InspectController.groovy index 35e50bc58a..6093d4dfb5 100644 --- a/src/main/groovy/io/seqera/wave/controller/InspectController.groovy +++ b/src/main/groovy/io/seqera/wave/controller/InspectController.groovy @@ -50,7 +50,7 @@ import static io.seqera.wave.util.ContainerHelper.patchPlatformEndpoint @Slf4j @CompileStatic @Controller("/") -@ExecuteOn(TaskExecutors.IO) +@ExecuteOn(TaskExecutors.BLOCKING) class InspectController { @Inject diff --git a/src/main/groovy/io/seqera/wave/controller/MetricsController.groovy b/src/main/groovy/io/seqera/wave/controller/MetricsController.groovy index a9ce6f89a3..f8008674e7 100644 --- a/src/main/groovy/io/seqera/wave/controller/MetricsController.groovy +++ b/src/main/groovy/io/seqera/wave/controller/MetricsController.groovy @@ -52,7 +52,7 @@ import static io.micronaut.http.HttpHeaders.WWW_AUTHENTICATE @Requires(property = 'wave.metrics.enabled', value = 'true') @Secured(SecurityRule.IS_AUTHENTICATED) @Controller -@ExecuteOn(TaskExecutors.IO) +@ExecuteOn(TaskExecutors.BLOCKING) class MetricsController { @Inject diff --git a/src/main/groovy/io/seqera/wave/controller/MirrorController.groovy b/src/main/groovy/io/seqera/wave/controller/MirrorController.groovy index 8aaf87a51d..c9c3bc8f7a 100644 --- a/src/main/groovy/io/seqera/wave/controller/MirrorController.groovy +++ b/src/main/groovy/io/seqera/wave/controller/MirrorController.groovy @@ -36,7 +36,7 @@ import jakarta.inject.Inject @Slf4j @CompileStatic @Controller("/") -@ExecuteOn(TaskExecutors.IO) +@ExecuteOn(TaskExecutors.BLOCKING) class MirrorController { @Inject diff --git a/src/main/groovy/io/seqera/wave/controller/RegistryProxyController.groovy b/src/main/groovy/io/seqera/wave/controller/RegistryProxyController.groovy index ec64165379..2637632997 100644 --- a/src/main/groovy/io/seqera/wave/controller/RegistryProxyController.groovy +++ b/src/main/groovy/io/seqera/wave/controller/RegistryProxyController.groovy @@ -67,7 +67,7 @@ import reactor.core.publisher.Mono @Slf4j @CompileStatic @Controller("/v2") -@ExecuteOn(TaskExecutors.IO) +@ExecuteOn(TaskExecutors.BLOCKING) class RegistryProxyController { @Inject diff --git a/src/main/groovy/io/seqera/wave/controller/ScanController.groovy b/src/main/groovy/io/seqera/wave/controller/ScanController.groovy index 230a8b4a91..43fc38a5da 100644 --- a/src/main/groovy/io/seqera/wave/controller/ScanController.groovy +++ b/src/main/groovy/io/seqera/wave/controller/ScanController.groovy @@ -39,7 +39,7 @@ import jakarta.inject.Inject @CompileStatic @Requires(bean = ContainerScanService) @Controller("/") -@ExecuteOn(TaskExecutors.IO) +@ExecuteOn(TaskExecutors.BLOCKING) class ScanController { @Inject diff --git a/src/main/groovy/io/seqera/wave/controller/ServiceInfoController.groovy b/src/main/groovy/io/seqera/wave/controller/ServiceInfoController.groovy index 207c6e8d5e..7f0a895e43 100644 --- a/src/main/groovy/io/seqera/wave/controller/ServiceInfoController.groovy +++ b/src/main/groovy/io/seqera/wave/controller/ServiceInfoController.groovy @@ -39,7 +39,7 @@ import io.seqera.wave.util.BuildInfo @Slf4j @Controller("/") @CompileStatic -@ExecuteOn(TaskExecutors.IO) +@ExecuteOn(TaskExecutors.BLOCKING) class ServiceInfoController { @Value('${wave.landing.url}') diff --git a/src/main/groovy/io/seqera/wave/controller/ValidateController.groovy b/src/main/groovy/io/seqera/wave/controller/ValidateController.groovy index f7136f5725..fd23fbdc6d 100644 --- a/src/main/groovy/io/seqera/wave/controller/ValidateController.groovy +++ b/src/main/groovy/io/seqera/wave/controller/ValidateController.groovy @@ -26,8 +26,8 @@ import io.seqera.wave.auth.RegistryAuthService import jakarta.inject.Inject import jakarta.validation.Valid -@ExecuteOn(TaskExecutors.IO) @Controller("/validate-creds") +@ExecuteOn(TaskExecutors.BLOCKING) class ValidateController { @Inject RegistryAuthService loginService diff --git a/src/main/groovy/io/seqera/wave/controller/ViewController.groovy b/src/main/groovy/io/seqera/wave/controller/ViewController.groovy index d30db91d89..68328eb36f 100644 --- a/src/main/groovy/io/seqera/wave/controller/ViewController.groovy +++ b/src/main/groovy/io/seqera/wave/controller/ViewController.groovy @@ -59,7 +59,7 @@ import static io.seqera.wave.util.DataTimeUtils.formatTimestamp @Slf4j @CompileStatic @Controller("/view") -@ExecuteOn(TaskExecutors.IO) +@ExecuteOn(TaskExecutors.BLOCKING) class ViewController { @Inject diff --git a/src/main/groovy/io/seqera/wave/core/ContainerAugmenter.groovy b/src/main/groovy/io/seqera/wave/core/ContainerAugmenter.groovy index 909cc8d5d4..bd3837e87f 100644 --- a/src/main/groovy/io/seqera/wave/core/ContainerAugmenter.groovy +++ b/src/main/groovy/io/seqera/wave/core/ContainerAugmenter.groovy @@ -280,7 +280,7 @@ class ContainerAugmenter { return result } - synchronized protected Map layerBlob(String image, ContainerLayer layer) { + protected Map layerBlob(String image, ContainerLayer layer) { log.debug "Adding layer: $layer to image: $client.registry.name/$image" // store the layer blob in the cache final String path = "$client.registry.name/v2/$image/blobs/$layer.gzipDigest" @@ -295,7 +295,6 @@ class ContainerAugmenter { protected Tuple2 updateImageManifest(String imageName, String imageManifest, String newImageConfigDigest, newImageConfigSize, boolean oci) { - // turn the json string into a json map // and append the new layer final manifest = (Map) new JsonSlurper().parseText(imageManifest) diff --git a/src/main/groovy/io/seqera/wave/http/HttpClientFactory.groovy b/src/main/groovy/io/seqera/wave/http/HttpClientFactory.groovy index f1301037e3..46c9730d24 100644 --- a/src/main/groovy/io/seqera/wave/http/HttpClientFactory.groovy +++ b/src/main/groovy/io/seqera/wave/http/HttpClientFactory.groovy @@ -22,6 +22,7 @@ import java.net.http.HttpClient import java.time.Duration import java.util.concurrent.ExecutorService import java.util.concurrent.Executors +import java.util.concurrent.locks.ReentrantLock import groovy.transform.CompileStatic import groovy.util.logging.Slf4j @@ -39,9 +40,9 @@ class HttpClientFactory { static private Duration timeout = Duration.ofSeconds(20) - static private final Object l1 = new Object() + static private final ReentrantLock l1 = new ReentrantLock() - static private final Object l2 = new Object() + static private final ReentrantLock l2 = new ReentrantLock() private static HttpClient client1 @@ -51,20 +52,26 @@ class HttpClientFactory { static HttpClient followRedirectsHttpClient() { if( client1!=null ) return client1 - synchronized (l1) { + l1.lock() + try { if( client1!=null ) return client1 return client1=followRedirectsHttpClient0() + } finally { + l1.unlock() } } static HttpClient neverRedirectsHttpClient() { if( client2!=null ) return client2 - synchronized (l2) { + l2.lock() + try { if( client2!=null ) return client2 return client2=neverRedirectsHttpClient0() + } finally { + l2.unlock() } } diff --git a/src/main/groovy/io/seqera/wave/service/data/stream/AbstractMessageStream.groovy b/src/main/groovy/io/seqera/wave/service/data/stream/AbstractMessageStream.groovy index 560a813060..ae16d7140b 100644 --- a/src/main/groovy/io/seqera/wave/service/data/stream/AbstractMessageStream.groovy +++ b/src/main/groovy/io/seqera/wave/service/data/stream/AbstractMessageStream.groovy @@ -98,6 +98,10 @@ abstract class AbstractMessageStream implements Closeable { * The {@link Predicate} to be invoked when a stream message is consumed (read from) the stream. */ void addConsumer(String streamId, MessageConsumer consumer) { + // the use of synchronized block is meant to prevent a race condition while + // updating the 'listeners' from concurrent invocations. + // however, considering the addConsumer is invoked during the initialization phase + // (and therefore in the same thread) in should not be really needed. synchronized (listeners) { if( listeners.containsKey(streamId)) throw new IllegalStateException("Only one consumer can be defined for each stream - offending streamId=$streamId; consumer=$consumer") diff --git a/src/main/groovy/io/seqera/wave/service/job/JobManager.groovy b/src/main/groovy/io/seqera/wave/service/job/JobManager.groovy index 1c4ec6c036..1127859f62 100644 --- a/src/main/groovy/io/seqera/wave/service/job/JobManager.groovy +++ b/src/main/groovy/io/seqera/wave/service/job/JobManager.groovy @@ -59,7 +59,6 @@ class JobManager { queue.addConsumer((job)-> processJob(job)) } - protected boolean processJob(JobSpec jobSpec) { try { return processJob0(jobSpec) diff --git a/src/main/groovy/io/seqera/wave/service/pairing/socket/PairingWebSocket.groovy b/src/main/groovy/io/seqera/wave/service/pairing/socket/PairingWebSocket.groovy index a4f7b07859..3e6682fa51 100644 --- a/src/main/groovy/io/seqera/wave/service/pairing/socket/PairingWebSocket.groovy +++ b/src/main/groovy/io/seqera/wave/service/pairing/socket/PairingWebSocket.groovy @@ -49,7 +49,7 @@ import static io.seqera.wave.util.LongRndKey.rndHex @Slf4j @CompileStatic @Singleton -@ExecuteOn(TaskExecutors.IO) +@ExecuteOn(TaskExecutors.BLOCKING) @ServerWebSocket("/pairing/{service}/token/{token}{?endpoint}") class PairingWebSocket { diff --git a/src/main/groovy/io/seqera/wave/tower/client/connector/TowerConnector.groovy b/src/main/groovy/io/seqera/wave/tower/client/connector/TowerConnector.groovy index ec213628a7..f9872fc3c5 100644 --- a/src/main/groovy/io/seqera/wave/tower/client/connector/TowerConnector.groovy +++ b/src/main/groovy/io/seqera/wave/tower/client/connector/TowerConnector.groovy @@ -82,7 +82,7 @@ abstract class TowerConnector { private SpillwayRateLimiter limiter @Inject - @Named(TaskExecutors.IO) + @Named(TaskExecutors.BLOCKING) private volatile ExecutorService ioExecutor private CacheLoader> loader = new CacheLoader>() { @@ -102,6 +102,10 @@ abstract class TowerConnector { return refreshCache } + protected ExecutorService getIoExecutor() { + return ioExecutor + } + /** * Generic async get with authorization * that converts to the provided json model T diff --git a/src/main/groovy/io/seqera/wave/tower/client/connector/WebSocketTowerConnector.groovy b/src/main/groovy/io/seqera/wave/tower/client/connector/WebSocketTowerConnector.groovy index c52646638f..04c591a099 100644 --- a/src/main/groovy/io/seqera/wave/tower/client/connector/WebSocketTowerConnector.groovy +++ b/src/main/groovy/io/seqera/wave/tower/client/connector/WebSocketTowerConnector.groovy @@ -19,19 +19,16 @@ package io.seqera.wave.tower.client.connector import java.util.concurrent.CompletableFuture -import java.util.concurrent.ExecutorService import java.util.function.Function import groovy.transform.CompileStatic import groovy.util.logging.Slf4j import io.micronaut.context.annotation.Requires -import io.micronaut.scheduling.TaskExecutors import io.seqera.wave.service.pairing.socket.PairingChannel import io.seqera.wave.service.pairing.socket.msg.PairingMessage import io.seqera.wave.service.pairing.socket.msg.ProxyHttpRequest import io.seqera.wave.service.pairing.socket.msg.ProxyHttpResponse import jakarta.inject.Inject -import jakarta.inject.Named import jakarta.inject.Singleton import static io.seqera.wave.service.pairing.PairingService.TOWER_SERVICE /** @@ -49,15 +46,11 @@ class WebSocketTowerConnector extends TowerConnector { @Inject private PairingChannel channel - @Inject - @Named(TaskExecutors.IO) - private ExecutorService ioExecutor - @Override CompletableFuture sendAsync(String endpoint, ProxyHttpRequest request) { return channel .sendRequest(TOWER_SERVICE, endpoint, request) - .thenApplyAsync(Function.identity() as Function, ioExecutor) + .thenApplyAsync(Function.identity() as Function, getIoExecutor()) } } From 7773d615c97a303b617feeda82ce04e97b90a7ca Mon Sep 17 00:00:00 2001 From: Paolo Di Tommaso Date: Sun, 17 Nov 2024 17:03:42 +0100 Subject: [PATCH 2/6] Migration to virtual threads - phase 2 Signed-off-by: Paolo Di Tommaso --- .../io/seqera/wave/filter/PullMetricsRequestsFilter.groovy | 2 +- .../seqera/wave/service/builder/impl/BuildStateStoreImpl.groovy | 2 +- .../wave/service/builder/impl/ContainerBuildServiceImpl.groovy | 2 +- .../io/seqera/wave/service/logs/BuildLogServiceImpl.groovy | 2 +- .../wave/service/mirror/ContainerMirrorServiceImpl.groovy | 2 +- .../io/seqera/wave/service/scan/ContainerScanServiceImpl.groovy | 2 +- .../io/seqera/wave/service/builder/BuildStoreLocalTest.groovy | 2 +- 7 files changed, 7 insertions(+), 7 deletions(-) diff --git a/src/main/groovy/io/seqera/wave/filter/PullMetricsRequestsFilter.groovy b/src/main/groovy/io/seqera/wave/filter/PullMetricsRequestsFilter.groovy index da19fc7c68..e45afbd549 100644 --- a/src/main/groovy/io/seqera/wave/filter/PullMetricsRequestsFilter.groovy +++ b/src/main/groovy/io/seqera/wave/filter/PullMetricsRequestsFilter.groovy @@ -68,7 +68,7 @@ class PullMetricsRequestsFilter implements HttpServerFilter { private RouteHandler routeHelper @Inject - @Named(TaskExecutors.IO) + @Named(TaskExecutors.BLOCKING) private ExecutorService executor @Override diff --git a/src/main/groovy/io/seqera/wave/service/builder/impl/BuildStateStoreImpl.groovy b/src/main/groovy/io/seqera/wave/service/builder/impl/BuildStateStoreImpl.groovy index d9562ce43d..963629aaaf 100644 --- a/src/main/groovy/io/seqera/wave/service/builder/impl/BuildStateStoreImpl.groovy +++ b/src/main/groovy/io/seqera/wave/service/builder/impl/BuildStateStoreImpl.groovy @@ -50,7 +50,7 @@ class BuildStateStoreImpl extends AbstractStateStore implements Buil private ExecutorService ioExecutor - BuildStateStoreImpl(StateProvider provider, BuildConfig buildConfig, @Named(TaskExecutors.IO) ExecutorService ioExecutor) { + BuildStateStoreImpl(StateProvider provider, BuildConfig buildConfig, @Named(TaskExecutors.BLOCKING) ExecutorService ioExecutor) { super(provider, new MoshiEncodeStrategy() {}) this.buildConfig = buildConfig this.ioExecutor = ioExecutor diff --git a/src/main/groovy/io/seqera/wave/service/builder/impl/ContainerBuildServiceImpl.groovy b/src/main/groovy/io/seqera/wave/service/builder/impl/ContainerBuildServiceImpl.groovy index 6dbe9f7878..1ab3df78e4 100644 --- a/src/main/groovy/io/seqera/wave/service/builder/impl/ContainerBuildServiceImpl.groovy +++ b/src/main/groovy/io/seqera/wave/service/builder/impl/ContainerBuildServiceImpl.groovy @@ -87,7 +87,7 @@ class ContainerBuildServiceImpl implements ContainerBuildService, JobHandler Date: Sun, 17 Nov 2024 17:29:09 +0100 Subject: [PATCH 3/6] Add jdk.tracePinnedThreads=short to default run options [ci skip] Signed-off-by: Paolo Di Tommaso --- build.gradle | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/build.gradle b/build.gradle index 7de40d2379..ff02cb7cb3 100644 --- a/build.gradle +++ b/build.gradle @@ -156,8 +156,7 @@ jib { run{ def envs = findProperty('micronautEnvs') - // note: "--enable-preview" is required to use virtual threads on Java 19 and 20 - def args = ["-Dmicronaut.environments=$envs","--enable-preview"] + def args = ["-Dmicronaut.environments=$envs","-Djdk.tracePinnedThreads=short"] if( environment['JVM_OPTS'] ) args.add(environment['JVM_OPTS']) jvmArgs args systemProperties 'DOCKER_USER': project.findProperty('DOCKER_USER') ?: environment['DOCKER_USER'], From 7352f79ccdf8537ada477fa6f9035d225dbfbc39 Mon Sep 17 00:00:00 2001 From: Paolo Di Tommaso Date: Sun, 17 Nov 2024 17:35:39 +0100 Subject: [PATCH 4/6] Remove deprecated ThreadPoolBuilder Signed-off-by: Paolo Di Tommaso --- .../seqera/wave/util/ThreadPoolBuilder.groovy | 176 ------------------ 1 file changed, 176 deletions(-) delete mode 100644 src/main/groovy/io/seqera/wave/util/ThreadPoolBuilder.groovy diff --git a/src/main/groovy/io/seqera/wave/util/ThreadPoolBuilder.groovy b/src/main/groovy/io/seqera/wave/util/ThreadPoolBuilder.groovy deleted file mode 100644 index 7afdf0e062..0000000000 --- a/src/main/groovy/io/seqera/wave/util/ThreadPoolBuilder.groovy +++ /dev/null @@ -1,176 +0,0 @@ -/* - * Wave, containers provisioning service - * Copyright (c) 2023-2024, Seqera Labs - * - * This program is free software: you can redistribute it and/or modify - * it under the terms of the GNU Affero General Public License as published by - * the Free Software Foundation, either version 3 of the License, or - * (at your option) any later version. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU Affero General Public License for more details. - * - * You should have received a copy of the GNU Affero General Public License - * along with this program. If not, see . - */ - -package io.seqera.wave.util - - - -import java.util.concurrent.BlockingQueue -import java.util.concurrent.LinkedBlockingQueue -import java.util.concurrent.RejectedExecutionHandler -import java.util.concurrent.ThreadFactory -import java.util.concurrent.ThreadPoolExecutor -import java.util.concurrent.TimeUnit -import java.util.concurrent.atomic.AtomicInteger - -import groovy.transform.CompileStatic -import groovy.util.logging.Slf4j -/** - * Builder class to create instance of {@link ThreadPoolExecutor} - * - * @author Paolo Di Tommaso - */ -@Slf4j -@CompileStatic -@Deprecated -class ThreadPoolBuilder { - - static AtomicInteger poolCount = new AtomicInteger() - - private String name - - private int minSize - - private int maxSize - - private BlockingQueue workQueue - - private int queueSize = -1 - - private Long keepAliveTime - - private RejectedExecutionHandler rejectionPolicy - - private ThreadFactory threadFactory - - private boolean allowCoreThreadTimeout - - String getName() { name } - - int getMinSize() { minSize } - - int getMaxSize() { maxSize } - - int getQueueSize() { queueSize } - - BlockingQueue getWorkQueue() { workQueue } - - Long getKeepAliveTime() { keepAliveTime } - - RejectedExecutionHandler getRejectionPolicy() { rejectionPolicy } - - ThreadFactory getThreadFactory() { threadFactory } - - boolean getAllowCoreThreadTimeout() { allowCoreThreadTimeout } - - ThreadPoolBuilder withName(String name) { - if( name ) { - this.name = name - this.threadFactory = new CustomThreadFactory(name) - } - return this - } - - ThreadPoolBuilder withThreadFactory(ThreadFactory threadFactory) { - assert !name || !threadFactory, "Property 'threadFactory' or 'name' was already set" - this.threadFactory = threadFactory - return this - } - - ThreadPoolBuilder withRejectionPolicy(RejectedExecutionHandler rejectionPolicy) { - this.rejectionPolicy = rejectionPolicy - return this - } - - ThreadPoolBuilder withMinSize(int min) { - this.minSize = min - return this - } - - ThreadPoolBuilder withMaxSize(int max) { - this.maxSize = max - return this - } - - ThreadPoolBuilder withQueueSize(int size) { - this.queueSize = size - this.workQueue = new LinkedBlockingQueue(size) - return this - } - - ThreadPoolBuilder withQueue(BlockingQueue workQueue) { - this.workQueue = workQueue - return this - } - - ThreadPoolBuilder withKeepAliveTime( long millis ) { - keepAliveTime = millis - return this - } - - ThreadPoolBuilder withAllowCoreThreadTimeout(boolean flag) { - this.allowCoreThreadTimeout = flag - return this - } - - ThreadPoolExecutor build() { - assert minSize <= maxSize - - if( !name ) - name = "nf-thread-pool-${poolCount.getAndIncrement()}" - - if(keepAliveTime==null) - keepAliveTime = 60_000 - if( workQueue==null ) - workQueue = new LinkedBlockingQueue<>() - if( rejectionPolicy==null ) - rejectionPolicy = new ThreadPoolExecutor.CallerRunsPolicy() - if( threadFactory==null ) - threadFactory = new CustomThreadFactory(name) - - log.debug "Creating thread pool '$name' minSize=$minSize; maxSize=$maxSize; workQueue=${workQueue.getClass().getSimpleName()}[${queueSize}]; allowCoreThreadTimeout=$allowCoreThreadTimeout" - - final result = new ThreadPoolExecutor( - minSize, - maxSize, - keepAliveTime, TimeUnit.MILLISECONDS, - workQueue, - threadFactory, - rejectionPolicy) - - result.allowCoreThreadTimeOut(allowCoreThreadTimeout) - - return result - } - - - static ThreadPoolExecutor io(String name=null) { - io(10, 100, 10_000, name) - } - - - static ThreadPoolExecutor io(int min, int max, int queue, String name=null) { - new ThreadPoolBuilder() - .withMinSize(min) - .withMaxSize(max) - .withQueueSize(queue) - .withName(name) - .build() - } - -} From 82a7d0a08adc324565d9511416381e15e4a4a983 Mon Sep 17 00:00:00 2001 From: Paolo Di Tommaso Date: Sun, 17 Nov 2024 18:10:40 +0100 Subject: [PATCH 5/6] Migration to virtual threads - phase 3 Signed-off-by: Paolo Di Tommaso --- .../io/seqera/wave/http/HttpClientFactory.groovy | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/src/main/groovy/io/seqera/wave/http/HttpClientFactory.groovy b/src/main/groovy/io/seqera/wave/http/HttpClientFactory.groovy index 46c9730d24..f0852219f1 100644 --- a/src/main/groovy/io/seqera/wave/http/HttpClientFactory.groovy +++ b/src/main/groovy/io/seqera/wave/http/HttpClientFactory.groovy @@ -20,13 +20,12 @@ package io.seqera.wave.http import java.net.http.HttpClient import java.time.Duration -import java.util.concurrent.ExecutorService import java.util.concurrent.Executors +import java.util.concurrent.ThreadFactory import java.util.concurrent.locks.ReentrantLock import groovy.transform.CompileStatic import groovy.util.logging.Slf4j -import io.seqera.wave.util.CustomThreadFactory /** * Java HttpClient factory * @@ -36,7 +35,9 @@ import io.seqera.wave.util.CustomThreadFactory @CompileStatic class HttpClientFactory { - static private ExecutorService threadPool = Executors.newCachedThreadPool(new CustomThreadFactory("HttpClientThread")) + static final private ThreadFactory customThreadFactory = Thread.ofVirtual() + .name("httpclient-virtual-thread-", 1) + .factory(); static private Duration timeout = Duration.ofSeconds(20) @@ -84,7 +85,7 @@ class HttpClientFactory { .version(HttpClient.Version.HTTP_1_1) .followRedirects(HttpClient.Redirect.NORMAL) .connectTimeout(timeout) - .executor(threadPool) + .executor(Executors.newThreadPerTaskExecutor(customThreadFactory)) .build() log.debug "Creating new followRedirectsHttpClient: $result" return result @@ -95,7 +96,7 @@ class HttpClientFactory { .version(HttpClient.Version.HTTP_1_1) .followRedirects(HttpClient.Redirect.NEVER) .connectTimeout(timeout) - .executor(threadPool) + .executor(Executors.newThreadPerTaskExecutor(customThreadFactory)) .build() log.debug "Creating new neverRedirectsHttpClient: $result" return result From c7ae5d1b446d9d487ed067f2fb2dc2dcc30d6472 Mon Sep 17 00:00:00 2001 From: Paolo Di Tommaso Date: Fri, 22 Nov 2024 22:47:52 +0100 Subject: [PATCH 6/6] Remove duplicate method Signed-off-by: Paolo Di Tommaso --- .../seqera/wave/tower/client/connector/TowerConnector.groovy | 4 ---- 1 file changed, 4 deletions(-) diff --git a/src/main/groovy/io/seqera/wave/tower/client/connector/TowerConnector.groovy b/src/main/groovy/io/seqera/wave/tower/client/connector/TowerConnector.groovy index 2f9eea9d07..27e12d87f1 100644 --- a/src/main/groovy/io/seqera/wave/tower/client/connector/TowerConnector.groovy +++ b/src/main/groovy/io/seqera/wave/tower/client/connector/TowerConnector.groovy @@ -106,10 +106,6 @@ abstract class TowerConnector { return ioExecutor } - protected ExecutorService getIoExecutor() { - return ioExecutor - } - /** * Generic async get with authorization * that converts to the provided json model T