Netty server side cancellation #3256

kciesielski · 2023-10-19T18:50:50Z

This PR changes the way effects are executed "into a Future", which is currently done by CE3 Dispatcher or ZIO Runtime. The change uses them differently, so that we get a cancellation callback, which can be later run by NettyServerHandler in order to attempt cancellation of long running requests when client disconnects.
For raw Netty Future server, the cancellation callback is a noop, I guess there's no way to cancel standard Scala Futures without adding quite some complexity (see https://viktorklang.com/blog/Futures-in-Scala-protips-6.html).

Based on https://github.com/http4s/http4s-netty/pull/396/files

adamw · 2023-10-20T12:27:01Z

server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyServerHandler.scala


-class NettyServerHandler[F[_]](route: Route[F], unsafeRunAsync: (() => F[Unit]) => Unit, maxContentLength: Option[Int])(implicit
-    me: MonadError[F]
+class NettyServerHandler[F[_]](route: Route[F], unsafeRunAsync: (() => F[Unit]) => (() => Future[Unit]), maxContentLength: Option[Int])(


I'd add a comment that unsafeRunAsync returns a function to cancel the execution

Comment added, and signature extended, it's even more complex now ;)

adamw · 2023-10-20T12:28:38Z

server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyServerHandler.scala

-              .ensure(me.eval(req.release()))
+              .ensure {
+                me.eval {
+                  pendingResponses.dequeue()


doesn't this make the assumption that the requests will end in the same order, in which they started?

They are requests from the same channel, so maybe that's OK. Anyway, in http4s implementation there additional elements like eventLoopContext and pendingResponses. I think they are crucial to make sure this ordering works as expected, so I'll study them further.

adamw · 2023-10-20T12:29:40Z

server/netty-server/zio/src/main/scala/sttp/tapir/server/netty/zio/NettyZioServer.scala

@@ -67,7 +74,7 @@ case class NettyZioServer[R](routes: Vector[RIO[R, Route[RIO[R, *]]]], options:
        config,
        new NettyServerHandler[RIO[R, *]](
          route,
-          (f: () => RIO[R, Unit]) => Unsafe.unsafe(implicit u => runtime.unsafe.runToFuture(f())),
+          unsafeRunAsync(runtime),


maybe we can add a test for either ZIO/cats, that a long-running request is indeed cancelled (sth similar was present in the http4s PR)

adamw · 2023-10-20T12:34:39Z

server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyServerHandler.scala

+    if (ctx.channel.isActive) {
+      initHandler(ctx)
+    }
+  override def channelActive(ctx: ChannelHandlerContext): Unit = initHandler(ctx)


is it so that either channelActive OR handlerAdded is called? won't both be called?

Indeed sometimes both can be called, but now I've added a check in initHandler which will prevent double listener registration.

adamw · 2023-10-20T12:35:32Z

server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyServerHandler.scala

+
+  override def channelReadComplete(ctx: ChannelHandlerContext): Unit = {
+    logger.trace(s"channelReadComplete: ctx = $ctx")
+    // The normal response to read complete is to issue another read,


hm is that needed for the pendingResponses logic?

These ctx.read() calls are IMO unnecessary. For some reason http4s does AUTO_READ=false, but I don't see this interferring with our implementation. Reactive streams also work as expected.

- Might be needed for slow CI

adamw · 2023-10-26T09:28:53Z

server/tests/src/main/scala/sttp/tapir/server/tests/ServerCancellationTests.scala

+      endpoint
+        .out(plainBody[String])
+        .serverLogic { _ =>
+          (m.eval(canceledSemaphore.acquire())) >> (async.sleep(15.seconds) >> pureResult("processing finished".asRight[Unit]))


I think I don't understand the async.sleep - we're waiting on the semaphore, and releasing it only after the cancel signal comes in - do we need to still sleep after that?

The semaphore will be acquired immediately here, the sleep is to simulate long processing. The semaphore is used to block the test code until backend's onCancel finishes setting the boolean.

ah I mixed up the release/aquire directions ;) yeah it's fine of course :)

adamw · 2023-10-26T09:29:46Z

server/tests/src/main/scala/sttp/tapir/server/tests/ServerCancellationTests.scala

+          case _: SttpClientException.TimeoutException => // expected, this is how we trigged client-side cancellation
+            IO(
+              assert(
+                canceledSemaphore.tryAcquire(30L, TimeUnit.SECONDS),


I don't get this either, the sempahore was released & acquired in the server/cancellation logic, why aquire it here again?>

adamw · 2023-10-26T09:31:13Z

server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyServerHandler.scala

    me: MonadError[F]
 ) extends SimpleChannelInboundHandler[HttpRequest] {

+  // By using the Netty event loop assigned to this channel we get two benefits:


let's maybe mention here that this is copied from http4's code, just to maintain proper attribution :)

adamw · 2023-10-26T09:32:20Z

server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyServerHandler.scala

+  // We keep track of the cancellation tokens for all the requests in flight. This gives us
+  // observability into the number of requests in flight and the ability to cancel them all
+  // if the connection gets closed.
+  private[this] val pendingResponses = MutableQueue.empty[() => Future[Unit]]


did you manage to drill down and understand why this is a queue - can you have multiple ongoing requests? is http 1 only?

If I understand correctly this is HTTP2, so netty can ingest a request, dispatch async processing to another thread pool, and pick up next request. The responses will be returned in order though, even if request 2 is finished before request 1. https://medium.com/@akhaku/netty-data-model-threading-and-gotchas-cab820e4815a

Edit: Turns out HTTP2 multiplexing is a different beast, which requires special setup of Netty, even an additional library (netty-codec-http2). It's powerful, because it allows opening a single connection and sending multiple requests and getting responses in any order, the protocol should take care of this. However, it's something different than what our server is capable of.

With this code, we are implementing support for cancellation in HTTP 1.1 pipelining: where a client can send multiple requests without waiting for the response to the first, and the server will process and respond to them in order. One of the main challenges is that responses must be returned in order, which can introduce head-of-line blocking if processing one request takes longer than others. HTTP/2 addresses these issues by introducing mentioned multiplexing.

adamw · 2023-10-26T09:35:59Z

server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyServerHandler.scala

+            pendingResponses.dequeue()
+            try {
+              handleResponse(ctx, req, serverResponse)
+              releaseReq()


don't we have to run this in case of exceptions as well? the .ensure in the old version alway released it

adamw · 2023-10-26T09:39:38Z

server/tests/src/main/scala/sttp/tapir/server/tests/ServerCancellationTests.scala

+import java.util.concurrent.{Semaphore, TimeUnit}
+import scala.concurrent.duration._
+
+class ServerCancellationTests[F[_], OPTIONS, ROUTE](createServerTest: CreateServerTest[F, Any, OPTIONS, ROUTE])(implicit


nice test :) I wonder which other backends support cancellation in this way ... but let's leave this for another task ;)

also, this will be interesting to integrate into tapir-loom, but that's also another story

kciesielski added 3 commits October 18, 2023 17:17

Support for server-side request cancellation (CE3, Future)

b265784

Add cancellation support to Netty ZIO

a7a5669

Adjust to Scala 2

d77a527

kciesielski added the Netty label Oct 19, 2023

kciesielski requested a review from adamw October 19, 2023 18:50

adamw reviewed Oct 20, 2023

View reviewed changes

kciesielski added 14 commits October 23, 2023 16:47

Fix handling of correct request/response ordering

64ff166

Remove custom implementation of channelReadComplete

c1ef9b4

Organize imports

239068e

Add tests for cancellation (CE3)

ef26db4

wip

8b047e5

Fix cancellation test

120a2da

Remove problematic redundant imports

94d2b3f

More refactoring and test tweaks

c877bc6

Fix warning for unused param

68b8684

Extend timeout when waiting for semaphore

0c7956f

- Might be needed for slow CI

Keep ordering responses even if there were failures

0bd6f49

Ensure that stream completes before client gives up

ffa9b61

Use readTimeout to ensure cancellation is triggered on JDK11

883698c

Remove dev stuff

4790c80

kciesielski requested a review from adamw October 26, 2023 09:08

adamw reviewed Oct 26, 2023

View reviewed changes

kciesielski added 2 commits October 26, 2023 13:17

Add proper attribution to http4s source code

995e6e0

Ensure that request release is called on errors

bd8b94a

adamw merged commit c969f0f into master Oct 31, 2023
13 checks passed

mergify bot deleted the netty-server-side-cancel branch October 31, 2023 10:11

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Netty server side cancellation #3256

Netty server side cancellation #3256

kciesielski commented Oct 19, 2023 •

edited

Loading

adamw Oct 20, 2023

kciesielski Oct 23, 2023

adamw Oct 20, 2023

kciesielski Oct 23, 2023

adamw Oct 20, 2023

adamw Oct 20, 2023

kciesielski Oct 23, 2023

adamw Oct 20, 2023

kciesielski Oct 23, 2023

adamw Oct 26, 2023

kciesielski Oct 26, 2023

adamw Oct 26, 2023

adamw Oct 26, 2023

adamw Oct 26, 2023

adamw Oct 26, 2023

kciesielski Oct 26, 2023

kciesielski Oct 27, 2023

adamw Oct 26, 2023

adamw Oct 26, 2023

adamw Oct 26, 2023

Netty server side cancellation #3256

Netty server side cancellation #3256

Conversation

kciesielski commented Oct 19, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kciesielski commented Oct 19, 2023 •

edited

Loading