Binder: HandleTable owns a mutex (#117)

python-multiplatform/src/commonMain/kotlin/python/multiplatform/ffi/upcall/AsyncUpcall.kt

@@ -275,8 +275,8 @@ internal object AsyncUpcall {
                 }
                 callSoonThreadsafe.invoke(settle, future, PyBool.from(true), marshalled)
             } finally {
-                // The completion half of the handle's lifetime; see [armCancellationNotice]. Under
-                // the GIL, which is the rule every HandleTable mutation lives under, and in a
+                // The completion half of the handle's lifetime; see [armCancellationNotice]. The
+                // table takes its own mutex (the GIL is not what protects it), and this sits in a
                 // `finally` so a delivery that raised does not turn into a permanent root. A
                 // release the Future's done callback already performed is a no-op.
                 if (handle != null) HandleTable.release(handle)

python-multiplatform/src/commonMain/kotlin/python/multiplatform/reflection/HandleTable.kt

@@ -48,11 +48,16 @@ package python.multiplatform.reflection
  *
  * ### Threading
  *
- * Not synchronised. Every mutation happens on a thread that holds the GIL: an upcall arrives
- * with it held, and handing a Kotlin object to Python requires it to build the proxy. That is
- * the same rule the rest of the boundary already lives under (`commonMain/README.md`), and it
- * is what makes the plain `ArrayList`s here safe. It stops being true under free-threading
- * (ROADMAP §9), where this needs a real lock.
+ * Every read and write of the table runs under the table's own [PlatformMutex]. The GIL was
+ * never a guarantee here, even on the default GIL build: on desktop `_pm_release` and
+ * `_pm_resolve` are bound through `ctypes.CFUNCTYPE`, and ctypes releases the GIL around a
+ * `CFUNCTYPE` foreign call, so `UpcallTrampoline.releaseObject` reaches [release] with no GIL
+ * while another thread registers. Under free-threading (ROADMAP §9) there is no GIL at all.
+ *
+ * Holding the mutex cannot deadlock a free-threaded stop-the-world pause: each critical section
+ * is bounded array and counter work and never calls Python, user code, a cleaner or a callback,
+ * so a thread holding it never waits on a thread that is parked for a pause. The mutex is not
+ * re-entrant on every platform, so no critical section takes it twice.
  */
 object HandleTable {
 
@@ -67,11 +72,13 @@ object HandleTable {
 
     private var live = 0
 
+    private val mutex = PlatformMutex()
+
     /** How many handles are currently rooting an object. The leak counter. */
-    val liveCount: Int get() = live
+    val liveCount: Int get() = mutex.withLock { live }
 
     /** How many slots the backing array holds. Bounded by peak [liveCount], not by total issued. */
-    val slotCount: Int get() = objects.size
+    val slotCount: Int get() = mutex.withLock { objects.size }
 
     /**
      * Roots [obj] and returns the handle Python will hold.
@@ -82,17 +89,19 @@ object HandleTable {
      * identity across the boundary -- is the Python side's question and is not settled here.)
      */
     fun register(obj: Any): ObjectReference {
-        val slot: Int
-        if (freeSlots.isEmpty()) {
-            slot = objects.size
-            objects.add(obj)
-            generations.add(1)
-        } else {
-            slot = freeSlots.removeAt(freeSlots.size - 1)
-            objects[slot] = obj
+        return mutex.withLock {
+            val slot: Int
+            if (freeSlots.isEmpty()) {
+                slot = objects.size
+                objects.add(obj)
+                generations.add(1)
+            } else {
+                slot = freeSlots.removeAt(freeSlots.size - 1)
+                objects[slot] = obj
+            }
+            live++
+            ObjectReference.encode(slot, generations[slot])
         }
-        live++
-        return ObjectReference.encode(slot, generations[slot])
     }
 
     /** The object [ref] roots, or `null` if the handle is [ObjectReference.NONE] or stale. */
@@ -108,9 +117,12 @@ object HandleTable {
     fun resolveRaw(raw: Long): Any? {
         if (raw == ObjectReference.NONE_RAW) return null
         val slot = (raw and ObjectReference.SLOT_MASK).toInt()
-        if (slot < 0 || slot >= objects.size) return null
-        if (generations[slot] != (raw ushr ObjectReference.GENERATION_SHIFT).toInt()) return null
-        return objects[slot]
+        val generation = (raw ushr ObjectReference.GENERATION_SHIFT).toInt()
+        return mutex.withLock {
+            if (slot < 0 || slot >= objects.size) null
+            else if (generations[slot] != generation) null
+            else objects[slot]
+        }
     }
 
     /**
@@ -133,11 +145,16 @@ object HandleTable {
         val raw = ref.raw
         if (raw == ObjectReference.NONE_RAW) return false
         val slot = (raw and ObjectReference.SLOT_MASK).toInt()
-        if (slot < 0 || slot >= objects.size) return false
-        if (generations[slot] != (raw ushr ObjectReference.GENERATION_SHIFT).toInt()) return false
-        if (objects[slot] == null) return false
-        releaseSlot(slot)
-        return true
+        val generation = (raw ushr ObjectReference.GENERATION_SHIFT).toInt()
+        return mutex.withLock {
+            if (slot < 0 || slot >= objects.size) false
+            else if (generations[slot] != generation) false
+            else if (objects[slot] == null) false
+            else {
+                releaseSlot(slot)
+                true
+            }
+        }
     }
 
     /**
@@ -147,11 +164,14 @@ object HandleTable {
      * handles issued before the call stay unresolvable afterwards.
      */
     fun releaseAll() {
-        for (slot in objects.indices) {
-            if (objects[slot] != null) releaseSlot(slot)
+        mutex.withLock {
+            for (slot in objects.indices) {
+                if (objects[slot] != null) releaseSlot(slot)
+            }
         }
     }
 
+    /** Caller holds [mutex]. */
     private fun releaseSlot(slot: Int) {
         objects[slot] = null
         generations[slot] = nextGeneration(generations[slot])

python-multiplatform/src/commonMain/kotlin/python/multiplatform/reflection/PlatformMutex.kt

@@ -0,0 +1,25 @@
+package python.multiplatform.reflection
+
+/**
+ * A minimal mutual-exclusion lock for the few tables that must be safe without the GIL.
+ *
+ * Callers hold it only for array and counter work: never across a call into Python, user code,
+ * a cleaner or a callback. Not required to be re-entrant (the JVM actual is, the native one is
+ * not), so a critical section must not take the same mutex again.
+ *
+ * Actuals: `jvmMain` (`ReentrantLock`), `nativeMain` (a POSIX mutex), `wasmJsMain` (no-op, the
+ * target is single-threaded).
+ */
+internal expect class PlatformMutex() {
+    fun lock()
+    fun unlock()
+}
+
+internal inline fun <T> PlatformMutex.withLock(block: () -> T): T {
+    lock()
+    try {
+        return block()
+    } finally {
+        unlock()
+    }
+}

python-multiplatform/src/commonMain/kotlin/python/multiplatform/reflection/UpcallTable.kt

@@ -35,7 +35,7 @@ interface FunctionTableFragment {
  * resolution even though both come from the same fragments.
  *
  * ### Threading
- * Not synchronised, for the same reason as [HandleTable]: installation happens once at start-up,
+ * Not synchronised (unlike [HandleTable], which locks): installation happens once at start-up,
  * or under the GIL for an upcall-driven [register], never concurrently with a lookup.
  */
 object UpcallTable {

python-multiplatform/src/jvmMain/kotlin/python/multiplatform/reflection/PlatformMutex.jvm.kt

@@ -0,0 +1,11 @@
+package python.multiplatform.reflection
+
+import java.util.concurrent.locks.ReentrantLock
+
+internal actual class PlatformMutex actual constructor() {
+    private val lock = ReentrantLock()
+
+    actual fun lock() = lock.lock()
+
+    actual fun unlock() = lock.unlock()
+}

python-multiplatform/src/jvmTest/kotlin/python/multiplatform/reflection/HandleTableConcurrencyTest.kt

@@ -0,0 +1,60 @@
+package python.multiplatform.reflection
+
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.atomic.AtomicReference
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+
+/**
+ * [HandleTable] must be safe with no GIL: the ctypes `CFUNCTYPE` path releases it around
+ * `_pm_release`, and a free-threaded build has none. Pure table, no interpreter needed.
+ */
+class HandleTableConcurrencyTest {
+
+    private class Token(val owner: Int, val seq: Int)
+
+    @Test
+    fun concurrentRegisterResolveReleaseKeepsEveryHandleItsOwn() {
+        val threads = 8
+        val cycles = 20_000
+        val before = HandleTable.liveCount
+        val start = CountDownLatch(1)
+        val done = CountDownLatch(threads)
+        val failure = AtomicReference<Throwable?>(null)
+
+        repeat(threads) { id ->
+            Thread {
+                try {
+                    start.await()
+                    val held = ArrayList<Pair<ObjectReference, Token>>()
+                    for (i in 0 until cycles) {
+                        val token = Token(id, i)
+                        val ref = HandleTable.register(token)
+                        held.add(ref to token)
+                        if (held.size >= 8) {
+                            for ((r, t) in held) {
+                                val got = HandleTable.resolve(r)
+                                assertTrue(got === t, "thread $id resolved a foreign or stale object")
+                            }
+                            for ((r, _) in held) assertTrue(HandleTable.release(r), "release lost a live entry")
+                            for ((r, _) in held) assertNull(HandleTable.resolve(r))
+                            held.clear()
+                        }
+                    }
+                    for ((r, _) in held) HandleTable.release(r)
+                } catch (t: Throwable) {
+                    failure.compareAndSet(null, t)
+                } finally {
+                    done.countDown()
+                }
+            }.start()
+        }
+        start.countDown()
+        done.await()
+
+        failure.get()?.let { throw AssertionError("concurrent table use failed", it) }
+        assertEquals(before, HandleTable.liveCount)
+    }
+}

python-multiplatform/src/nativeMain/kotlin/python/multiplatform/reflection/PlatformMutex.native.kt

@@ -0,0 +1,34 @@
+package python.multiplatform.reflection
+
+import kotlinx.cinterop.ExperimentalForeignApi
+import kotlinx.cinterop.alloc
+import kotlinx.cinterop.nativeHeap
+import kotlinx.cinterop.ptr
+import platform.posix.pthread_mutex_init
+import platform.posix.pthread_mutex_lock
+import platform.posix.pthread_mutex_t
+import platform.posix.pthread_mutex_unlock
+
+/**
+ * A default (non-recursive) POSIX mutex, the one primitive iOS and androidNative both provide.
+ *
+ * The `pthread_mutex_t` lives in [nativeHeap] and is never freed: the only user is the
+ * [HandleTable] singleton, which lives for the life of the process, so there is nothing to
+ * release it for.
+ */
+@OptIn(ExperimentalForeignApi::class)
+internal actual class PlatformMutex actual constructor() {
+    private val mutex = nativeHeap.alloc<pthread_mutex_t>()
+
+    init {
+        pthread_mutex_init(mutex.ptr, null)
+    }
+
+    actual fun lock() {
+        pthread_mutex_lock(mutex.ptr)
+    }
+
+    actual fun unlock() {
+        pthread_mutex_unlock(mutex.ptr)
+    }
+}

python-multiplatform/src/wasmJsMain/kotlin/python/multiplatform/reflection/PlatformMutex.wasmJs.kt

@@ -0,0 +1,8 @@
+package python.multiplatform.reflection
+
+/** wasmJs is single-threaded, so there is nothing to exclude. */
+internal actual class PlatformMutex actual constructor() {
+    actual fun lock() {}
+
+    actual fun unlock() {}
+}