From 74aebe038ff8c125591ec5d690b93ec6df671c87 Mon Sep 17 00:00:00 2001 From: slinkydeveloper Date: Tue, 7 Jul 2026 11:57:20 +0200 Subject: [PATCH] Add support for configurable event loops and enable multi-event-loop setup by default. In spring this is configurable in the server properties as restate.sdk.http.eventLoops In vanilla SDK this is configurable using the env variable RESTATE_SDK_HTTP_EVENT_LOOPS or sys property dev.restate.sdk.http.eventLoops --- .../sdk/http/vertx/RestateHttpServer.java | 93 ++++++++++++---- .../springboot/RestateHttpEndpointBean.java | 101 +++++++++++++++--- .../RestateHttpServerProperties.java | 13 ++- 3 files changed, 172 insertions(+), 35 deletions(-) diff --git a/sdk-http-vertx/src/main/java/dev/restate/sdk/http/vertx/RestateHttpServer.java b/sdk-http-vertx/src/main/java/dev/restate/sdk/http/vertx/RestateHttpServer.java index 33c412056..881085200 100644 --- a/sdk-http-vertx/src/main/java/dev/restate/sdk/http/vertx/RestateHttpServer.java +++ b/sdk-http-vertx/src/main/java/dev/restate/sdk/http/vertx/RestateHttpServer.java @@ -9,13 +9,16 @@ package dev.restate.sdk.http.vertx; import dev.restate.sdk.endpoint.Endpoint; -import io.vertx.core.Future; +import io.vertx.core.AbstractVerticle; +import io.vertx.core.DeploymentOptions; +import io.vertx.core.Promise; import io.vertx.core.Vertx; -import io.vertx.core.http.Http2Settings; +import io.vertx.core.VertxOptions; import io.vertx.core.http.HttpServer; import io.vertx.core.http.HttpServerOptions; import java.util.Optional; import java.util.concurrent.CompletionException; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -42,9 +45,8 @@ public class RestateHttpServer { private static final int DEFAULT_PORT = Optional.ofNullable(System.getenv("PORT")).map(Integer::parseInt).orElse(9080); - private static final HttpServerOptions DEFAULT_OPTIONS = - new HttpServerOptions() - .setInitialSettings(new Http2Settings().setMaxConcurrentStreams(Integer.MAX_VALUE)); + private static final HttpServerOptions DEFAULT_HTTP_SERVER_OPTIONS = new HttpServerOptions(); + private static final int DEFAULT_EVENT_LOOPS = VertxOptions.DEFAULT_EVENT_LOOP_POOL_SIZE; /** * Start serving the provided {@code endpoint} on the port specified by the environment variable @@ -54,10 +56,10 @@ public class RestateHttpServer { * non-blocking variant, manually create the server with {@link #fromEndpoint(Endpoint)} and start * listening it. * - * @return The listening port + * @return The listening port, use 0 for random port. */ public static int listen(Endpoint endpoint) { - return handleStart(fromEndpoint(endpoint).listen(DEFAULT_PORT)); + return listen(endpoint, DEFAULT_PORT); } /** Like {@link #listen(Endpoint)} */ @@ -72,10 +74,10 @@ public static int listen(Endpoint.Builder endpointBuilder) { * non-blocking variant, manually create the server with {@link #fromEndpoint(Endpoint)} and start * listening it. * - * @return The listening port + * @return The listening port, use 0 for random port. */ public static int listen(Endpoint endpoint, int port) { - return handleStart(fromEndpoint(endpoint).listen(port)); + return listen(HttpEndpointRequestHandler.fromEndpoint(endpoint), port); } /** Like {@link #listen(Endpoint, int)} */ @@ -90,12 +92,12 @@ public static int listen(HttpEndpointRequestHandler requestHandler) { /** Like {@link #listen(Endpoint, int)}, with an already built request handler */ public static int listen(HttpEndpointRequestHandler requestHandler, int port) { - return handleStart(fromHandler(requestHandler).listen(port)); + return listenBlocking(requestHandler, port); } /** Create a Vert.x {@link HttpServer} from the provided endpoint. */ public static HttpServer fromEndpoint(Endpoint endpoint) { - return fromEndpoint(endpoint, DEFAULT_OPTIONS); + return fromEndpoint(endpoint, DEFAULT_HTTP_SERVER_OPTIONS); } /** Like {@link #fromEndpoint(Endpoint)} */ @@ -119,7 +121,7 @@ public static HttpServer fromEndpoint( /** Create a Vert.x {@link HttpServer} from the provided endpoint. */ public static HttpServer fromEndpoint(Vertx vertx, Endpoint endpoint) { - return fromEndpoint(vertx, endpoint, DEFAULT_OPTIONS); + return fromEndpoint(vertx, endpoint, DEFAULT_HTTP_SERVER_OPTIONS); } /** Like {@link #fromEndpoint(Vertx, Endpoint)} */ @@ -143,7 +145,7 @@ public static HttpServer fromEndpoint( /** Create a Vert.x {@link HttpServer} from the provided {@link HttpEndpointRequestHandler}. */ public static HttpServer fromHandler(HttpEndpointRequestHandler handler) { - return fromHandler(handler, DEFAULT_OPTIONS); + return fromHandler(handler, DEFAULT_HTTP_SERVER_OPTIONS); } /** @@ -157,7 +159,7 @@ public static HttpServer fromHandler( /** Create a Vert.x {@link HttpServer} from the provided {@link HttpEndpointRequestHandler}. */ public static HttpServer fromHandler(Vertx vertx, HttpEndpointRequestHandler handler) { - return fromHandler(vertx, handler, DEFAULT_OPTIONS); + return fromHandler(vertx, handler, DEFAULT_HTTP_SERVER_OPTIONS); } /** @@ -171,19 +173,74 @@ public static HttpServer fromHandler( return server; } - private static int handleStart(Future fut) { + private static int listenBlocking(HttpEndpointRequestHandler handler, int port) { + String eventLoopsOverride = + System.getProperty( + "dev.restate.sdk.http.eventLoops", System.getenv("RESTATE_SDK_HTTP_EVENT_LOOPS")); + int instances = + eventLoopsOverride != null ? Integer.parseInt(eventLoopsOverride) : DEFAULT_EVENT_LOOPS; + Vertx vertx = Vertx.vertx(new VertxOptions().setEventLoopPoolSize(instances)); + // A negative port makes all the server instances share the same random port, whereas port 0 + // would make each instance bind a different random port. + int listenPort = port == 0 ? -1 : port; + AtomicInteger actualPort = new AtomicInteger(); try { - HttpServer server = fut.toCompletionStage().toCompletableFuture().join(); - LOG.info("Restate HTTP Endpoint server started on port {}", server.actualPort()); - return server.actualPort(); + vertx + .deployVerticle( + () -> + new RestateServerVerticle( + handler, DEFAULT_HTTP_SERVER_OPTIONS, listenPort, actualPort), + new DeploymentOptions().setInstances(instances)) + .toCompletionStage() + .toCompletableFuture() + .join(); + LOG.info( + "Restate HTTP Endpoint server started on port {} with {} event loop(s)", + actualPort.get(), + instances); + return actualPort.get(); } catch (CompletionException e) { LOG.error("Restate HTTP Endpoint server start failed", e.getCause()); + vertx.close(); sneakyThrow(e.getCause()); // This is never reached return -1; } } + private static final class RestateServerVerticle extends AbstractVerticle { + + private final HttpEndpointRequestHandler handler; + private final HttpServerOptions options; + private final int port; + private final AtomicInteger actualPort; + + private RestateServerVerticle( + HttpEndpointRequestHandler handler, + HttpServerOptions options, + int port, + AtomicInteger actualPort) { + this.handler = handler; + this.options = options; + this.port = port; + this.actualPort = actualPort; + } + + @Override + public void start(Promise startPromise) { + HttpServer server = vertx.createHttpServer(options); + server.requestHandler(handler); + server + .listen(port) + .onSuccess( + s -> { + actualPort.set(s.actualPort()); + startPromise.complete(); + }) + .onFailure(startPromise::fail); + } + } + @SuppressWarnings("unchecked") private static void sneakyThrow(Throwable e) throws E { throw (E) e; diff --git a/sdk-spring-boot/src/main/java/dev/restate/sdk/springboot/RestateHttpEndpointBean.java b/sdk-spring-boot/src/main/java/dev/restate/sdk/springboot/RestateHttpEndpointBean.java index 24e465d2a..6fb43faea 100644 --- a/sdk-spring-boot/src/main/java/dev/restate/sdk/springboot/RestateHttpEndpointBean.java +++ b/sdk-spring-boot/src/main/java/dev/restate/sdk/springboot/RestateHttpEndpointBean.java @@ -12,8 +12,14 @@ import dev.restate.sdk.endpoint.Endpoint; import dev.restate.sdk.http.vertx.HttpEndpointRequestHandler; import dev.restate.sdk.http.vertx.RestateHttpServer; -import io.vertx.core.http.HttpServer; +import io.vertx.core.AbstractVerticle; +import io.vertx.core.DeploymentOptions; +import io.vertx.core.Promise; +import io.vertx.core.Vertx; +import io.vertx.core.VertxOptions; import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import org.jspecify.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; @@ -29,15 +35,18 @@ public class RestateHttpEndpointBean implements InitializingBean, SmartLifecycle private volatile boolean running; - private final HttpServer server; + private final @Nullable HttpEndpointRequestHandler handler; + + private volatile @Nullable Vertx vertx; + + private volatile int actualPort = -1; public RestateHttpEndpointBean( Endpoint endpoint, RestateHttpServerProperties restateHttpServerProperties) { this.restateHttpServerProperties = restateHttpServerProperties; - this.server = - RestateHttpServer.fromHandler( - HttpEndpointRequestHandler.fromEndpoint( - endpoint, this.restateHttpServerProperties.isDisableBidirectionalStreaming())); + this.handler = + HttpEndpointRequestHandler.fromEndpoint( + endpoint, restateHttpServerProperties.isDisableBidirectionalStreaming()); } @Deprecated @@ -52,7 +61,7 @@ public RestateHttpEndpointBean( if (restateComponents.isEmpty()) { logger.info("No @RestateComponent discovered"); - this.server = null; + this.handler = null; // Don't start anything, if no service is registered return; } @@ -80,10 +89,10 @@ public RestateHttpEndpointBean( RestateRequestIdentityVerifier.fromKey(restateEndpointProperties.getIdentityKey())); } - this.server = - RestateHttpServer.fromHandler( - HttpEndpointRequestHandler.fromEndpoint( - builder.build(), restateHttpServerProperties.isDisableBidirectionalStreaming())); + Endpoint endpoint = builder.build(); + this.handler = + HttpEndpointRequestHandler.fromEndpoint( + endpoint, restateHttpServerProperties.isDisableBidirectionalStreaming()); } @Deprecated @@ -92,13 +101,39 @@ public void afterPropertiesSet() {} @Override public void start() { + HttpEndpointRequestHandler handler = this.handler; + if (handler == null) { + // No service registered, nothing to start. + this.running = true; + return; + } + int instances = + this.restateHttpServerProperties.getEventLoops() > 0 + ? this.restateHttpServerProperties.getEventLoops() + : VertxOptions.DEFAULT_EVENT_LOOP_POOL_SIZE; + Vertx vertx = Vertx.vertx(new VertxOptions().setEventLoopPoolSize(instances)); + this.vertx = vertx; + // A negative port makes all the server instances share the same random port, whereas port 0 + // would make each instance bind a different random port. + int listenPort = + this.restateHttpServerProperties.getPort() == 0 + ? -1 + : this.restateHttpServerProperties.getPort(); + AtomicInteger actualPortHolder = new AtomicInteger(); try { - this.server - .listen(this.restateHttpServerProperties.getPort()) + // Deploy one HttpServer (and one handler) per event loop, spreading connections across them. + vertx + .deployVerticle( + () -> new ServerVerticle(handler, listenPort, actualPortHolder), + new DeploymentOptions().setInstances(instances)) .toCompletionStage() .toCompletableFuture() .get(); - logger.info("Started Restate Spring HTTP server on port {}", this.server.actualPort()); + this.actualPort = actualPortHolder.get(); + logger.info( + "Started Restate Spring HTTP server on port {} with {} event loop(s)", + this.actualPort, + instances); } catch (Exception e) { logger.error( "Error when starting Restate Spring HTTP server on port {}", @@ -111,11 +146,16 @@ public void start() { @Override public void stop() { try { - this.server.close().toCompletionStage().toCompletableFuture().get(); + Vertx vertx = this.vertx; + if (vertx != null) { + vertx.close().toCompletionStage().toCompletableFuture().get(); + this.vertx = null; + } logger.info("Stopped Restate Spring HTTP server"); } catch (Exception e) { logger.error("Error when stopping the Restate Spring HTTP server", e); } + this.actualPort = -1; this.running = false; } @@ -128,6 +168,35 @@ public int actualPort() { if (!this.isRunning()) { return -1; } - return this.server.actualPort(); + return this.actualPort; + } + + /** + * Verticle deployed once per event loop, each creating its own {@link + * HttpEndpointRequestHandler}. + */ + private static final class ServerVerticle extends AbstractVerticle { + + private final HttpEndpointRequestHandler handler; + private final int port; + private final AtomicInteger actualPort; + + private ServerVerticle(HttpEndpointRequestHandler handler, int port, AtomicInteger actualPort) { + this.handler = handler; + this.port = port; + this.actualPort = actualPort; + } + + @Override + public void start(Promise startPromise) { + RestateHttpServer.fromHandler(vertx, handler) + .listen(port) + .onSuccess( + server -> { + actualPort.set(server.actualPort()); + startPromise.complete(); + }) + .onFailure(startPromise::fail); + } } } diff --git a/sdk-spring-boot/src/main/java/dev/restate/sdk/springboot/RestateHttpServerProperties.java b/sdk-spring-boot/src/main/java/dev/restate/sdk/springboot/RestateHttpServerProperties.java index 0baa78de9..9fcded763 100644 --- a/sdk-spring-boot/src/main/java/dev/restate/sdk/springboot/RestateHttpServerProperties.java +++ b/sdk-spring-boot/src/main/java/dev/restate/sdk/springboot/RestateHttpServerProperties.java @@ -18,14 +18,17 @@ public class RestateHttpServerProperties { private final int port; private final boolean disableBidirectionalStreaming; + private final int eventLoops; @ConstructorBinding public RestateHttpServerProperties( @Name("port") @DefaultValue(value = "9080") int port, @Name("disableBidirectionalStreaming") @DefaultValue(value = "false") - boolean disableBidirectionalStreaming) { + boolean disableBidirectionalStreaming, + @Name("eventLoops") @DefaultValue(value = "0") int eventLoops) { this.port = port; this.disableBidirectionalStreaming = disableBidirectionalStreaming; + this.eventLoops = eventLoops; } /** Port to expose the HTTP server. */ @@ -33,6 +36,14 @@ public int getPort() { return port; } + /** + * Number of event loops to run, spreading incoming connections across that many CPU cores. When + * {@code 0}, the default {@code 2 * availableProcessors} is used. + */ + public int getEventLoops() { + return eventLoops; + } + /** * If true, disable bidirectional streaming with HTTP/2 requests. Restate initiates for each * invocation a bidirectional streaming using HTTP/2 between restate-server and the SDK. In some