feat: Add loom Support. (#159)

Motivation:
Add Loom support.
refs: #90 , again 

Modification:
1. add `handlerExecutor()` and some helper methods for virtual threads.
2. some documents.

Result:
Virtual threads supported.

`wrk` is needed to run the benchmark

- miniAppWithSleep 100ms : ↑2300%
- todoDb:           ~↓6%
- staticFiles:      ~↓6%

```shell
./mill --no-build-lock benchmark.runBenchmarks
```

Results of same 4 Carrier/Platform threads:

```scala
[1] staticFilesWithLoom result with (platform threads):
[1] Running 30s test @ http://localhost:8080/
[1]   4 threads and 100 connections
[1]   Thread Stats   Avg      Stdev     Max   +/- Stdev
[1]     Latency     1.23ms  436.88us  28.64ms   98.91%
[1]     Req/Sec    20.54k     1.08k   22.69k    93.44%
[1]   2461250 requests in 30.10s, 342.70MB read
[1] Requests/sec:  81766.27
[1] Transfer/sec:     11.38MB
[1] 
[1] staticFilesWithLoom result with (virtual threads):
[1] Running 30s test @ http://localhost:8080/
[1]   4 threads and 100 connections
[1]   Thread Stats   Avg      Stdev     Max   +/- Stdev
[1]     Latency     4.41ms   14.69ms 157.91ms   95.15%
[1]     Req/Sec    19.00k     4.29k   22.09k    88.63%
[1]   2266289 requests in 30.02s, 315.55MB read
[1] Requests/sec:  75488.32
[1] Transfer/sec:     10.51MB
[1] 
[1] todoDbWithLoom result with (platform threads):
[1] Running 30s test @ http://localhost:8080/
[1]   4 threads and 100 connections
[1]   Thread Stats   Avg      Stdev     Max   +/- Stdev
[1]     Latency     1.22ms  172.42us  11.67ms   96.46%
[1]     Req/Sec    20.62k     1.29k   42.72k    93.01%
[1]   2466143 requests in 30.10s, 395.12MB read
[1]   Non-2xx or 3xx responses: 2466143
[1] Requests/sec:  81929.40
[1] Transfer/sec:     13.13MB
[1] 
[1] todoDbWithLoom result with (virtual threads):
[1] Running 30s test @ http://localhost:8080/
[1]   4 threads and 100 connections
[1]   Thread Stats   Avg      Stdev     Max   +/- Stdev
[1]     Latency     3.97ms   13.20ms 160.29ms   95.28%
[1]     Req/Sec    19.21k     3.68k   22.32k    89.41%
[1]   2284539 requests in 30.03s, 366.02MB read
[1]   Non-2xx or 3xx responses: 2284539
[1] Requests/sec:  76072.80
[1] Transfer/sec:     12.19MB
[1] 
[1] minimalApplicationWithLoom result with (platform threads):
[1] Running 30s test @ http://localhost:8080/
[1]   4 threads and 100 connections
[1]   Thread Stats   Avg      Stdev     Max   +/- Stdev
[1]     Latency     1.05s   575.57ms   1.99s    57.89%
[1]     Req/Sec    10.05      3.94    30.00     81.66%
[1]   1152 requests in 30.03s, 172.12KB read
[1]   Socket errors: connect 0, read 0, write 0, timeout 1076
[1] Requests/sec:     38.36
[1] Transfer/sec:      5.73KB
[1] 
[1] 
[1] minimalApplicationWithLoom result with (virtual threads):
[1] Running 30s test @ http://localhost:8080/
[1]   4 threads and 100 connections
[1]   Thread Stats   Avg      Stdev     Max   +/- Stdev
[1]     Latency   106.11ms    2.74ms 126.59ms   77.47%
[1]     Req/Sec   239.19     36.74   252.00     92.68%
[1]   28100 requests in 30.03s, 4.10MB read
[1] Requests/sec:    935.74
[1] Transfer/sec:    139.81KB



```



Some design choices:
1.  Using MethodHandle/Reflect to make it compile on Java 8 too.
2. Using MethodHandle to name the virtual threads that are needed, JPMS
code can be added to open the can by default, but that is a little
over-killed, so better with an explicitly `--add-opens
java.base/java.lang=ALL-UNNAMED`
3. Add a virtualize or screen method to create a virtual thread executor
from a **platform** thread pool, this is useful, especially if you want
to limit the underlying queue size, FJP is unbounded.
4. Users can override the `handleExecutor` directly too.

Author: He-Pin

.github/workflows/actions.yml

@@ -16,7 +16,7 @@ jobs:
     strategy:
       matrix:
         java: [ '11', '17', '21' ]
-    name: Tests for Java ${{ matrix.Java }}
+    name: Tests local for Java ${{ matrix.Java }}
     steps:
       - uses: actions/checkout@v3
       - name: Setup java
@@ -27,14 +27,18 @@ jobs:
       - name: Run tests
         run: |
           set -eux
-          ./mill -ikj1 --disable-ticker __.testLocal
+          if [ "${{ matrix.java }}" == "21" ]; then
+             JAVA_OPTS='--add-opens java.base/java.lang=ALL-UNNAMED -Dcask.virtual-threads.enabled=true' ./mill -ikj1 --disable-ticker __.testLocal
+          else
+            ./mill -ikj1 --disable-ticker __.testLocal
+          fi
 
   test-examples:
     runs-on: ubuntu-latest
     strategy:
       matrix:
         java: [ '11', '17', '21' ]
-    name: Tests for Java ${{ matrix.Java }}
+    name: Tests examples for Java ${{ matrix.Java }}
     steps:
       - uses: actions/checkout@v3
       - name: Setup java
@@ -45,8 +49,13 @@ jobs:
       - name: Run tests
         run: |
           set -eux
-          ./mill __.publishLocal
-          ./mill -ikj1 --disable-ticker testExamples
+            if [ "${{ matrix.java }}" == "21" ]; then
+                ./mill __.publishLocal
+                JAVA_OPTS='--add-opens java.base/java.lang=ALL-UNNAMED -Dcask.virtual-threads.enabled=true' ./mill -ikj1 --disable-ticker testExamples
+            else
+                ./mill __.publishLocal
+                ./mill -ikj1 --disable-ticker testExamples
+            fi
 
   publish-sonatype:
     if: github.repository == 'com-lihaoyi/cask' && contains(github.ref, 'refs/tags/')

build.mill

@@ -85,6 +85,89 @@ object cask extends Cross[CaskMainModule](scalaVersions) {
   }
 }
 
