Use the current Ox scope when running a Netty sync-server to create threads (forks) (#4752)

Closes #4747

Author: adamw

perf-tests/README.md

@@ -12,9 +12,9 @@ perfTests/runMain sttp.tapir.perf.apis.ServerRunner http4s.TapirMulti
 Run it without server name to see a list of all available names. 
 Exception: If you're testing `NettySyncServer` (tapir-server-netty-sync), its server runner is located elsewhere:
 ```
-nettyServerSync3/Test/runMain sttp.tapir.netty.sync.perf.NettySyncServerRunner
+nettyServerSync3/Test/runMain sttp.tapir.server.netty.sync.perf.NettySyncServerRunner
 ```
-This is caused by `perf-tests` using Scala 2.13 forced by Gatling, while `NettySyncServer` is written excluisively for Scala 3.
+This is caused by `perf-tests` using Scala 2.13 forced by Gatling, while `NettySyncServer` is written exclusively for Scala 3.
 
 ## Configuring and running simulations
 
@@ -37,7 +37,7 @@ If not set, default values will be used (see `sttp.tapir.perf.CommonSimulations`
 
 ## Profiling 
 
-To atach the profiler to a running server, it is recommended to use [async-profiler](https://github.com/async-profiler/async-profiler).
+To attach the profiler to a running server, it is recommended to use [async-profiler](https://github.com/async-profiler/async-profiler).
 Start the profiler by calling:
 ```
 asprof -e cpu,alloc,lock -f profile.jfr <PID>

server/netty-server/sync/src/main/scala/sttp/tapir/server/netty/sync/NettySyncServer.scala

@@ -16,18 +16,17 @@ import sttp.tapir.server.netty.{NettyConfig, NettyResponse, Route}
 import java.net.{InetSocketAddress, SocketAddress}
 import java.nio.file.Path
 import java.util.concurrent.atomic.AtomicBoolean
-import java.util.concurrent.{Executors, Future as JFuture}
 import scala.concurrent.duration.FiniteDuration
 import scala.concurrent.{Future, Promise}
 import scala.util.control.NonFatal
+import java.util.concurrent.atomic.AtomicReference
 
 case class NettySyncServer(
     serverEndpoints: Vector[ServerEndpoint[OxStreams & WebSockets, Identity]],
     otherRoutes: Vector[IdRoute],
     options: NettySyncServerOptions,
     config: NettyConfig
 ):
-  private val executor = Executors.newVirtualThreadPerTaskExecutor()
   private val logger = LoggerFactory.getLogger(getClass.getName)
 
   def addEndpoint(se: ServerEndpoint[OxStreams & WebSockets, Identity]): NettySyncServer = addEndpoints(List(se))
@@ -93,11 +92,6 @@ case class NettySyncServer(
       never
     }
 
-  private[netty] def start(route: Route[Identity]): NettySyncServerBinding =
-    startUsingSocketOverride[InetSocketAddress](route, None) match
-      case (socket, stop) =>
-        NettySyncServerBinding(socket, stop)
-
   def startUsingDomainSocket(path: Path)(using Ox): NettySyncDomainSocketBinding =
     startUsingSocketOverride(Some(new DomainSocketAddress(path.toFile)), inScopeRunner()) match
       case (socket, stop) =>
@@ -109,28 +103,52 @@ case class NettySyncServer(
   ): (SA, () => Unit) =
     val endpointRoute = NettySyncServerInterpreter(options).toRoute(serverEndpoints.toList, inScopeRunner)
     val route = Route.combine(endpointRoute +: otherRoutes)
-    startUsingSocketOverride(route, socketOverride)
+    startUsingSocketOverride(route, socketOverride, inScopeRunner)
 
-  private def startUsingSocketOverride[SA <: SocketAddress](route: Route[Identity], socketOverride: Option[SA]): (SA, () => Unit) =
+  private def startUsingSocketOverride[SA <: SocketAddress](
+      route: Route[Identity],
+      socketOverride: Option[SA],
+      inScopeRunner: InScopeRunner
+  ): (SA, () => Unit) =
     val eventLoopGroup = config.eventLoopConfig.initEventLoopGroup()
 
     def unsafeRunF(
         callToExecute: () => Identity[ServerResponse[NettyResponse]]
     ): (Future[ServerResponse[NettyResponse]], () => Future[Unit]) =
       val scalaPromise = Promise[ServerResponse[NettyResponse]]()
-      val jFuture: JFuture[?] = executor.submit(new Runnable {
-        override def run(): Unit = try {
-          val result = callToExecute()
-          scalaPromise.success(result)
-        } catch {
-          case NonFatal(e) => scalaPromise.failure(e)
-        }
-      })
+
+      // used to cancel the fork, possible states:
+      // - Cancelled: needed if cancellation happens after the fork starts, but before it is set in the atomic reference
+      // - CancellableFork: used when cancellation using interruption is enabled
+      // - None: means that the fork was not set yet
+      object Cancelled
+      val runningFork = new AtomicReference[None.type | CancellableFork[Unit] | Cancelled.type](None)
+      // #4747: we are creating forks using the concurrency scope within which the netty server is running
+      // however, this is called on a Netty-managed thread, hence we need to use the inScopeRunner
+      inScopeRunner.async {
+        def run(): Unit =
+          if runningFork.get() != Cancelled then
+            try scalaPromise.success(callToExecute())
+            catch case NonFatal(e) => scalaPromise.failure(e)
+
+        if options.interruptServerLogicWhenRequestCancelled then
+          val forked = forkCancellable(run())
+          // we only update the state if it's not cancelled already
+          runningFork.getAndAccumulate(forked, (cur, giv) => if cur != Cancelled then giv else cur) match {
+            case None                     => // common "happy path" case
+            case _: CancellableFork[Unit] => throw new IllegalStateException("Another fork was already set")
+            case Cancelled                => forked.cancelNow() // cancellation happened before the fork was set
+          }
+        else forkDiscard(run())
+      }
 
       (
         scalaPromise.future,
         () => {
-          jFuture.cancel(options.interruptServerLogicWhenRequestCancelled)
+          runningFork.getAndSet(Cancelled) match {
+            case fork: CancellableFork[Unit] => fork.cancelNow()
+            case _                           => // skip - fork not yet set
+          }
           Future.unit
         }
       )

server/netty-server/sync/src/test/scala/sttp/tapir/server/netty/sync/NettySyncTestServerInterpreter.scala

@@ -32,15 +32,7 @@ class NettySyncTestServerInterpreter(eventLoopGroup: NioEventLoopGroup)
   override def server(
       routes: NonEmptyList[IdRoute],
       gracefulShutdownTimeout: Option[FiniteDuration] = None
-  ): Resource[IO, Port] = {
-    val config =
-      NettyConfig.default.eventLoopGroup(eventLoopGroup).randomPort.withDontShutdownEventLoopGroupOnClose.noGracefulShutdown
-    val customizedConfig = gracefulShutdownTimeout.map(config.withGracefulShutdownTimeout).getOrElse(config)
-    val options = NettySyncServerOptions.default
-    val bind = IO.blocking(NettySyncServer(options, customizedConfig).start(Route.combine(routes.toList)))
-
-    Resource.make(bind)(server => IO.blocking(server.stop())).map(_.port)
-  }
+  ): Resource[IO, Port] = throw new UnsupportedOperationException // instead, scoped* variants are used by NettySyncCreateServerTest
 
   def scopedServerWithRoutesStop(
       routes: NonEmptyList[IdRoute],
@@ -50,7 +42,7 @@ class NettySyncTestServerInterpreter(eventLoopGroup: NioEventLoopGroup)
       NettyConfig.default.eventLoopGroup(eventLoopGroup).randomPort.withDontShutdownEventLoopGroupOnClose.noGracefulShutdown
     val customizedConfig = gracefulShutdownTimeout.map(config.withGracefulShutdownTimeout).getOrElse(config)
     val options = NettySyncServerOptions.default
-    useInScope(NettySyncServer(options, customizedConfig).start(Route.combine(routes.toList)))(_.stop())
+    useInScope(NettySyncServer(options, customizedConfig).addRoute(Route.combine(routes.toList)).start())(_.stop())
 
   def scopedServerWithInterceptorsStop(
       endpoint: ServerEndpoint[OxStreams with WebSockets, Identity],

server/netty-server/sync/src/test/scala/sttp/tapir/server/netty/sync/perf/NettySyncServerRunner.scala

@@ -74,7 +74,7 @@ object NettySyncServerRunner {
     )
   val wsServerEndpoint = wsEndpoint.handleSuccess(_ => wsPipe)
 
-  val endpoints = genEndpointsId(1)
+  val endpoints = genEndpointsId(128)
 
   def main(args: Array[String]): Unit = {
     val declaredPort = 8080