KTOR-5199 re-run multi perform loop to schedule requests

ktorio · Dec 19, 2024 · f996e9d · f996e9d
1 parent eb46722
commit f996e9d
Show file tree

Hide file tree

Showing 3 changed files with 53 additions and 5 deletions.
diff --git a/ktor-client/ktor-client-curl/desktop/src/io/ktor/client/engine/curl/CurlProcessor.kt b/ktor-client/ktor-client-curl/desktop/src/io/ktor/client/engine/curl/CurlProcessor.kt
@@ -18,6 +18,12 @@ internal class RequestContainer(
     val completionHandler: CompletableDeferred<CurlSuccess>
 )
 
+/**
+ * A class responsible for processing requests asynchronously.
+ *
+ * It holds a dispatcher interacting with curl multi interface API,
+ * which requires API calls from single thread.
+ */
 internal class CurlProcessor(coroutineContext: CoroutineContext) {
     @OptIn(InternalAPI::class)
     private val curlDispatcher: CloseableCoroutineDispatcher =
@@ -28,6 +34,7 @@ internal class CurlProcessor(coroutineContext: CoroutineContext) {
 
     private val curlScope = CoroutineScope(coroutineContext + curlDispatcher)
     private val requestQueue: Channel<RequestContainer> = Channel(Channel.UNLIMITED)
+    private val requestCounter = atomic(0L)
     private val curlProtocols by lazy { getCurlProtocols() }
 
     init {
@@ -48,8 +55,9 @@ internal class CurlProcessor(coroutineContext: CoroutineContext) {
         }
 
         val result = CompletableDeferred<CurlSuccess>()
-        requestQueue.send(RequestContainer(request, result))
-        curlApi!!.wakeup()
+        nextRequest {
+            requestQueue.send(RequestContainer(request, result))
+        }
         return result.await()
     }
 
@@ -59,7 +67,7 @@ internal class CurlProcessor(coroutineContext: CoroutineContext) {
             val api = curlApi!!
             while (!requestQueue.isClosedForReceive) {
                 drainRequestQueue(api)
-                api.perform()
+                api.perform(requestCounter)
             }
         }
     }
@@ -91,6 +99,8 @@ internal class CurlProcessor(coroutineContext: CoroutineContext) {
         if (!closed.compareAndSet(false, true)) return
 
         requestQueue.close()
+        nextRequest()
+
         GlobalScope.launch(curlDispatcher) {
             curlScope.coroutineContext[Job]!!.join()
             curlApi!!.close()
@@ -105,4 +115,10 @@ internal class CurlProcessor(coroutineContext: CoroutineContext) {
             curlApi!!.cancelRequest(easyHandle, cause)
         }
     }
+
+    private inline fun nextRequest(body: (Long) -> Unit = {}) = try {
+        body(requestCounter.incrementAndGet())
+    } finally {
+        curlApi!!.wakeup()
+    }
 }
diff --git a/...t/ktor-client-curl/desktop/src/io/ktor/client/engine/curl/internal/CurlMultiApiHandler.kt b/...t/ktor-client-curl/desktop/src/io/ktor/client/engine/curl/internal/CurlMultiApiHandler.kt
@@ -8,6 +8,7 @@ import io.ktor.client.plugins.*
 import io.ktor.utils.io.*
 import io.ktor.utils.io.core.*
 import io.ktor.utils.io.locks.*
+import kotlinx.atomicfu.*
 import kotlinx.cinterop.*
 import kotlinx.coroutines.*
 import kotlinx.io.*
@@ -25,6 +26,11 @@ private class RequestHolder @OptIn(ExperimentalForeignApi::class) constructor(
     }
 }
 
+/**
+ * Handles requests using libcurl with multi interface.
+ *
+ * @see <a href="https://curl.se/libcurl/c/libcurl-multi.html">Multi interface overview</a>
+ */
 @OptIn(InternalAPI::class)
 internal class CurlMultiApiHandler : Closeable {
     @OptIn(ExperimentalForeignApi::class)
@@ -130,11 +136,12 @@ internal class CurlMultiApiHandler : Closeable {
     }
 
     @OptIn(ExperimentalForeignApi::class)
-    internal fun perform() {
+    internal fun perform(counter: AtomicLong) {
         if (activeHandles.isEmpty()) return
 
         memScoped {
             val transfersRunning = alloc<IntVar>()
+            val requestId = counter.value
             do {
                 synchronized(easyHandlesToUnpauseLock) {
                     var handle = easyHandlesToUnpause.removeFirstOrNull()
@@ -150,7 +157,7 @@ internal class CurlMultiApiHandler : Closeable {
                 if (transfersRunning.value < activeHandles.size) {
                     handleCompleted()
                 }
-            } while (transfersRunning.value != 0)
+            } while (transfersRunning.value != 0 && requestId == counter.value)
         }
     }
 

diff --git a/...lient/ktor-client-curl/desktop/test/io/ktor/client/engine/curl/test/CurlWebSocketTests.kt b/...lient/ktor-client-curl/desktop/test/io/ktor/client/engine/curl/test/CurlWebSocketTests.kt
@@ -7,13 +7,15 @@ package io.ktor.client.engine.curl.test
 import io.ktor.client.*
 import io.ktor.client.engine.curl.*
 import io.ktor.client.plugins.websocket.*
+import io.ktor.client.request.*
 import io.ktor.websocket.*
 import kotlinx.coroutines.*
 import kotlinx.serialization.json.*
 import kotlin.test.*
 
 class CurlWebSocketTests {
 
+    private val TEST_SERVER: String = "http://127.0.0.1:8080"
     private val TEST_WEBSOCKET_SERVER: String = "ws://127.0.0.1:8080"
 
     @Test
@@ -75,4 +77,27 @@ class CurlWebSocketTests {
             }
         }
     }
+
+    @Test
+    fun testParallelSessions() {
+        val client = HttpClient(Curl) {
+            install(WebSockets)
+        }
+
+        runBlocking {
+            val websocketInitialized = CompletableDeferred<Boolean>()
+
+            launch {
+                client.webSocket("$TEST_WEBSOCKET_SERVER/websockets/echo") {
+                    websocketInitialized.complete(true)
+                    delay(20)
+                }
+            }
+
+            websocketInitialized.await()
+
+            val response = client.get(TEST_SERVER)
+            assertEquals(200, response.status.value)
+        }
+    }
 }