Add support for virtual-pool? and connector threads configuration

Author: k13gomez

.github/workflows/test.yml

@@ -2,6 +2,9 @@ name: Tests
 on: [push, pull_request]
 jobs:
   test:
+    strategy:
+      matrix:
+        java_version: [17, 19, 21, 24]
     runs-on: ubuntu-latest
     steps:
       - name: Checkout
@@ -11,7 +14,7 @@ jobs:
         uses: actions/setup-java@v3
         with:
           distribution: 'zulu'
-          java-version: '17'
+          java-version: ${{matrix.java_version}}
 
       - name: Install clojure tools
         uses: DeLaGuardo/setup-clojure@10.1
@@ -46,4 +49,6 @@ jobs:
         working-directory: ./ring-jakarta-servlet
 
       - name: Run tests
+        env:
+          _JAVA_OPTIONS: ${{ matrix.java_version == 19 && '--enable-preview' || '' }}
         run: lein sub test-all

ring-jetty-adapter/src/ring/adapter/jetty.clj

@@ -19,7 +19,7 @@
            [org.eclipse.jetty.unixdomain.server UnixDomainServerConnector]
            [org.eclipse.jetty.ee9.servlet ServletContextHandler ServletHandler]
            [org.eclipse.jetty.util BlockingArrayQueue]
-           [org.eclipse.jetty.util.thread ThreadPool QueuedThreadPool]
+           [org.eclipse.jetty.util.thread ThreadPool QueuedThreadPool VirtualThreadPool]
            [org.eclipse.jetty.util.ssl SslContextFactory$Server KeyStoreScanner]
            [org.eclipse.jetty.ee9.nested Request]
            [org.eclipse.jetty.ee9.websocket.server
@@ -168,14 +168,38 @@
     (.setAllowNullPathInfo true)
     (JettyWebSocketServletContainerInitializer/configure nil)))
 
-(defn- server-connector ^ServerConnector [^Server server & factories]
-  (ServerConnector. server #^"[Lorg.eclipse.jetty.server.ConnectionFactory;"
-                             (into-array ConnectionFactory factories)))
+(defn- get-connector-thread-options [options]
+  (let [acceptors (options :acceptor-threads
+                           (when (:virtual-pool? options)
+                             1))
+        selectors (options :selector-threads
+                           (when acceptors
+                             (-> (Runtime/getRuntime) (.availableProcessors))))]
+    {:acceptors acceptors
+     :selectors selectors}))
+
+(defn- server-connector ^ServerConnector [^Server server factories options]
+  (let [{:keys [acceptors
+                selectors]} (get-connector-thread-options options)]
+    (if (and acceptors selectors)
+      (ServerConnector. server (int acceptors) (int selectors)
+                        ^{:tag "[Lorg.eclipse.jetty.server.ConnectionFactory;"}
+                        (into-array ConnectionFactory factories))
+      (ServerConnector. server
+                        ^{:tag "[Lorg.eclipse.jetty.server.ConnectionFactory;"}
+                        (into-array ConnectionFactory factories)))))
 
 (defn- unix-domain-server-connector ^UnixDomainServerConnector
-  [^Server server & factories]
-  (UnixDomainServerConnector. server #^"[Lorg.eclipse.jetty.server.ConnectionFactory;"
-                                       (into-array ConnectionFactory factories)))
+  [^Server server factories options]
+  (let [{:keys [acceptors
+                selectors]} (get-connector-thread-options options)]
+    (if (and acceptors selectors)
+      (UnixDomainServerConnector. server (int acceptors) (int selectors)
+                                  ^{:tag "[Lorg.eclipse.jetty.server.ConnectionFactory;"}
+                                  (into-array ConnectionFactory factories))
+      (UnixDomainServerConnector. server
+                                  ^{:tag "[Lorg.eclipse.jetty.server.ConnectionFactory;"}
+                                  (into-array ConnectionFactory factories)))))
 
 (defn- http-config ^HttpConfiguration [options]
   (doto (HttpConfiguration.)
@@ -187,7 +211,7 @@
 
 (defn- http-connector ^ServerConnector [server options]
   (let [http-factory (HttpConnectionFactory. (http-config options))]
-    (doto (server-connector server http-factory)
+    (doto (server-connector server [http-factory] options)
       (.setPort (options :port 80))
       (.setHost (options :host))
       (.setIdleTimeout (options :max-idle-time 200000)))))
@@ -239,7 +263,7 @@
     (when-let [scan-interval (options :keystore-scan-interval)]
       (.addBean server (doto (KeyStoreScanner. ssl-context)
                          (.setScanInterval scan-interval))))
-    (doto (server-connector server ssl-factory http-factory)
+    (doto (server-connector server [ssl-factory http-factory] options)
       (.setPort ssl-port)
       (.setHost (options :host))
       (.setIdleTimeout (options :max-idle-time 200000)))))
@@ -248,7 +272,7 @@
   (let [http-factory (HttpConnectionFactory. (http-config options))
         socket (io/file (options :unix-socket))]
     (.deleteOnExit socket)
-    (doto (unix-domain-server-connector server http-factory)
+    (doto (unix-domain-server-connector server [http-factory] options)
       (.setUnixDomainPath (.toPath socket))
       (.setIdleTimeout (options :max-idle-time 200000)))))
 
@@ -269,8 +293,16 @@
       (.setDaemon pool true))
     pool))
 
