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

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) {
+}

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

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