fruits) {
+}
diff --git a/benchmark-runner/src/main/java/io/netty/loom/benchmark/runner/HandoffHttpServer.java b/benchmark-runner/src/main/java/io/netty/loom/benchmark/runner/HandoffHttpServer.java
new file mode 100644
index 0000000..89d1921
--- /dev/null
+++ b/benchmark-runner/src/main/java/io/netty/loom/benchmark/runner/HandoffHttpServer.java
@@ -0,0 +1,336 @@
+/*
+ * Copyright 2026 The Netty VirtualThread Scheduler Project
+ *
+ * The Netty VirtualThread Scheduler Project licenses this file to you 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:
+ *
+ * https://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 io.netty.loom.benchmark.runner;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.IoEventLoop;
+import io.netty.channel.MultiThreadIoEventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.epoll.EpollIoHandler;
+import io.netty.channel.epoll.EpollServerSocketChannel;
+import io.netty.channel.nio.NioIoHandler;
+import io.netty.channel.socket.ServerSocketChannel;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpHeaderValues;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpServerCodec;
+import io.netty.handler.codec.http.HttpUtil;
+import io.netty.handler.codec.http.HttpVersion;
+import io.netty.loom.VirtualMultithreadIoEventLoopGroup;
+import io.netty.util.AsciiString;
+import io.netty.util.CharsetUtil;
+import org.apache.hc.client5.http.ConnectionKeepAliveStrategy;
+import org.apache.hc.client5.http.classic.methods.HttpGet;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
+import org.apache.hc.client5.http.impl.classic.HttpClientBuilder;
+import org.apache.hc.client5.http.impl.io.BasicHttpClientConnectionManager;
+import org.apache.hc.core5.http.HttpEntity;
+import org.apache.hc.core5.http.HttpResponse;
+import org.apache.hc.core5.http.io.entity.EntityUtils;
+import org.apache.hc.core5.http.protocol.HttpContext;
+import org.apache.hc.core5.util.TimeValue;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.function.Supplier;
+
+/**
+ * HTTP server that demonstrates handoff benchmark patterns.
+ *
+ * Processing flow: 1. Receive HTTP request on Netty event loop 2. Hand off to
+ * virtual thread (optionally using custom scheduler) 3. Make blocking HTTP call
+ * to mock server using JDK HttpClient 4. Parse JSON response with Jackson into
+ * Fruit objects 5. Re-encode to JSON and write back to client via event loop
+ *
+ * Usage: java -cp benchmark-runner.jar
+ * io.netty.loom.benchmark.runner.HandoffHttpServer \ --port 8081 \ --mock-url
+ * http://localhost:8080/fruits \ --threads 2 \ --use-custom-scheduler true \
+ * --io epoll
+ */
+public class HandoffHttpServer {
+
+ public enum IO {
+ EPOLL, NIO
+ }
+
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+ private static final ByteBuf HEALTH_RESPONSE = Unpooled
+ .unreleasableBuffer(Unpooled.copiedBuffer("OK", CharsetUtil.UTF_8));
+
+ private final int port;
+ private final String mockUrl;
+ private final int threads;
+ private final boolean useCustomScheduler;
+ private final IO io;
+ private final boolean silent;
+
+ private MultiThreadIoEventLoopGroup workerGroup;
+ private Channel serverChannel;
+ private Supplier threadFactorySupplier;
+
+ public HandoffHttpServer(int port, String mockUrl, int threads, boolean useCustomScheduler, IO io) {
+ this(port, mockUrl, threads, useCustomScheduler, io, false);
+ }
+
+ public HandoffHttpServer(int port, String mockUrl, int threads, boolean useCustomScheduler, IO io, boolean silent) {
+ this.port = port;
+ this.mockUrl = mockUrl;
+ this.threads = threads;
+ this.useCustomScheduler = useCustomScheduler;
+ this.io = io;
+ this.silent = silent;
+ }
+
+ public void start() throws InterruptedException {
+ var ioHandlerFactory = switch (io) {
+ case NIO -> NioIoHandler.newFactory();
+ case EPOLL -> EpollIoHandler.newFactory();
+ };
+
+ Class serverChannelClass = switch (io) {
+ case NIO -> NioServerSocketChannel.class;
+ case EPOLL -> EpollServerSocketChannel.class;
+ };
+
+ if (useCustomScheduler) {
+ var group = new VirtualMultithreadIoEventLoopGroup(threads, ioHandlerFactory);
+ threadFactorySupplier = group::vThreadFactory;
+ workerGroup = group;
+ } else {
+ workerGroup = new MultiThreadIoEventLoopGroup(threads, ioHandlerFactory);
+ var defaultFactory = Thread.ofVirtual().factory();
+ threadFactorySupplier = () -> defaultFactory;
+ }
+ ServerBootstrap b = new ServerBootstrap();
+ b.group(workerGroup).channel(serverChannelClass).childHandler(new ChannelInitializer() {
+ @Override
+ protected void initChannel(SocketChannel ch) {
+ ChannelPipeline p = ch.pipeline();
+ p.addLast(new HttpServerCodec());
+ p.addLast(new HttpObjectAggregator(65536));
+ p.addLast(new HandoffHandler());
+ }
+ });
+
+ serverChannel = b.bind(port).sync().channel();
+ if (!silent) {
+ System.out.printf("Handoff HTTP Server started on port %d%n", port);
+ System.out.printf(" Mock URL: %s%n", mockUrl);
+ System.out.printf(" Threads: %d%n", threads);
+ System.out.printf(" Custom Scheduler: %s%n", useCustomScheduler);
+ System.out.printf(" I/O: %s%n", io);
+ }
+ }
+
+ public void stop() {
+ if (serverChannel != null) {
+ serverChannel.close();
+ }
+ if (workerGroup != null) {
+ workerGroup.shutdownGracefully();
+ }
+ if (!silent) {
+ System.out.println("Server stopped");
+ }
+ }
+
+ public void awaitTermination() throws InterruptedException {
+ serverChannel.closeFuture().sync();
+ }
+
+ private class HandoffHandler extends SimpleChannelInboundHandler {
+
+ private final CloseableHttpClient httpClient;
+ private ExecutorService orderedExecutorService;
+
+ HandoffHandler() {
+ ConnectionKeepAliveStrategy keepAliveStrategy = (HttpResponse response,
+ HttpContext context) -> TimeValue.NEG_ONE_MILLISECOND;
+ BasicHttpClientConnectionManager cm = new BasicHttpClientConnectionManager();
+ httpClient = HttpClientBuilder.create().setConnectionManager(cm).setConnectionManagerShared(false)
+ .setKeepAliveStrategy(keepAliveStrategy).build();
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) throws Exception {
+ orderedExecutorService = Executors.newSingleThreadExecutor(threadFactorySupplier.get());
+ super.channelActive(ctx);
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
+ String uri = request.uri();
+ boolean keepAlive = HttpUtil.isKeepAlive(request);
+ IoEventLoop eventLoop = (IoEventLoop) ctx.channel().eventLoop();
+
+ if (uri.equals("/health")) {
+ sendResponse(ctx, HEALTH_RESPONSE.duplicate(), HttpHeaderValues.TEXT_PLAIN, keepAlive);
+ return;
+ }
+
+ if (uri.equals("/") || uri.startsWith("/fruits")) {
+ // Hand off to virtual thread for blocking processing
+ orderedExecutorService.execute(() -> {
+ doBlockingProcessing(ctx, eventLoop, keepAlive);
+ });
+ return;
+ }
+
+ // 404 for unknown paths
+ ByteBuf content = Unpooled.copiedBuffer("Not Found", CharsetUtil.UTF_8);
+ FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.NOT_FOUND,
+ content);
+ response.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN);
+ response.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, content.readableBytes());
+ ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
+ }
+
+ private void doBlockingProcessing(ChannelHandlerContext ctx, IoEventLoop eventLoop, boolean keepAlive) {
+ try {
+ // 1. Make blocking HTTP call to mock server
+ HttpGet httpGet = new HttpGet(mockUrl);
+ try (CloseableHttpResponse httpResponse = httpClient.execute(httpGet)) {
+ HttpEntity entity = httpResponse.getEntity();
+ if (entity == null)
+ throw new IOException("No response entity");
+ try (InputStream is = entity.getContent()) {
+ // 2. Parse JSON into Fruit objects using Jackson
+ FruitsResponse fruitsResponse = OBJECT_MAPPER.readValue(is, FruitsResponse.class);
+ // 3. Re-encode to JSON bytes
+ byte[] responseBytes = OBJECT_MAPPER.writeValueAsBytes(fruitsResponse);
+ // 4. Post write back to event loop (non-blocking)
+ eventLoop.execute(() -> {
+ ByteBuf content = Unpooled.wrappedBuffer(responseBytes);
+ sendResponse(ctx, content, HttpHeaderValues.APPLICATION_JSON, keepAlive);
+ });
+ }
+ EntityUtils.consumeQuietly(entity);
+ }
+ } catch (Throwable e) {
+ eventLoop.execute(() -> {
+ ByteBuf content = Unpooled.copiedBuffer("{\"error\":\"" + e.getMessage() + "\"}",
+ CharsetUtil.UTF_8);
+ FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
+ HttpResponseStatus.INTERNAL_SERVER_ERROR, content);
+ response.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON);
+ response.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, content.readableBytes());
+ ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
+ });
+ }
+ }
+
+ private void sendResponse(ChannelHandlerContext ctx, ByteBuf content, AsciiString contentType,
+ boolean keepAlive) {
+ FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK,
+ content);
+ response.headers().set(HttpHeaderNames.CONTENT_TYPE, contentType);
+ response.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, content.readableBytes());
+
+ if (keepAlive) {
+ response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
+ ctx.writeAndFlush(response);
+ } else {
+ ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
+ }
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+ ctx.close();
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ try {
+ orderedExecutorService.execute(() -> {
+ try {
+ httpClient.close();
+ } catch (IOException e) {
+ } finally {
+ orderedExecutorService.shutdown();
+ }
+ });
+ } finally {
+ super.channelInactive(ctx);
+ }
+ }
+ }
+
+ public static void main(String[] args) throws InterruptedException {
+ // Parse arguments
+ int port = 8081;
+ String mockUrl = "http://localhost:8080/fruits";
+ int threads = 1;
+ boolean useCustomScheduler = false;
+ IO io = IO.EPOLL;
+ boolean silent = false;
+
+ for (int i = 0; i < args.length; i++) {
+ switch (args[i]) {
+ case "--port" -> port = Integer.parseInt(args[++i]);
+ case "--mock-url" -> mockUrl = args[++i];
+ case "--threads" -> threads = Integer.parseInt(args[++i]);
+ case "--use-custom-scheduler" -> useCustomScheduler = Boolean.parseBoolean(args[++i]);
+ case "--io" -> io = IO.valueOf(args[++i].toUpperCase());
+ case "--silent" -> silent = true;
+ case "--help" -> {
+ printUsage();
+ return;
+ }
+ }
+ }
+
+ HandoffHttpServer server = new HandoffHttpServer(port, mockUrl, threads, useCustomScheduler, io, silent);
+ server.start();
+
+ // Shutdown hook
+ Runtime.getRuntime().addShutdownHook(new Thread(server::stop));
+
+ server.awaitTermination();
+ }
+
+ private static void printUsage() {
+ System.out.println("""
+ Usage: java -cp benchmark-runner.jar io.netty.loom.benchmark.runner.HandoffHttpServer [options]
+
+ Options:
+ --port HTTP port (default: 8081)
+ --mock-url Mock server URL (default: http://localhost:8080/fruits)
+ --threads Number of event loop threads (default: 1)
+ --use-custom-scheduler Use custom Netty scheduler (default: false)
+ --io I/O type (default: epoll)
+ --silent Suppress output messages
+ --help Show this help
+ """);
+ }
+}
diff --git a/benchmark-runner/src/main/java/io/netty/loom/benchmark/runner/MockHttpServer.java b/benchmark-runner/src/main/java/io/netty/loom/benchmark/runner/MockHttpServer.java
new file mode 100644
index 0000000..f1b0ba7
--- /dev/null
+++ b/benchmark-runner/src/main/java/io/netty/loom/benchmark/runner/MockHttpServer.java
@@ -0,0 +1,228 @@
+/*
+ * Copyright 2026 The Netty VirtualThread Scheduler Project
+ *
+ * The Netty VirtualThread Scheduler Project licenses this file to you 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:
+ *
+ * https://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 io.netty.loom.benchmark.runner;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpHeaderValues;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpServerCodec;
+import io.netty.handler.codec.http.HttpUtil;
+import io.netty.handler.codec.http.HttpVersion;
+import io.netty.util.CharsetUtil;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A minimal HTTP 1.1 mock server using plain Netty. Fast startup, configurable
+ * thread count and think time.
+ *
+ * Run as JAR:
+ *
+ *
+ * java -cp benchmark-runner.jar io.netty.loom.benchmark.runner.MockHttpServer 8080 100 1
+ *
+ *
+ * Arguments: [port] [thinkTimeMs] [threads]
+ *
+ * - port - HTTP port (default: 8080)
+ * - thinkTimeMs - delay before response in ms (default: 100)
+ * - threads - number of event loop threads (default: 1)
+ *
+ */
+public class MockHttpServer {
+
+ // Pre-computed cached response - a JSON list of fruits
+ private static final ByteBuf CACHED_RESPONSE = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("""
+ {
+ "fruits": [
+ {"name": "Apple", "color": "Red", "price": 1.20},
+ {"name": "Banana", "color": "Yellow", "price": 0.50},
+ {"name": "Orange", "color": "Orange", "price": 0.80},
+ {"name": "Grape", "color": "Purple", "price": 2.00},
+ {"name": "Mango", "color": "Yellow", "price": 1.50},
+ {"name": "Strawberry", "color": "Red", "price": 3.00},
+ {"name": "Blueberry", "color": "Blue", "price": 4.00},
+ {"name": "Pineapple", "color": "Yellow", "price": 2.50},
+ {"name": "Watermelon", "color": "Green", "price": 5.00},
+ {"name": "Kiwi", "color": "Brown", "price": 1.00}
+ ]
+ }
+ """, CharsetUtil.UTF_8));
+
+ private static final ByteBuf HEALTH_RESPONSE = Unpooled
+ .unreleasableBuffer(Unpooled.copiedBuffer("OK", CharsetUtil.UTF_8));
+
+ private final int port;
+ private final long thinkTimeMs;
+ private final int threads;
+ private final boolean silent;
+ private EventLoopGroup workerGroup;
+ private Channel serverChannel;
+
+ public MockHttpServer(int port, long thinkTimeMs, int threads) {
+ this(port, thinkTimeMs, threads, false);
+ }
+
+ public MockHttpServer(int port, long thinkTimeMs, int threads, boolean silent) {
+ this.port = port;
+ this.thinkTimeMs = thinkTimeMs;
+ this.threads = threads;
+ this.silent = silent;
+ }
+
+ public void start() throws InterruptedException {
+ workerGroup = new NioEventLoopGroup(threads);
+
+ ServerBootstrap b = new ServerBootstrap();
+ b.group(workerGroup).channel(NioServerSocketChannel.class)
+ .childHandler(new ChannelInitializer() {
+ @Override
+ protected void initChannel(SocketChannel ch) {
+ ChannelPipeline p = ch.pipeline();
+ p.addLast(new HttpServerCodec());
+ p.addLast(new HttpObjectAggregator(65536));
+ p.addLast(new HttpHandler(thinkTimeMs));
+ }
+ });
+
+ serverChannel = b.bind(port).sync().channel();
+ if (!silent) {
+ System.out.printf("Mock HTTP Server started on port %d with %dms think time using %d thread(s)%n", port,
+ thinkTimeMs, threads);
+ }
+ }
+
+ public void stop() {
+ if (serverChannel != null) {
+ serverChannel.close();
+ }
+ if (workerGroup != null) {
+ workerGroup.shutdownGracefully();
+ }
+ if (!silent) {
+ System.out.println("Server stopped");
+ }
+ }
+
+ private static class HttpHandler extends SimpleChannelInboundHandler {
+ private final long thinkTimeMs;
+
+ HttpHandler(long thinkTimeMs) {
+ this.thinkTimeMs = thinkTimeMs;
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
+ String uri = request.uri();
+ boolean keepAlive = HttpUtil.isKeepAlive(request);
+
+ if (uri.equals("/health")) {
+ sendResponse(ctx, HEALTH_RESPONSE.duplicate(), "text/plain", keepAlive);
+ return;
+ }
+
+ if (uri.equals("/fruits") || uri.equals("/")) {
+ if (thinkTimeMs > 0) {
+ // Schedule response after think time delay
+ ctx.executor().schedule(
+ () -> sendResponse(ctx, CACHED_RESPONSE.duplicate(), "application/json", keepAlive),
+ thinkTimeMs, TimeUnit.MILLISECONDS);
+ } else {
+ sendResponse(ctx, CACHED_RESPONSE.duplicate(), "application/json", keepAlive);
+ }
+ return;
+ }
+
+ // 404 for unknown paths
+ ByteBuf content = Unpooled.copiedBuffer("Not Found", CharsetUtil.UTF_8);
+ FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.NOT_FOUND,
+ content);
+ response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain");
+ response.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, content.readableBytes());
+ ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
+ }
+
+ private void sendResponse(ChannelHandlerContext ctx, ByteBuf content, String contentType, boolean keepAlive) {
+ FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK,
+ content);
+ response.headers().set(HttpHeaderNames.CONTENT_TYPE, contentType);
+ response.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, content.readableBytes());
+
+ if (keepAlive) {
+ response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
+ ctx.writeAndFlush(response);
+ } else {
+ ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
+ }
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+ cause.printStackTrace();
+ ctx.close();
+ }
+ }
+
+ public static void main(String[] args) throws InterruptedException {
+ int port = 8080;
+ long thinkTimeMs = 100;
+ int threads = 1;
+ boolean silent = false;
+
+ for (int i = 0; i < args.length; i++) {
+ switch (args[i]) {
+ case "--port" -> port = Integer.parseInt(args[++i]);
+ case "--think-time" -> thinkTimeMs = Long.parseLong(args[++i]);
+ case "--threads" -> threads = Integer.parseInt(args[++i]);
+ case "--silent" -> silent = true;
+ default -> {
+ // Legacy positional args support
+ if (i == 0)
+ port = Integer.parseInt(args[i]);
+ else if (i == 1)
+ thinkTimeMs = Long.parseLong(args[i]);
+ else if (i == 2)
+ threads = Integer.parseInt(args[i]);
+ }
+ }
+ }
+
+ MockHttpServer server = new MockHttpServer(port, thinkTimeMs, threads, silent);
+ server.start();
+
+ // Add shutdown hook for graceful shutdown
+ Runtime.getRuntime().addShutdownHook(new Thread(server::stop));
+
+ // Block main thread
+ server.serverChannel.closeFuture().sync();
+ }
+}
diff --git a/benchmark-runner/src/test/java/io/netty/loom/benchmark/runner/BenchmarkIntegrationTest.java b/benchmark-runner/src/test/java/io/netty/loom/benchmark/runner/BenchmarkIntegrationTest.java
new file mode 100644
index 0000000..91a2dff
--- /dev/null
+++ b/benchmark-runner/src/test/java/io/netty/loom/benchmark/runner/BenchmarkIntegrationTest.java
@@ -0,0 +1,213 @@
+/*
+ * Copyright 2026 The Netty VirtualThread Scheduler Project
+ *
+ * The Netty VirtualThread Scheduler Project licenses this file to you 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:
+ *
+ * https://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 io.netty.loom.benchmark.runner;
+
+import io.restassured.http.ContentType;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
+
+import static io.restassured.RestAssured.given;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.Matchers.*;
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Integration test that verifies MockHttpServer and HandoffHttpServer work
+ * correctly together with different configurations.
+ *
+ * Tests cover:
+ *
+ * - NIO I/O with default scheduler
+ * - NIO I/O with custom scheduler
+ * - EPOLL I/O with default scheduler
+ * - EPOLL I/O with custom scheduler
+ *
+ */
+class BenchmarkIntegrationTest {
+
+ private static final AtomicInteger PORT_COUNTER = new AtomicInteger(19000);
+
+ private MockHttpServer mockServer;
+ private HandoffHttpServer handoffServer;
+ private int mockPort;
+ private int handoffPort;
+
+ static Stream serverConfigurations() {
+ return Stream.of(
+ // IO type, use custom scheduler, description
+ Arguments.of(HandoffHttpServer.IO.NIO, false, "NIO with default scheduler"),
+ Arguments.of(HandoffHttpServer.IO.NIO, true, "NIO with custom scheduler"),
+ Arguments.of(HandoffHttpServer.IO.EPOLL, false, "EPOLL with default scheduler"),
+ Arguments.of(HandoffHttpServer.IO.EPOLL, true, "EPOLL with custom scheduler"));
+ }
+
+ void startServers(HandoffHttpServer.IO ioType, boolean useCustomScheduler) throws Exception {
+ // Use unique ports for each test to avoid conflicts
+ mockPort = PORT_COUNTER.getAndIncrement();
+ handoffPort = PORT_COUNTER.getAndIncrement();
+
+ // Start mock server with minimal think time for fast tests
+ mockServer = new MockHttpServer(mockPort, 0, 1);
+ mockServer.start();
+
+ // Wait for mock server to be ready
+ await().atMost(5, TimeUnit.SECONDS).until(() -> {
+ try {
+ return given().port(mockPort).when().get("/health").statusCode() == 200;
+ } catch (Exception e) {
+ return false;
+ }
+ });
+
+ // Start handoff server with specified configuration
+ handoffServer = new HandoffHttpServer(handoffPort, "http://localhost:" + mockPort + "/fruits", 1,
+ useCustomScheduler, ioType);
+ handoffServer.start();
+
+ // Wait for handoff server to be ready
+ await().atMost(5, TimeUnit.SECONDS).until(() -> {
+ try {
+ return given().port(handoffPort).when().get("/health").statusCode() == 200;
+ } catch (Exception e) {
+ return false;
+ }
+ });
+ }
+
+ @AfterEach
+ void stopServers() {
+ if (handoffServer != null) {
+ handoffServer.stop();
+ handoffServer = null;
+ }
+ if (mockServer != null) {
+ mockServer.stop();
+ mockServer = null;
+ }
+ }
+
+ @ParameterizedTest(name = "{2}")
+ @MethodSource("serverConfigurations")
+ void mockServerHealthEndpoint(HandoffHttpServer.IO ioType, boolean useCustomScheduler, String description)
+ throws Exception {
+ startServers(ioType, useCustomScheduler);
+ given().port(mockPort).when().get("/health").then().statusCode(200).body(equalTo("OK"));
+ }
+
+ @ParameterizedTest(name = "{2}")
+ @MethodSource("serverConfigurations")
+ void mockServerFruitsEndpoint(HandoffHttpServer.IO ioType, boolean useCustomScheduler, String description)
+ throws Exception {
+ startServers(ioType, useCustomScheduler);
+ given().port(mockPort).when().get("/fruits").then().statusCode(200).contentType(ContentType.JSON)
+ .body("fruits", hasSize(10)).body("fruits[0].name", equalTo("Apple"))
+ .body("fruits[0].color", equalTo("Red")).body("fruits[0].price", equalTo(1.20f));
+ }
+
+ @ParameterizedTest(name = "{2}")
+ @MethodSource("serverConfigurations")
+ void handoffServerHealthEndpoint(HandoffHttpServer.IO ioType, boolean useCustomScheduler, String description)
+ throws Exception {
+ startServers(ioType, useCustomScheduler);
+ given().port(handoffPort).when().get("/health").then().statusCode(200).body(equalTo("OK"));
+ }
+
+ @ParameterizedTest(name = "{2}")
+ @MethodSource("serverConfigurations")
+ void handoffServerFruitsEndpoint(HandoffHttpServer.IO ioType, boolean useCustomScheduler, String description)
+ throws Exception {
+ startServers(ioType, useCustomScheduler);
+ given().port(handoffPort).when().get("/fruits").then().statusCode(200).contentType(ContentType.JSON)
+ .body("fruits", hasSize(10)).body("fruits[0].name", equalTo("Apple"))
+ .body("fruits[0].color", equalTo("Red"));
+ }
+
+ @ParameterizedTest(name = "{2}")
+ @MethodSource("serverConfigurations")
+ void handoffServerRootEndpoint(HandoffHttpServer.IO ioType, boolean useCustomScheduler, String description)
+ throws Exception {
+ startServers(ioType, useCustomScheduler);
+ given().port(handoffPort).when().get("/").then().statusCode(200).contentType(ContentType.JSON).body("fruits",
+ hasSize(10));
+ }
+
+ @ParameterizedTest(name = "{2}")
+ @MethodSource("serverConfigurations")
+ void handoffServer404ForUnknownPath(HandoffHttpServer.IO ioType, boolean useCustomScheduler, String description)
+ throws Exception {
+ startServers(ioType, useCustomScheduler);
+ given().port(handoffPort).when().get("/unknown").then().statusCode(404);
+ }
+
+ @ParameterizedTest(name = "{2}")
+ @MethodSource("serverConfigurations")
+ void handoffServerReturnsAllFruits(HandoffHttpServer.IO ioType, boolean useCustomScheduler, String description)
+ throws Exception {
+ startServers(ioType, useCustomScheduler);
+
+ List fruitNames = given().port(handoffPort).when().get("/fruits").then().statusCode(200).extract()
+ .jsonPath().getList("fruits.name", String.class);
+
+ assertEquals(10, fruitNames.size());
+ assertTrue(fruitNames.contains("Apple"));
+ assertTrue(fruitNames.contains("Banana"));
+ assertTrue(fruitNames.contains("Kiwi"));
+ }
+
+ @ParameterizedTest(name = "{2}")
+ @MethodSource("serverConfigurations")
+ void handoffServerHandlesMultipleRequests(HandoffHttpServer.IO ioType, boolean useCustomScheduler,
+ String description) throws Exception {
+ startServers(ioType, useCustomScheduler);
+
+ // Send multiple requests to verify server handles concurrent load
+ for (int i = 0; i < 10; i++) {
+ given().port(handoffPort).when().get("/fruits").then().statusCode(200).body("fruits", hasSize(10));
+ }
+ }
+
+ @ParameterizedTest(name = "{2}")
+ @MethodSource("serverConfigurations")
+ void verifyEndToEndJsonParsing(HandoffHttpServer.IO ioType, boolean useCustomScheduler, String description)
+ throws Exception {
+ startServers(ioType, useCustomScheduler);
+
+ // This test verifies the complete flow:
+ // 1. HandoffHttpServer receives request
+ // 2. Makes blocking call to MockHttpServer
+ // 3. Parses JSON with Jackson into Fruit objects
+ // 4. Re-encodes and returns
+
+ FruitsResponse response = given().port(handoffPort).when().get("/fruits").then().statusCode(200).extract()
+ .as(FruitsResponse.class);
+
+ assertNotNull(response);
+ assertNotNull(response.fruits());
+ assertEquals(10, response.fruits().size());
+
+ Fruit apple = response.fruits().stream().filter(f -> "Apple".equals(f.name())).findFirst().orElse(null);
+
+ assertNotNull(apple);
+ assertEquals("Red", apple.color());
+ assertEquals(1.20, apple.price(), 0.01);
+ }
+}
diff --git a/benchmarks/pom.xml b/benchmarks/pom.xml
index efa129a..23bf59d 100644
--- a/benchmarks/pom.xml
+++ b/benchmarks/pom.xml
@@ -13,6 +13,11 @@
netty-virtualthread-benchmarks
+
+ org.jctools
+ jctools-core
+ 4.0.5
+
io.netty.loom
netty-virtualthread-core
diff --git a/benchmarks/src/main/java/io/netty/loom/benchmark/SchedulerHandoffBenchmark.java b/benchmarks/src/main/java/io/netty/loom/benchmark/SchedulerHandoffBenchmark.java
new file mode 100644
index 0000000..a156b1a
--- /dev/null
+++ b/benchmarks/src/main/java/io/netty/loom/benchmark/SchedulerHandoffBenchmark.java
@@ -0,0 +1,281 @@
+/*
+ * Copyright 2026 The Netty VirtualThread Scheduler Project
+ *
+ * The Netty VirtualThread Scheduler Project licenses this file to you 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:
+ *
+ * https://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 io.netty.loom.benchmark;
+
+import io.netty.channel.IoEventLoop;
+import io.netty.channel.MultiThreadIoEventLoopGroup;
+import io.netty.channel.epoll.EpollIoHandler;
+import io.netty.channel.nio.NioIoHandler;
+import io.netty.channel.uring.IoUringIoHandler;
+import io.netty.loom.VirtualMultithreadIoEventLoopGroup;
+import io.netty.util.concurrent.FastThreadLocal;
+import org.HdrHistogram.Histogram;
+import org.jctools.queues.MpscArrayQueue;
+import org.jctools.util.Pow2;
+import org.openjdk.jmh.annotations.*;
+import org.openjdk.jmh.infra.BenchmarkParams;
+import org.openjdk.jmh.infra.Blackhole;
+
+import java.util.Arrays;
+import java.util.Queue;
+import java.util.concurrent.*;
+import java.util.function.Supplier;
+
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@Warmup(iterations = 10, time = 1)
+@Measurement(iterations = 10, time = 1)
+@State(Scope.Benchmark)
+public class SchedulerHandoffBenchmark {
+
+ public enum IO {
+ EPOLL, NIO, IO_URING
+ }
+
+ // default is 1000 rps per user
+ @Param({"100"})
+ int serviceTimeUs;
+
+ @Param({"200000"})
+ int rate;
+
+ @Param({"512"})
+ int requestBytes;
+
+ @Param({"1024"})
+ int responseBytes;
+
+ // the total throughput will be roughly concurrency * 1000 / serviceTimeMs
+ @Param({"50"})
+ int concurrency;
+
+ private static final int EL_THREADS = Integer.getInteger("elThreads", -1);
+
+ @Param({"EPOLL"})
+ IO io;
+
+ MultiThreadIoEventLoopGroup group;
+
+ Supplier threadFactory;
+
+ record RequestData(byte[] request, byte[] response) {
+ }
+
+ // let's make
+ Queue requestData;
+
+ private static final CopyOnWriteArrayList histograms = new CopyOnWriteArrayList<>();
+
+ private static final FastThreadLocal rttHistogram = new FastThreadLocal<>() {
+ @Override
+ public Histogram initialValue() {
+ var histo = new Histogram(3);
+ histograms.add(histo);
+ return histo;
+ }
+ };
+
+ private long fireTimePeriodNs;
+ private boolean allOutThroughput;
+ private long nextFireTime;
+
+ @Setup(Level.Trial)
+ public void resetHistograms() {
+ // TODO verify if the trial is the right level
+ histograms.forEach(Histogram::reset);
+ }
+
+ @Setup
+ public void setup(BenchmarkParams params) throws ExecutionException, InterruptedException {
+ if (rate < 0) {
+ allOutThroughput = true;
+ } else {
+ fireTimePeriodNs = (long) (1000_000_000d / rate);
+ }
+ if (EL_THREADS <= 0) {
+ throw new IllegalStateException("Please set the elThreads system property to a positive integer");
+ }
+ var ioFactory = switch (io) {
+ case NIO -> NioIoHandler.newFactory();
+ case IO_URING -> IoUringIoHandler.newFactory();
+ case EPOLL -> EpollIoHandler.newFactory();
+ };
+ if (params.getBenchmark().contains("custom")) {
+ var group = new VirtualMultithreadIoEventLoopGroup(EL_THREADS, ioFactory);
+ threadFactory = group::vThreadFactory;
+ this.group = group;
+ } else {
+ group = new MultiThreadIoEventLoopGroup(EL_THREADS, ioFactory);
+ var sameFactory = Thread.ofVirtual().factory();
+ threadFactory = () -> sameFactory;
+ }
+ if (concurrency > 0) {
+ requestData = new MpscArrayQueue<>(Pow2.roundToPowerOfTwo(concurrency));
+ for (int i = 0; i < concurrency; i++) {
+ requestData.offer(new RequestData(new byte[requestBytes], new byte[responseBytes]));
+ }
+ } else {
+ requestData = null;
+ }
+ nextFireTime = System.nanoTime();
+ }
+
+ private static void spinUntil(long targetTimeNs) {
+ while (System.nanoTime() < targetTimeNs) {
+ Thread.onSpinWait();
+ }
+ }
+
+ @Benchmark
+ @Fork(value = 2, jvmArgs = {"--add-opens=java.base/java.lang=ALL-UNNAMED", "-XX:+UnlockExperimentalVMOptions",
+ "-XX:-DoJVMTIVirtualThreadTransitions", "-Djdk.trackAllThreads=false",
+ "-Djdk.virtualThreadScheduler.implClass=io.netty.loom.NettyScheduler", "-Djdk.pollerMode=3",
+ "-DelThreads=1"})
+ public void customScheduler(Blackhole bh) throws InterruptedException {
+ doRequest(bh);
+ }
+
+ @Benchmark
+ @Fork(value = 2, jvmArgs = {"--add-opens=java.base/java.lang=ALL-UNNAMED", "-XX:+UnlockExperimentalVMOptions",
+ "-XX:-DoJVMTIVirtualThreadTransitions", "-Djdk.trackAllThreads=false",
+ "-Djdk.virtualThreadScheduler.implClass=io.netty.loom.NettyScheduler", "-Djdk.pollerMode=3",
+ "-DelThreads=2"})
+ public void customSchedulerTwoEL(Blackhole bh) throws InterruptedException {
+ doRequest(bh);
+ }
+
+ @Benchmark
+ @Fork(value = 2, jvmArgs = {"--add-opens=java.base/java.lang=ALL-UNNAMED", "-XX:+UnlockExperimentalVMOptions",
+ "-XX:-DoJVMTIVirtualThreadTransitions", "-Djdk.trackAllThreads=false", "-DelThreads=1",
+ "-Djdk.virtualThreadScheduler.parallelism=1"})
+ public void defaultScheduler(Blackhole bh) throws InterruptedException {
+ doRequest(bh);
+ }
+
+ // just burn a full core on this!
+ private void doRequest(Blackhole bh) {
+ if (!allOutThroughput) {
+ spinUntil(nextFireTime);
+ }
+ var data = spinWaitRequest();
+ // avoid coordinated omission!
+ long startRequest = allOutThroughput ? System.nanoTime() : nextFireTime;
+ if (!allOutThroughput) {
+ nextFireTime += fireTimePeriodNs;
+ }
+ // write some data into the request
+ byte[] request = data == null ? new byte[requestBytes] : data.request;
+ byte[] response = data == null ? null : data.response;
+
+ Arrays.fill(request, (byte) 1);
+
+ // this is handing off this to the loom scheduler
+ var el = group.next();
+ el.execute(() -> {
+ // This is running in a Netty event loop thread
+ // process the request by reading it
+ for (byte b : request) {
+ bh.consume(b);
+ }
+ // off-load the actual (blocking) processing to a virtual thread
+ threadFactory.get().newThread(() -> {
+ blockingProcess(bh, el, startRequest, data, response);
+ }).start();
+ });
+ }
+
+ // this is required to make sure NONE of the fine grain ops like writeByte won't
+ // be inlined
+ @CompilerControl(CompilerControl.Mode.DONT_INLINE)
+ private void blockingProcess(Blackhole bh, IoEventLoop el, long startRequest, RequestData data, byte[] response) {
+ try {
+ // simulate processing time:
+ // NOTE: if we're using sleep here, the built-in scheduler will use the FJ
+ // built-in one
+ // but the custom scheduler, nope, see
+ // https://github.com/openjdk/loom/blob/3d9e866f60bdebc55b59b9fd40a4898002c35e96/src/java.base/share/classes/java/lang/VirtualThread.java#L1629
+ if (serviceTimeUs > 0) {
+ TimeUnit.MICROSECONDS.sleep(serviceTimeUs);
+ }
+ // write the response content
+ if (response == null) {
+ response = new byte[responseBytes];
+ }
+ for (int i = 0; i < responseBytes; i++) {
+ response[i] = (byte) 42;
+ }
+ byte[] responseCreated = response;
+ el.execute(() -> {
+ nonBlockingCompleteProcessing(bh, startRequest, data, responseCreated);
+ });
+ } catch (Throwable e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ // this is required to make sure NONE of the fine grain ops like writeByte won't
+ // be inlined
+ @CompilerControl(CompilerControl.Mode.DONT_INLINE)
+ private void nonBlockingCompleteProcessing(Blackhole bh, long startRequest, RequestData data, byte[] response) {
+ // read the response
+ int toRead = this.responseBytes;
+ for (int i = 0; i < toRead; i++) {
+ bh.consume(response[i]);
+ }
+ // record RTT
+ long rttNs = System.nanoTime() - startRequest;
+ Histogram histogram = rttHistogram.get();
+ histogram.recordValue(rttNs);
+ // offer it just at the end
+ if (data != null) {
+ requestData.add(data);
+ }
+ }
+
+ private RequestData spinWaitRequest() {
+ Queue requestData = this.requestData;
+ if (requestData == null) {
+ return null;
+ }
+ var request = requestData.poll();
+ while (request == null) {
+ Thread.onSpinWait();
+ request = requestData.poll();
+ }
+ return request;
+ }
+
+ @TearDown
+ public void shutdown() throws ExecutionException, InterruptedException {
+ // wait for all tasks to complete
+ if (concurrency > 0) {
+ for (int i = 0; i < concurrency; i++) {
+ spinWaitRequest();
+ }
+ } else {
+ // TODO enqueue for each EL a request waiting to complete?
+ }
+ group.shutdownGracefully().get();
+ // print percentiles of RTT
+ Histogram combined = new Histogram(3);
+ histograms.forEach(combined::add);
+ histograms.clear();
+ // Print percentile distribution
+ System.out.printf("RTT (µs) - Avg: %.2f, P50: %.2f, P90: %.2f, P99: %.2f, Max: %.2f%n",
+ combined.getMean() / 1000.0, combined.getValueAtPercentile(50) / 1000.0,
+ combined.getValueAtPercentile(90) / 1000.0, combined.getValueAtPercentile(99) / 1000.0,
+ combined.getMaxValue() / 1000.0);
+ }
+}
diff --git a/core/src/main/java/io/netty/loom/EventLoopScheduler.java b/core/src/main/java/io/netty/loom/EventLoopScheduler.java
index 4e54987..1276880 100644
--- a/core/src/main/java/io/netty/loom/EventLoopScheduler.java
+++ b/core/src/main/java/io/netty/loom/EventLoopScheduler.java
@@ -149,15 +149,21 @@ private void nettyEventLoop() {
boolean canBlock = false;
while (!ioEventLoop.isShuttingDown()) {
canBlock = runIO(canBlock);
- Thread.yield();
+ if (!runQueue.isEmpty()) {
+ Thread.yield();
+ }
// try running leftover write tasks before checking for I/O tasks
canBlock &= ioEventLoop.runNonBlockingTasks(RUNNING_YIELD_US) == 0;
- Thread.yield();
+ if (!runQueue.isEmpty()) {
+ Thread.yield();
+ }
}
// we are shutting down, it shouldn't take long so let's spin a bit :P
while (!ioEventLoop.isTerminated()) {
ioEventLoop.runNow();
- Thread.yield();
+ if (!runQueue.isEmpty()) {
+ Thread.yield();
+ }
}
}
@@ -284,7 +290,9 @@ public boolean execute(Thread.VirtualThreadTask task) {
// continuations
// whilst is just woken up for I/O
assert eventLoopContinuatioToRun == null;
- Thread.yield();
+ if (!runQueue.isEmpty()) {
+ Thread.yield();
+ }
}
return true;
}
diff --git a/pom.xml b/pom.xml
index bc9a66b..7a5097e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -12,6 +12,7 @@
core
benchmarks
+ benchmark-runner
example-echo