KTOR-7729 Add logging for tracking concurrent access in the dev mode (#…

…4477) * KTOR-7729 Add logging for tracking concurrent access in the dev mode
ktorio · Nov 14, 2024 · 2bbee41 · 2bbee41
1 parent abc423f
commit 2bbee41
Show file tree

Hide file tree

Showing 6 changed files with 61 additions and 4 deletions.
diff --git a/ktor-io/api/ktor-io.api b/ktor-io/api/ktor-io.api
@@ -174,7 +174,8 @@ public abstract interface class io/ktor/utils/io/ChannelJob {
 }
 
 public final class io/ktor/utils/io/ConcurrentIOException : java/lang/IllegalStateException {
-	public fun <init> (Ljava/lang/String;)V
+	public fun <init> (Ljava/lang/String;Ljava/lang/Throwable;)V
+	public synthetic fun <init> (Ljava/lang/String;Ljava/lang/Throwable;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
 }
 
 public final class io/ktor/utils/io/CountedByteReadChannel : io/ktor/utils/io/ByteReadChannel {

diff --git a/ktor-io/api/ktor-io.klib.api b/ktor-io/api/ktor-io.klib.api
@@ -212,7 +212,7 @@ final class io.ktor.utils.io/ByteChannel : io.ktor.utils.io/BufferedByteWriteCha
 }
 
 final class io.ktor.utils.io/ConcurrentIOException : kotlin/IllegalStateException { // io.ktor.utils.io/ConcurrentIOException|null[0]
-    constructor <init>(kotlin/String) // io.ktor.utils.io/ConcurrentIOException.<init>|<init>(kotlin.String){}[0]
+    constructor <init>(kotlin/String, kotlin/Throwable? = ...) // io.ktor.utils.io/ConcurrentIOException.<init>|<init>(kotlin.String;kotlin.Throwable?){}[0]
 }
 
 final class io.ktor.utils.io/CountedByteReadChannel : io.ktor.utils.io/ByteReadChannel { // io.ktor.utils.io/CountedByteReadChannel|null[0]

diff --git a/ktor-io/common/src/io/ktor/utils/io/ByteChannel.kt b/ktor-io/common/src/io/ktor/utils/io/ByteChannel.kt
@@ -12,6 +12,7 @@ import kotlin.concurrent.Volatile
 import kotlin.coroutines.*
 import kotlin.jvm.*
 
+internal expect val DEVELOPMENT_MODE: Boolean
 internal const val CHANNEL_MAX_SIZE: Int = 1024 * 1024
 
 /**
@@ -189,13 +190,16 @@ public class ByteChannel(public val autoFlush: Boolean = false) : ByteReadChanne
         // Resume the previous task
         when (previous) {
             is TaskType ->
-                previous.resume(ConcurrentIOException(slot.taskName()))
+                previous.resume(ConcurrentIOException(slot.taskName(), previous.created))
+
             is Slot.Task ->
                 previous.resume()
+
             is Slot.Closed -> {
                 slot.resume(previous.cause)
                 return
             }
+
             Slot.Empty -> {}
         }
 
@@ -219,6 +223,8 @@ public class ByteChannel(public val autoFlush: Boolean = false) : ByteReadChanne
         data class Closed(val cause: Throwable?) : Slot
 
         sealed interface Task : Slot {
+            val created: Throwable?
+
             val continuation: Continuation<Unit>
 
             fun taskName(): String
@@ -231,10 +237,30 @@ public class ByteChannel(public val autoFlush: Boolean = false) : ByteReadChanne
         }
 
         class Read(override val continuation: Continuation<Unit>) : Task {
+            override var created: Throwable? = null
+
+            init {
+                if (DEVELOPMENT_MODE) {
+                    created = Throwable("ReadTask 0x${continuation.hashCode().toString(16)}").also {
+                        it.stackTraceToString()
+                    }
+                }
+            }
+
             override fun taskName(): String = "read"
         }
 
         class Write(override val continuation: Continuation<Unit>) : Task {
+            override var created: Throwable? = null
+
+            init {
+                if (DEVELOPMENT_MODE) {
+                    created = Throwable("WriteTask 0x${continuation.hashCode().toString(16)}").also {
+                        it.stackTraceToString()
+                    }
+                }
+            }
+
             override fun taskName(): String = "write"
         }
     }
@@ -243,4 +269,8 @@ public class ByteChannel(public val autoFlush: Boolean = false) : ByteReadChanne
 /**
  * Thrown when a coroutine awaiting I/O is replaced by another.
  */
-public class ConcurrentIOException(taskName: String) : IllegalStateException("Concurrent $taskName attempts")
+public class ConcurrentIOException(
+    taskName: String,
+    cause: Throwable? = null
+) : IllegalStateException("Concurrent $taskName attempts", cause) {
+}
diff --git a/ktor-io/jvm/src/io/ktor/utils/io/ByteChannel.jvm.kt b/ktor-io/jvm/src/io/ktor/utils/io/ByteChannel.jvm.kt
@@ -0,0 +1,10 @@
+/*
+ * Copyright 2014-2024 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package io.ktor.utils.io
+
+private const val DEVELOPMENT_MODE_KEY: String = "io.ktor.development"
+
+internal actual val DEVELOPMENT_MODE: Boolean
+    get() = System.getProperty(DEVELOPMENT_MODE_KEY)?.toBoolean() == true
diff --git a/ktor-io/posix/src/io/ktor/utils/io/ByteChannel.posix.kt b/ktor-io/posix/src/io/ktor/utils/io/ByteChannel.posix.kt
@@ -0,0 +1,8 @@
+/*
+ * Copyright 2014-2024 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package io.ktor.utils.io
+
+internal actual val DEVELOPMENT_MODE: Boolean
+    get() = false
diff --git a/ktor-io/src/jsAndWasmSharedMain/kotlin/io/ktor/utils/io/ByteChannel.jsAndWasmShared.kt b/ktor-io/src/jsAndWasmSharedMain/kotlin/io/ktor/utils/io/ByteChannel.jsAndWasmShared.kt
@@ -0,0 +1,8 @@
+/*
+ * Copyright 2014-2024 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package io.ktor.utils.io
+
+internal actual val DEVELOPMENT_MODE: Boolean
+    get() = false