Add support for configurable event loops and enable multi-event-loop setup by default.

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<HttpServer> 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<Void> 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 <E extends Throwable> void sneakyThrow(Throwable e) throws E {
     throw (E) e;

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<Void> startPromise) {
+      RestateHttpServer.fromHandler(vertx, handler)
+          .listen(port)
+          .onSuccess(
+              server -> {
+                actualPort.set(server.actualPort());
+                startPromise.complete();
+              })
+          .onFailure(startPromise::fail);
+    }
   }
 }

sdk-spring-boot/src/main/java/dev/restate/sdk/springboot/RestateHttpServerProperties.java

@@ -18,21 +18,32 @@ 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. */
   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