KTOR-3455 Fix reading request body backpressure in Netty (#3361)

ktorio · Jan 16, 2023 · 77fbd34 · 77fbd34
1 parent b3c696e
commit 77fbd34
Showing 1 changed file with 6 additions and 1 deletion.
diff --git a/ktor-server/ktor-server-netty/jvm/src/io/ktor/server/netty/cio/RequestBodyHandler.kt b/ktor-server/ktor-server-netty/jvm/src/io/ktor/server/netty/cio/RequestBodyHandler.kt
@@ -9,6 +9,7 @@ import io.netty.buffer.*
 import io.netty.channel.*
 import io.netty.handler.codec.http.*
 import io.netty.util.*
+import kotlinx.atomicfu.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel
 import java.lang.Integer.*
@@ -18,6 +19,7 @@ internal class RequestBodyHandler(
     val context: ChannelHandlerContext
 ) : ChannelInboundHandlerAdapter(), CoroutineScope {
     private val handlerJob = CompletableDeferred<Nothing>()
+    private val buffersInProcessingCount = atomic(0)
 
     private val queue = Channel<Any>(Channel.UNLIMITED)
 
@@ -137,7 +139,9 @@ internal class RequestBodyHandler(
     }
 
     private fun requestMoreEvents() {
-        context.read()
+        if (buffersInProcessingCount.decrementAndGet() == 0) {
+            context.read()
+        }
     }
 
     @OptIn(ExperimentalCoroutinesApi::class)
@@ -169,6 +173,7 @@ internal class RequestBodyHandler(
     }
 
     private fun handleBytesRead(content: ReferenceCounted) {
+        buffersInProcessingCount.incrementAndGet()
         if (!queue.trySend(content).isSuccess) {
             content.release()
             throw IllegalStateException("Unable to process received buffer: queue offer failed")