+(defn- create-virtual-threadpool [options]
+  (let [max-threads         (options :max-threads 50)
+        pool                (VirtualThreadPool. max-threads)]
+    pool))
+
 (defn- create-server ^Server [options]
-  (let [pool   (or (:thread-pool options) (create-threadpool options))
+  (let [pool   (or (:thread-pool options)
+                   (when (:virtual-pool? options false)
+                     (create-virtual-threadpool options))
+                   (create-threadpool options))
         server (Server. ^ThreadPool pool)]
     (when (:http? options true)
       (.addConnector server (http-connector server options)))
@@ -317,8 +349,13 @@
   :keystore-scan-interval - if not nil, the interval in seconds to scan for an
                             updated keystore
   :thread-pool            - custom thread pool instance for Jetty to use
+  :virtual-pool?          - use a VirtualThreadPool (requires Java 19+)
   :truststore             - a truststore to use for SSL connections
   :trust-password         - the password to the truststore
+  :acceptor-threads       - the number of acceptor threads to use
+                            (default 1 if using virtual-pool?, otherwise not set)
+  :selector-threads       - the number of selector threads to use
+                            (default to available processors if using virtual-pool?, otherwise not set)
   :max-threads            - the maximum number of threads to use (default 50)
   :min-threads            - the minimum number of threads to use (default 8)
   :max-queued-requests    - the maximum number of requests to be queued

ring-jetty-adapter/test/ring/adapter/test/jetty.clj

@@ -11,7 +11,7 @@
   (:import [java.io File]
            [java.nio ByteBuffer]
            [java.nio.file Paths]
-           [org.eclipse.jetty.util.thread QueuedThreadPool]
+           [org.eclipse.jetty.util.thread QueuedThreadPool VirtualThreadPool]
            [org.eclipse.jetty.server Server Request SslConnectionFactory]
            [org.eclipse.jetty.server.handler AbstractHandler]
            [org.eclipse.jetty.io ClientConnector Transport$TCPUnix]
@@ -103,6 +103,11 @@
    :join?           false
    :sni-host-check? false})
 
+(def java-version
+  (->> (System/getProperty "java.version")
+       (re-find #"\A\d+")
+       (Integer/parseInt)))
+
 (deftest test-run-jetty
   (testing "HTTP server"
     (with-server hello-world {:port test-port}
@@ -112,23 +117,29 @@
                          "text/plain"))
         (is (= (:body response) "Hello World")))))
 
-  (let [java-version (->> (System/getProperty "java.version")
-                          (re-find #"\A\d+")
-                          (Integer/parseInt))]
-    (when (>= java-version 16)
-      (testing "UNIX Socket server"
-        (with-server hello-world {:http? false
-                                  :unix-socket test-unix-domain-socket}
-          (let [path (Paths/get test-unix-domain-socket (make-array String 0))
-                transport (HttpClientTransportOverHTTP. (ClientConnector.))
-                client    (doto (HttpClient. transport) (.start))
-                response  (-> client
-                              (.newRequest  "http://localhost")
-                              (.transport (Transport$TCPUnix. path))
-                              (.send))]
-            (is (= (.getStatus response) 200))
-            (is (.getMediaType response) "text/plain")
-            (is (= (.getContentAsString response) "Hello World")))))))
+  (when (>= java-version 16)
+    (testing "UNIX Socket server"
+      (with-server hello-world {:http? false
+                                :unix-socket test-unix-domain-socket}
+        (let [path (Paths/get test-unix-domain-socket (make-array String 0))
+              transport (HttpClientTransportOverHTTP. (ClientConnector.))
+              client    (doto (HttpClient. transport) (.start))
+              response  (-> client
+                            (.newRequest  "http://localhost")
+                            (.transport (Transport$TCPUnix. path))
+                            (.send))]
+          (is (= (.getStatus response) 200))
+          (is (.getMediaType response) "text/plain")
+          (is (= (.getContentAsString response) "Hello World"))))
+
+      (when (>= java-version 19)
+        (testing "with custom connector options"
+          (let [server (run-jetty hello-world {:http? false
+                                               :unix-socket test-unix-domain-socket
+                                               :join? false
+                                               :acceptor-threads 2})]
+            (is (= 2 (-> server (.getConnectors) first (.getAcceptors))))
+            (.stop server))))))
 
   (testing "HTTPS server"
     (with-server hello-world {:port test-port
@@ -331,6 +342,36 @@
       (is (= 1000 (. thread-pool getIdleTimeout)))
       (.stop server)))
 
+  (when (>= java-version 19)
+    (testing "using default connector options with virtual-pool?"
+      (let [server (run-jetty hello-world {:port test-port
+                                           :join? false
+                                           :virtual-pool? true})]
+        (is (instance? VirtualThreadPool (.getThreadPool server)))
+        (is (= 1 (-> server (.getConnectors) first (.getAcceptors))))
+        (is (= (-> (Runtime/getRuntime) (.availableProcessors))
+               (-> server (.getConnectors) first (.getSelectorManager) (.getSelectorCount))))
+        (.stop server))))
+
+  (testing "using missing connector options with default pool"
+    (let [server (run-jetty hello-world {:port test-port
+                                         :join? false})]
+      (is (instance? QueuedThreadPool (.getThreadPool server)))
+      (is (= 1 (-> server (.getConnectors) first (.getAcceptors))))
+      (is (> (-> (Runtime/getRuntime) (.availableProcessors))
+             (-> server (.getConnectors) first (.getSelectorManager) (.getSelectorCount))))
+      (.stop server)))
+
+  (testing "using custom connector options"
+    (let [server (run-jetty hello-world {:port test-port
+                                         :join? false
+                                         :acceptor-threads 2
+                                         :selector-threads 8})]
+      (is (= 2 (-> server (.getConnectors) first (.getAcceptors))))
+      (is (= 8
+             (-> server (.getConnectors) first (.getSelectorManager) (.getSelectorCount))))
+      (.stop server)))
+
   (testing "providing custom thread-pool"
     (let [pool   (QueuedThreadPool.)
           server (run-jetty hello-world {:port test-port