diff --git a/maestro-engine/src/main/java/com/netflix/maestro/engine/concurrency/EngineExecutors.java b/maestro-engine/src/main/java/com/netflix/maestro/engine/concurrency/EngineExecutors.java new file mode 100644 index 00000000..ff611bdf --- /dev/null +++ b/maestro-engine/src/main/java/com/netflix/maestro/engine/concurrency/EngineExecutors.java @@ -0,0 +1,74 @@ +/* + * Copyright 2024 Netflix, Inc. + * + * Licensed 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 + * + * http://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.netflix.maestro.engine.concurrency; + +import com.netflix.maestro.engine.properties.ThreadingModel; +import com.netflix.maestro.engine.properties.ThreadingProperties; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.atomic.AtomicLong; +import lombok.extern.slf4j.Slf4j; + +/** + * Factory for engine I/O executors. + * + *

Selects between Java 21 virtual threads and a bounded platform-thread pool based on {@link + * ThreadingProperties}. Virtual threads lift the previous hard cap on concurrent in-flight JDBC and + * signal lookups while preserving a one-line rollback path. + */ +@Slf4j +public final class EngineExecutors { + private EngineExecutors() {} + + /** + * Build a fresh I/O-bound executor based on the supplied threading properties. + * + *

Callers own the returned executor's lifecycle; call {@link ExecutorService#shutdown()} on + * application stop. + * + * @param props the threading properties; must not be null + * @return a new executor service + */ + public static ExecutorService newIoExecutor(ThreadingProperties props) { + ThreadingModel model = props.getModel() == null ? ThreadingModel.VIRTUAL : props.getModel(); + if (model == ThreadingModel.VIRTUAL) { + String prefix = + props.getVirtualThreadNamePrefix() == null + ? "maestro-vt-" + : props.getVirtualThreadNamePrefix(); + LOG.info("Creating virtual-thread I/O executor (name-prefix={})", prefix); + return Executors.newThreadPerTaskExecutor(virtualThreadFactory(prefix)); + } + int size = Math.max(1, props.getPlatformThreadPoolSize()); + LOG.info("Creating platform-thread I/O executor (size={})", size); + return Executors.newFixedThreadPool(size, daemonThreadFactory("maestro-io-")); + } + + private static ThreadFactory virtualThreadFactory(String prefix) { + AtomicLong seq = new AtomicLong(); + return runnable -> { + Thread t = Thread.ofVirtual().name(prefix + seq.incrementAndGet()).unstarted(runnable); + return t; + }; + } + + private static ThreadFactory daemonThreadFactory(String prefix) { + AtomicLong seq = new AtomicLong(); + return runnable -> { + Thread t = new Thread(runnable, prefix + seq.incrementAndGet()); + t.setDaemon(true); + return t; + }; + } +} diff --git a/maestro-engine/src/main/java/com/netflix/maestro/engine/eval/MaestroParamExtensionRepo.java b/maestro-engine/src/main/java/com/netflix/maestro/engine/eval/MaestroParamExtensionRepo.java index 2f244aca..af10c65f 100644 --- a/maestro-engine/src/main/java/com/netflix/maestro/engine/eval/MaestroParamExtensionRepo.java +++ b/maestro-engine/src/main/java/com/netflix/maestro/engine/eval/MaestroParamExtensionRepo.java @@ -15,34 +15,59 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import com.netflix.maestro.annotations.Nullable; +import com.netflix.maestro.engine.concurrency.EngineExecutors; import com.netflix.maestro.engine.dao.MaestroStepInstanceDao; import com.netflix.maestro.engine.handlers.SignalHandler; +import com.netflix.maestro.engine.properties.ThreadingProperties; import com.netflix.maestro.exceptions.MaestroUnprocessableEntityException; import com.netflix.maestro.utils.Checks; import com.netflix.sel.ext.Extension; import com.netflix.sel.type.SelUtilFunc; import java.util.Map; import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; import lombok.extern.slf4j.Slf4j; -/** A repository to hold maestro param extensions for the param evaluation. */ +/** + * A repository to hold maestro param extensions for the param evaluation. + * + *

Owns the I/O-bound executor that backs {@link MaestroParamExtension}'s blocking JDBC and + * signal lookups. The executor is constructed by {@link EngineExecutors} during {@link + * #initialize()} according to the supplied {@link ThreadingProperties}, and is shut down on + * {@link #shutdown()}. + */ @Slf4j public class MaestroParamExtensionRepo { - private static final int THREAD_NUM = 3; private final ThreadLocal repos = new ThreadLocal<>(); private final MaestroStepInstanceDao stepInstanceDao; private final ObjectMapper objectMapper; private final String env; - private ExecutorService executor; + private final ThreadingProperties threadingProperties; + private ExecutorService ioExecutor; - /** Constructor. */ + /** Constructor with default threading properties (virtual threads). */ public MaestroParamExtensionRepo( MaestroStepInstanceDao stepInstanceDao, String env, ObjectMapper objectMapper) { + this(stepInstanceDao, env, objectMapper, new ThreadingProperties()); + } + + /** + * Constructor. + * + * @param stepInstanceDao DAO used by the param extension for JDBC lookups + * @param env execution environment identifier + * @param objectMapper Jackson object mapper used to deserialize runtime summaries + * @param threadingProperties controls the executor model (virtual vs. platform) and tuning + */ + public MaestroParamExtensionRepo( + MaestroStepInstanceDao stepInstanceDao, + String env, + ObjectMapper objectMapper, + ThreadingProperties threadingProperties) { this.stepInstanceDao = stepInstanceDao; this.objectMapper = objectMapper; this.env = env; + this.threadingProperties = threadingProperties; } /** Reset repo by creating a new param extension wrapper for the current thread. */ @@ -52,7 +77,7 @@ public void reset( InstanceWrapper instanceWrapper) { Extension ext = new MaestroParamExtension( - executor, + ioExecutor, stepInstanceDao, env, allStepOutputData, @@ -72,20 +97,30 @@ public Extension get() { return repos.get(); } - /** Initialize the ExtensionRepo. */ + /** Initialize the ExtensionRepo and its I/O executor. */ void initialize() { LOG.info("Initializing ExtensionRepo within Spring boot..."); SelUtilFunc.register("toJson", this::toJsonExtFunction); - executor = Executors.newFixedThreadPool(THREAD_NUM); - ((ThreadPoolExecutor) executor).prestartAllCoreThreads(); + ioExecutor = EngineExecutors.newIoExecutor(threadingProperties); } - /** Gracefully shutdown the ExtensionRepo. */ - @SuppressWarnings({"PMD.NullAssignment"}) + /** Gracefully shutdown the ExtensionRepo and its I/O executor. */ void shutdown() { LOG.info("Shutdown ExtensionRepo within Spring boot..."); - executor.shutdown(); - executor = null; + ExecutorService toShutdown = ioExecutor; + ioExecutor = null; + if (toShutdown == null) { + return; + } + toShutdown.shutdown(); + try { + if (!toShutdown.awaitTermination(5, TimeUnit.SECONDS)) { + toShutdown.shutdownNow(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + toShutdown.shutdownNow(); + } } // Add a SEL function to convert the input object to a JSON string. If there are more, will diff --git a/maestro-engine/src/main/java/com/netflix/maestro/engine/properties/ThreadingModel.java b/maestro-engine/src/main/java/com/netflix/maestro/engine/properties/ThreadingModel.java new file mode 100644 index 00000000..102c2480 --- /dev/null +++ b/maestro-engine/src/main/java/com/netflix/maestro/engine/properties/ThreadingModel.java @@ -0,0 +1,31 @@ +/* + * Copyright 2024 Netflix, Inc. + * + * Licensed 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 + * + * http://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.netflix.maestro.engine.properties; + +/** + * Threading model selection for engine I/O executors. + * + *

Used to gate a safe, incremental rollout of Java 21 virtual threads. {@link #VIRTUAL} is the + * recommended default; {@link #PLATFORM} is preserved as a rollback target for environments where + * virtual threads are not yet desirable. + */ +public enum ThreadingModel { + /** Java 21 virtual threads via {@link java.util.concurrent.Executors#newThreadPerTaskExecutor}. */ + VIRTUAL, + + /** + * Legacy bounded platform-thread pool, retained only for safe rollback. Hard-caps the number of + * concurrent in-flight I/O calls and is therefore a throughput ceiling. + */ + PLATFORM +} diff --git a/maestro-engine/src/main/java/com/netflix/maestro/engine/properties/ThreadingProperties.java b/maestro-engine/src/main/java/com/netflix/maestro/engine/properties/ThreadingProperties.java new file mode 100644 index 00000000..9c88bcd1 --- /dev/null +++ b/maestro-engine/src/main/java/com/netflix/maestro/engine/properties/ThreadingProperties.java @@ -0,0 +1,36 @@ +/* + * Copyright 2024 Netflix, Inc. + * + * Licensed 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 + * + * http://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.netflix.maestro.engine.properties; + +import lombok.Getter; +import lombok.Setter; + +/** + * Threading model configuration for engine I/O executors. + * + *

Bound from {@code maestro.threading.*} in {@code application.yml}. Defaults to virtual threads; + * flip to {@link ThreadingModel#PLATFORM} for safe rollback if a virtual-thread regression is + * observed. + */ +@Getter +@Setter +public class ThreadingProperties { + /** Active threading model. Defaults to virtual threads. */ + private ThreadingModel model = ThreadingModel.VIRTUAL; + + /** Size of the platform-thread fallback pool. Ignored when {@link #model} is {@code VIRTUAL}. */ + private int platformThreadPoolSize = 3; + + /** Name prefix for virtual threads spawned by the engine. Useful for observability. */ + private String virtualThreadNamePrefix = "maestro-vt-"; +} diff --git a/maestro-engine/src/test/java/com/netflix/maestro/engine/concurrency/EngineExecutorsTest.java b/maestro-engine/src/test/java/com/netflix/maestro/engine/concurrency/EngineExecutorsTest.java new file mode 100644 index 00000000..b9b76a59 --- /dev/null +++ b/maestro-engine/src/test/java/com/netflix/maestro/engine/concurrency/EngineExecutorsTest.java @@ -0,0 +1,107 @@ +/* + * Copyright 2024 Netflix, Inc. + * + * Licensed 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 + * + * http://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.netflix.maestro.engine.concurrency; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.netflix.maestro.engine.properties.ThreadingModel; +import com.netflix.maestro.engine.properties.ThreadingProperties; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import org.junit.Test; + +public class EngineExecutorsTest { + + private static final int CONCURRENT_TASKS = 16; + private static final long PER_TASK_SLEEP_MILLIS = 100; + + @Test + public void virtualThreadsRunIOTasksConcurrently() throws Exception { + ThreadingProperties props = new ThreadingProperties(); + props.setModel(ThreadingModel.VIRTUAL); + + ExecutorService executor = EngineExecutors.newIoExecutor(props); + try { + long elapsed = runConcurrentSleepTasks(executor); + // With virtual threads, all 16 tasks should run concurrently. The total + // wall-clock time should be much less than the serialized 16 * 100 ms = 1600 ms + // lower bound. A generous bound accommodates JIT and scheduling jitter. + assertThat(elapsed) + .as("virtual threads should not serialize independent I/O-bound tasks") + .isLessThan(CONCURRENT_TASKS * PER_TASK_SLEEP_MILLIS / 2L); + } finally { + executor.shutdownNow(); + } + } + + @Test + public void platformThreadPoolCapsConcurrencyAtConfiguredSize() throws Exception { + ThreadingProperties props = new ThreadingProperties(); + props.setModel(ThreadingModel.PLATFORM); + props.setPlatformThreadPoolSize(2); + + ExecutorService executor = EngineExecutors.newIoExecutor(props); + try { + long elapsed = runConcurrentSleepTasks(executor); + // With a 2-thread pool and 16 tasks each sleeping 100 ms, the total wall-clock + // time must be at least 16/2 * 100 = 800 ms. We allow some headroom. + long lowerBound = + ((long) Math.ceil((double) CONCURRENT_TASKS / props.getPlatformThreadPoolSize())) + * PER_TASK_SLEEP_MILLIS; + assertThat(elapsed) + .as("bounded platform-thread pool must serialize tasks beyond pool size") + .isGreaterThanOrEqualTo(lowerBound - 50L); + } finally { + executor.shutdownNow(); + } + } + + @Test + public void virtualThreadExecutorUsesVirtualThreads() throws Exception { + ThreadingProperties props = new ThreadingProperties(); + props.setModel(ThreadingModel.VIRTUAL); + + ExecutorService executor = EngineExecutors.newIoExecutor(props); + try { + Future isVirtual = executor.submit(() -> Thread.currentThread().isVirtual()); + assertThat(isVirtual.get(1, TimeUnit.SECONDS)).isTrue(); + } finally { + executor.shutdownNow(); + } + } + + private static long runConcurrentSleepTasks(ExecutorService executor) + throws InterruptedException, java.util.concurrent.ExecutionException, java.util.concurrent.TimeoutException { + CountDownLatch start = new CountDownLatch(1); + List> futures = new ArrayList<>(); + long startNanos = System.nanoTime(); + for (int i = 0; i < CONCURRENT_TASKS; i++) { + futures.add( + executor.submit( + () -> { + start.await(); + Thread.sleep(PER_TASK_SLEEP_MILLIS); + return Thread.currentThread().threadId(); + })); + } + start.countDown(); + for (Future f : futures) { + f.get(5, TimeUnit.SECONDS); + } + return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos); + } +} diff --git a/maestro-server/src/main/java/com/netflix/maestro/server/config/MaestroEngineConfiguration.java b/maestro-server/src/main/java/com/netflix/maestro/server/config/MaestroEngineConfiguration.java index e21ec335..6ec15b84 100644 --- a/maestro-server/src/main/java/com/netflix/maestro/server/config/MaestroEngineConfiguration.java +++ b/maestro-server/src/main/java/com/netflix/maestro/server/config/MaestroEngineConfiguration.java @@ -130,10 +130,14 @@ public ParamsManager paramsManager(DefaultParamManager defaultParamManager) { public MaestroParamExtensionRepo maestroParamExtensionRepo( MaestroStepInstanceDao stepInstanceDao, StepRuntimeProperties stepRuntimeProperties, + MaestroProperties maestroProperties, @Qualifier(Constants.MAESTRO_QUALIFIER) ObjectMapper objectMapper) { LOG.info("Creating Maestro MaestroParamExtensionRepo within Spring boot..."); return new MaestroParamExtensionRepo( - stepInstanceDao, stepRuntimeProperties.getEnv(), objectMapper); + stepInstanceDao, + stepRuntimeProperties.getEnv(), + objectMapper, + maestroProperties.getThreading()); } @Bean(initMethod = "postConstruct", destroyMethod = "preDestroy") diff --git a/maestro-server/src/main/java/com/netflix/maestro/server/properties/MaestroProperties.java b/maestro-server/src/main/java/com/netflix/maestro/server/properties/MaestroProperties.java index d3806b58..5444ae5b 100644 --- a/maestro-server/src/main/java/com/netflix/maestro/server/properties/MaestroProperties.java +++ b/maestro-server/src/main/java/com/netflix/maestro/server/properties/MaestroProperties.java @@ -14,6 +14,7 @@ import com.netflix.maestro.engine.properties.SelProperties; import com.netflix.maestro.engine.properties.StepActionProperties; +import com.netflix.maestro.engine.properties.ThreadingProperties; import com.netflix.maestro.models.Constants; import com.netflix.maestro.queue.properties.QueueProperties; import lombok.AllArgsConstructor; @@ -29,10 +30,16 @@ public class MaestroProperties { private final SelProperties sel; private final ParamEvaluatorProperties paramEvaluator; private final StepActionProperties stepAction; + private final ThreadingProperties threading; private final MaestroIdNameValidationProperties maestroIdNameValidation; /** Returns the param evaluator properties, defaulting to {@code __} separator if not set. */ public ParamEvaluatorProperties getParamEvaluator() { return paramEvaluator != null ? paramEvaluator : new ParamEvaluatorProperties(); } + + /** Returns the threading properties, defaulting to virtual threads if not set. */ + public ThreadingProperties getThreading() { + return threading != null ? threading : new ThreadingProperties(); + } } diff --git a/maestro-server/src/main/resources/application.yml b/maestro-server/src/main/resources/application.yml index 6ac8394a..22fe917b 100644 --- a/maestro-server/src/main/resources/application.yml +++ b/maestro-server/src/main/resources/application.yml @@ -50,6 +50,10 @@ maestro: step-action: action-timeout: 5000 check-interval: 500 + threading: + model: VIRTUAL + platform-thread-pool-size: 3 + virtual-thread-name-prefix: "maestro-vt-" engine: configs: