KTOR-9373 Make ConcurrentMap iteration safe on Native (#5407)

* KTOR-9373 Add test for CIOEngine.close() with active endpoints

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* KTOR-9373 Make ConcurrentMap iteration safe on Native

Return snapshot copies from entries/keys/values to match JVM's
ConcurrentHashMap weakly-consistent iteration semantics. Also
synchronize size and isEmpty() which were reading without the lock.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Clean up CIOEngine close test: remove issue prefix and comments

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Address review feedback (iteration 1)

- Fix ConcurrentModificationException in ConcurrentMap.entries on Native:
  use LinkedHashMap(delegate).entries to create entries that don't hold
  references to the original backing map's EntryRef objects

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: e5l

ktor-client/ktor-client-cio/common/test/CIOEngineTest.kt

@@ -171,6 +171,20 @@ class CIOEngineTest : ClientEngineTest<CIOEngineConfig>(CIO) {
         }
     }
 
+    @Test
+    fun `close engine with active endpoints does not throw ConcurrentModificationException`() = testClient {
+        test { client ->
+            val jobs = (1..5).map { i ->
+                client.async {
+                    runCatching {
+                        client.get("$TEST_SERVER/content/hello?i=$i")
+                    }
+                }
+            }
+            jobs.forEach { it.await() }
+        }
+    }
+
     @Test
     fun testErrorMessageWhenServerDontRespondWithUpgrade() = testClient {
         config {

ktor-utils/posix/src/io/ktor/util/collections/ConcurrentMapNative.kt

@@ -32,24 +32,24 @@ public actual class ConcurrentMap<Key, Value> public actual constructor(
     }
 
     actual override val size: Int
-        get() = delegate.size
+        get() = synchronized(lock) { delegate.size }
 
     actual override fun containsKey(key: Key): Boolean = synchronized(lock) { delegate.containsKey(key) }
 
     actual override fun containsValue(value: Value): Boolean = synchronized(lock) { delegate.containsValue(value) }
 
     actual override fun get(key: Key): Value? = synchronized(lock) { delegate[key] }
 
-    actual override fun isEmpty(): Boolean = delegate.isEmpty()
+    actual override fun isEmpty(): Boolean = synchronized(lock) { delegate.isEmpty() }
 
     actual override val entries: MutableSet<MutableMap.MutableEntry<Key, Value>>
-        get() = synchronized(lock) { delegate.entries }
+        get() = synchronized(lock) { LinkedHashMap(delegate).entries }
 
     actual override val keys: MutableSet<Key>
-        get() = synchronized(lock) { delegate.keys }
+        get() = synchronized(lock) { LinkedHashSet(delegate.keys) }
 
     actual override val values: MutableCollection<Value>
-        get() = synchronized(lock) { delegate.values }
+        get() = synchronized(lock) { ArrayList(delegate.values) }
 
     actual override fun clear() {
         synchronized(lock) {