+trait BenchmarkModule extends CrossScalaModule {
+  def moduleDeps = Seq(cask(crossScalaVersion))
+  def ivyDeps = Agg[Dep](
+  )
+}
+
+object benchmark extends Cross[BenchmarkModule](build.scalaVersions) with RunModule {
+
+  def waitForServer(url: String, maxAttempts: Int = 120): Boolean = {
+    (1 to maxAttempts).exists { attempt =>
+      try {
+        Thread.sleep(3000)
+        println("Checking server... Attempt " + attempt)
+        os.proc("curl", "-s", "-o", "/dev/null", "-w", "%{http_code}", url)
+          .call(check = false)
+          .exitCode == 0
+      } catch {
+        case _: Throwable =>
+          Thread.sleep(3000)
+          false
+      }
+    }
+  }
+
+  def runBenchMark(projectRoot: os.Path, example: String, vt: Boolean) = {
+    def runMillBackground(example: String, vt: Boolean) = {
+      println(s"Running $example with vt: $vt")
+      println("projectRoot: " + projectRoot)
+      os.proc(
+          "mill",
+          s"example.$example.app[$scala213].run")
+        .spawn(
+          cwd = projectRoot,
+          env = Map("CASK_VIRTUAL_THREAD" -> vt.toString),
+          stdout = os.Inherit,
+          stderr = os.Inherit)
+    }
+
+    val duration = "30s"
+    val threads = "4"
+    val connections = "100"
+    val url = "http://localhost:8080/"
+    val serverApp = runMillBackground(example, vt)
+
+    println(s"Waiting for server to start..., vt:$vt")
+    if (!waitForServer(url)) {
+      serverApp.destroy()
+      println("Failed to start server")
+      sys.exit(1)
+    }
+
+    val results = os.proc("wrk",
+      "-t", threads,
+      "-c", connections,
+      "-d", duration,
+      url
+    ).call(stderr = os.Pipe)
+    serverApp.destroyForcibly()
+    Thread.sleep(1000)
+
+    println(s"""\n$example result with ${if (vt) "(virtual threads)" else "(platform threads)"}:""")
+    println(results.out.text())
+  }
+
+  def runBenchmarks() = T.command {
+    val projectRoot = T.workspace
+    if (os.proc("which", "wrk").call(check = false).exitCode != 0) {
+      println("Error: wrk is not installed. Please install wrk first.")
+      sys.exit(1)
+    }
+    for (example <- Seq(
+      "staticFilesWithLoom",
+      "todoDbWithLoom",
+      "minimalApplicationWithLoom")) {
+      println(s"target server started, starting run benchmark with wrk for :$example with VT:false")
+      runBenchMark(projectRoot, example, vt = false)
+      println(s"target server started, starting run benchmark with wrk for :$example with VT:true")
+      runBenchMark(projectRoot, example, vt = true)
+    }
+
+  }
+}
+
 trait LocalModule extends CrossScalaModule{
   override def millSourcePath = super.millSourcePath / "app"
   def moduleDeps = Seq(cask(crossScalaVersion))
@@ -111,13 +194,16 @@ def zippedExamples = T {
     build.example.httpMethods.millSourcePath,
     build.example.minimalApplication.millSourcePath,
     build.example.minimalApplication2.millSourcePath,
+    build.example.minimalApplicationWithLoom.millSourcePath,
     build.example.redirectAbort.millSourcePath,
     build.example.scalatags.millSourcePath,
     build.example.staticFiles.millSourcePath,
+    build.example.staticFilesWithLoom.millSourcePath,
     build.example.staticFiles2.millSourcePath,
     build.example.todo.millSourcePath,
     build.example.todoApi.millSourcePath,
     build.example.todoDb.millSourcePath,
+    build.example.todoDbWithLoom.millSourcePath,
     build.example.twirl.millSourcePath,
     build.example.variableRoutes.millSourcePath,
     build.example.queryParams.millSourcePath,

cask/src/cask/internal/ThreadBlockingHandler.scala

@@ -0,0 +1,18 @@
+package cask.internal
+
+import io.undertow.server.{HttpHandler, HttpServerExchange}
+
+import java.util.concurrent.Executor
+
+/**
+ * A handler that dispatches the request to the given handler using the given executor.
+ * */
+final class ThreadBlockingHandler(executor: Executor, handler: HttpHandler) extends HttpHandler {
+  require(executor ne null, "Executor should not be null")
+  require(handler ne null, "Handler should not be null")
+
+  def handleRequest(exchange: HttpServerExchange): Unit = {
+    exchange.startBlocking()
+    exchange.dispatch(executor, handler)
+  }
+}

cask/src/cask/internal/Util.scala

@@ -1,24 +1,121 @@
 package cask.internal
 
 import java.io.{InputStream, PrintWriter, StringWriter}
-
 import scala.collection.generic.CanBuildFrom
 import scala.collection.mutable
 import java.io.OutputStream
-
+import java.lang.invoke.{MethodHandles, MethodType}
+import java.util.concurrent.{Executor, ExecutorService, ForkJoinPool, ThreadFactory}
 import scala.annotation.switch
 import scala.concurrent.{ExecutionContext, Future, Promise}
+import scala.util.Try
+import scala.util.control.NonFatal
 
 object Util {
+  private val lookup = MethodHandles.lookup()
+
+  import cask.util.Logger.Console.globalLogger
+
+  /**
+   * Create a virtual thread executor with the given executor as the scheduler.
+   * */
+  def createVirtualThreadExecutor(executor: Executor): Option[ExecutorService] = {
+    (for {
+      factory <- Try(createVirtualThreadFactory("cask-handler-executor", executor))
+      executor <- Try(createNewThreadPerTaskExecutor(factory))
+    } yield executor).toOption
+  }
+
+  /**
+   * Create a default cask virtual thread executor if possible.
+   * */
+  def createDefaultCaskVirtualThreadExecutor: Option[ExecutorService] = {
+    for {
+      scheduler <- getDefaultVirtualThreadScheduler
+      executor <- createVirtualThreadExecutor(scheduler)
+    } yield executor
+  }
+
+  /**
+   * Try to get the default virtual thread scheduler, or null if not supported.
+   * */
+  def getDefaultVirtualThreadScheduler: Option[ForkJoinPool] = {
+    try {
+      val virtualThreadClass = Class.forName("java.lang.VirtualThread")
+      val privateLookup = MethodHandles.privateLookupIn(virtualThreadClass, lookup)
+      val defaultSchedulerField = privateLookup.findStaticVarHandle(virtualThreadClass, "DEFAULT_SCHEDULER", classOf[ForkJoinPool])
+      Option(defaultSchedulerField.get().asInstanceOf[ForkJoinPool])
+    } catch {
+      case NonFatal(e) =>
+        //--add-opens java.base/java.lang=ALL-UNNAMED
+        globalLogger.exception(e)
+        None
+    }
+  }
+
+  def createNewThreadPerTaskExecutor(threadFactory: ThreadFactory): ExecutorService = {
+    try {
+      val executorsClazz = ClassLoader.getSystemClassLoader.loadClass("java.util.concurrent.Executors")
+      val newThreadPerTaskExecutorMethod = lookup.findStatic(
+        executorsClazz,
+        "newThreadPerTaskExecutor",
+        MethodType.methodType(classOf[ExecutorService], classOf[ThreadFactory]))
+      newThreadPerTaskExecutorMethod.invoke(threadFactory)
+        .asInstanceOf[ExecutorService]
+    } catch {
+      case NonFatal(e) =>
+        globalLogger.exception(e)
+        throw new UnsupportedOperationException("Failed to create newThreadPerTaskExecutor.", e)
+    }
+  }
+
+  /**
+   * Create a virtual thread factory with a executor, the executor will be used as the scheduler of
+   * virtual thread.
+   *
+   * The executor should run task on platform threads.
+   *
+   * returns null if not supported.
+   */
+  def createVirtualThreadFactory(prefix: String,
+                                 executor: Executor): ThreadFactory =
+    try {
+      val builderClass = ClassLoader.getSystemClassLoader.loadClass("java.lang.Thread$Builder")
+      val ofVirtualClass = ClassLoader.getSystemClassLoader.loadClass("java.lang.Thread$Builder$OfVirtual")
+      val ofVirtualMethod = lookup.findStatic(classOf[Thread], "ofVirtual", MethodType.methodType(ofVirtualClass))
+      var builder = ofVirtualMethod.invoke()
+      if (executor != null) {
+        val clazz = builder.getClass
+        val privateLookup = MethodHandles.privateLookupIn(
+          clazz,
+          lookup
+        )
+        val schedulerFieldSetter = privateLookup
+          .findSetter(clazz, "scheduler", classOf[Executor])
+        schedulerFieldSetter.invoke(builder, executor)
+      }
+      val nameMethod = lookup.findVirtual(ofVirtualClass, "name",
+        MethodType.methodType(ofVirtualClass, classOf[String], classOf[Long]))
+      val factoryMethod = lookup.findVirtual(builderClass, "factory", MethodType.methodType(classOf[ThreadFactory]))
+      builder = nameMethod.invoke(builder, prefix + "-virtual-thread-", 0L)
+      factoryMethod.invoke(builder).asInstanceOf[ThreadFactory]
+    } catch {
+      case NonFatal(e) =>
+        globalLogger.exception(e)
+        //--add-opens java.base/java.lang=ALL-UNNAMED
+        throw new UnsupportedOperationException("Failed to create virtual thread factory.", e)
+    }
+
   def firstFutureOf[T](futures: Seq[Future[T]])(implicit ec: ExecutionContext) = {
     val p = Promise[T]
     futures.foreach(_.foreach(p.trySuccess))
     p.future
   }
+
   /**
-    * Convert a string to a C&P-able literal. Basically
-    * copied verbatim from the uPickle source code.
-    */
+   * Convert a string to a C&P-able literal. Basically
+   * copied verbatim from the uPickle source code.
+   */
   def literalize(s: IndexedSeq[Char], unicode: Boolean = true) = {
     val sb = new StringBuilder
     sb.append('"')
@@ -47,29 +144,30 @@ object Util {
   def transferTo(in: InputStream, out: OutputStream) = {
     val buffer = new Array[Byte](8192)
 
-    while ({
-      in.read(buffer) match{
+    while ( {
+      in.read(buffer) match {
         case -1 => false
         case n =>
           out.write(buffer, 0, n)
           true
       }
     }) ()
   }
+
   def pluralize(s: String, n: Int) = {
     if (n == 1) s else s + "s"
   }
 
   /**
-    * Splits a string into path segments; automatically removes all
-    * leading/trailing slashes, and ignores empty path segments.
-    *
-    * Written imperatively for performance since it's used all over the place.
-    */
+   * Splits a string into path segments; automatically removes all
+   * leading/trailing slashes, and ignores empty path segments.
+   *
+   * Written imperatively for performance since it's used all over the place.
+   */
   def splitPath(p: String): collection.IndexedSeq[String] = {
     val pLength = p.length
     var i = 0
-    while(i < pLength && p(i) == '/') i += 1
+    while (i < pLength && p(i) == '/') i += 1
     var segmentStart = i
     val out = mutable.ArrayBuffer.empty[String]
 
@@ -81,7 +179,7 @@ object Util {
       segmentStart = i + 1
     }
 
-    while(i < pLength){
+    while (i < pLength) {
       if (p(i) == '/') complete()
       i += 1
     }
@@ -96,33 +194,35 @@ object Util {
     pw.flush()
     trace.toString
   }
+
   def softWrap(s: String, leftOffset: Int, maxWidth: Int) = {
     val oneLine = s.linesIterator.mkString(" ").split(' ')
 
     lazy val indent = " " * leftOffset
 
     val output = new StringBuilder(oneLine.head)
     var currentLineWidth = oneLine.head.length
-    for(chunk <- oneLine.tail){
+    for (chunk <- oneLine.tail) {
       val addedWidth = currentLineWidth + chunk.length + 1
-      if (addedWidth > maxWidth){
+      if (addedWidth > maxWidth) {
         output.append("\n" + indent)
         output.append(chunk)
         currentLineWidth = chunk.length
-      } else{
+      } else {
         currentLineWidth = addedWidth
         output.append(' ')
         output.append(chunk)
       }
     }
     output.mkString
   }
+
   def sequenceEither[A, B, M[X] <: TraversableOnce[X]](in: M[Either[A, B]])(
     implicit cbf: CanBuildFrom[M[Either[A, B]], B, M[B]]): Either[A, M[B]] = {
     in.foldLeft[Either[A, mutable.Builder[B, M[B]]]](Right(cbf(in))) {
-      case (acc, el) =>
-        for (a <- acc; e <- el) yield a += e
-    }
+        case (acc, el) =>
+          for (a <- acc; e <- el) yield a += e
+      }
       .map(_.result())
   }
 }