Introduce mutex based on new parking

Author: bbrockbernd

atomicfu/api/atomicfu.api

@@ -135,6 +135,20 @@ public final class kotlinx/atomicfu/TraceKt {
 	public static final fun named (Lkotlinx/atomicfu/TraceBase;Ljava/lang/String;)Lkotlinx/atomicfu/TraceBase;
 }
 
+public final class kotlinx/atomicfu/locks/Mutex {
+	public fun <init> ()V
+	public fun <init> (Ljava/util/concurrent/locks/ReentrantLock;)V
+	public final fun getReentrantLock ()Ljava/util/concurrent/locks/ReentrantLock;
+	public final fun isLocked ()Z
+	public final fun lock ()V
+	public final fun tryLock ()Z
+	public final fun unlock ()V
+}
+
+public final class kotlinx/atomicfu/locks/MutexKt {
+	public static final fun withLock (Lkotlinx/atomicfu/locks/Mutex;Lkotlin/jvm/functions/Function0;)Ljava/lang/Object;
+}
+
 public final class kotlinx/atomicfu/parking/KThread {
 	public static final field Companion Lkotlinx/atomicfu/parking/KThread$Companion;
 }

atomicfu/build.gradle.kts

@@ -119,6 +119,7 @@ kotlin {
 
         jvmTest {
             dependencies {
+                implementation("org.jetbrains.kotlinx:lincheck:2.35")
                 implementation("org.jetbrains.kotlin:kotlin-reflect")
                 implementation("org.jetbrains.kotlin:kotlin-test")
                 implementation("org.jetbrains.kotlin:kotlin-test-junit")

atomicfu/src/commonMain/kotlin/kotlinx/atomicfu/locks/Mutex.kt

@@ -0,0 +1,79 @@
+package kotlinx.atomicfu.locks
+
+import kotlin.contracts.ExperimentalContracts
+import kotlin.contracts.InvocationKind
+import kotlin.contracts.contract
+
+/**
+ * Mutual exclusion for Kotlin Multiplatform.
+ * 
+ * It can protect a shared resource or critical section from multiple thread accesses.
+ * Threads can acquire the lock by calling [lock] and release the lock by calling [unlock].
+ * 
+ * When a thread calls [lock] while another thread is locked, it will suspend until the lock is released.
+ * When multiple threads are waiting for the lock, they will acquire it in a fair order (first in first out).
+ * 
+ * It is reentrant, meaning the lock holding thread can call [lock] multiple times without suspending.
+ * To release the lock (after multiple [lock] calls) an equal number of [unlock] calls are required.
+ * 
+ * This Mutex should not be used in combination with coroutines and `suspend` functions
+ * as it blocks the waiting thread.
+ * Use the `Mutex` from the coroutines library instead.
+ * 
+ * ```Kotlin
+ * mutex.withLock {
+ *     // Critical section only executed by
+ *     // one thread at a time.
+ * }
+ * ```
+ */
+expect class Mutex() {
+    /**
+     * Returns `true` if this mutex is locked.
+     */
+    fun isLocked(): Boolean
+
+    /**
+     * Tries to lock this mutex, returning `false` if this mutex is already locked.
+     *
+     * It is recommended to use [withLock] for safety reasons, so that the acquired lock is always
+     * released at the end of your critical section, and [unlock] is never invoked before a successful
+     * lock acquisition.     
+     */
+    fun tryLock(): Boolean
+
+    /**
+     * Locks the mutex, suspends the thread until the lock is acquired.
+     * 
+     * It is recommended to use [withLock] for safety reasons, so that the acquired lock is always
+     * released at the end of your critical section, and [unlock] is never invoked before a successful
+     * lock acquisition.
+     */
+    fun lock()
+
+    /**
+     * Releases the lock.
+     * Throws [IllegalStateException] when the current thread is not holding the lock.
+     * 
+     * It is recommended to use [withLock] for safety reasons, so that the acquired lock is always
+     * released at the end of the critical section, and [unlock] is never invoked before a successful
+     * lock acquisition.
+     */
+    fun unlock() 
+}
+
+/**
+ * Executes the given code [block] under this mutex's lock.
+ * 
+ * @return result of [block]
+ */
+@OptIn(ExperimentalContracts::class)
+inline fun <T> Mutex.withLock(block: () -> T): T {
+    contract { callsInPlace(block, InvocationKind.EXACTLY_ONCE) }
+    lock()
+    return try {
+        block()
+    } finally {
+        unlock()
+    }
+}

atomicfu/src/concurrentMain/kotlin/kotlinx/atomicfu/locks/Mutex.kt

@@ -0,0 +1,150 @@
+package kotlinx.atomicfu.locks
+
+import kotlinx.atomicfu.AtomicRef
+import kotlinx.atomicfu.atomic
+import kotlinx.atomicfu.parking.ThreadParker
+import kotlinx.atomicfu.parking.currentThreadId
+
+internal class NativeMutex {
+    /**
+     * Mutex implementation for Kotlin/Native.
+     * In concurrentMain sourceSet to be testable with Lincheck.
+     *
+     * The [state] variable stands for: 0 -> Lock is free
+     *                                  1 -> Lock is locked but no waiters
+     *                                  4 -> Lock is locked with 3 waiters
+     *
+     * The state.incrementAndGet() call makes my claim on the lock.
+     * The returned value either means I acquired it (when it is 1).
+     * Or I need to enqueue and park (when it is > 1).
+     *
+     * The [holdCount] variable is to enable reentrancy.
+     *
+     * Works by using a [parkingQueue].
+     * When a thread tries to acquire the lock, but finds it is already locked it enqueues by appending to the [parkingQueue].
+     * On enqueue the parking queue provides the second last node, this node is used to park on.
+     * When our thread is woken up that means that the thread parked on the thrid last node called unpark on the second last node.
+     * Since a woken up thread is first inline it means that it's node is the head and can therefore dequeue.
+     *
+     * Unlocking happens by calling state.decrementAndGet().
+     * When the returned value is 0 it means the lock is free and we can simply return.
+     * If the new state is > 0, then there are waiters. We wake up the first by unparking the head of the queue.
+     * This even works when a thread is not parked yet,
+     * since the ThreadParker can be pre-unparked resulting in the parking call to return immediately.
+     */
+    private val parkingQueue = ParkingQueue()
+    private val owningThread = atomic(-1L)
+    private val state = atomic(0)
+    private val holdCount = atomic(0)
+
+
+    fun lock() {
+        val currentThreadId = currentThreadId()
+
+        // Has to be checked in this order!
+        if (holdCount.value > 0 && currentThreadId == owningThread.value) {
+            // Is reentring thread 
+            holdCount.incrementAndGet()
+            return
+        }
+
+        // Otherwise try acquire lock
+        val newState = state.incrementAndGet()
+        // If new state 1 than I have acquired lock skipping queue.
+        if (newState == 1) {
+            owningThread.value = currentThreadId
+            holdCount.incrementAndGet()
+            return
+        }
+
+        // If state larger than 1 -> enqueue and park
+        // When woken up thread has acquired lock and his node in the queue is therefore at the head.
+        // Remove head
+        if (newState > 1) {
+            val prevNode = parkingQueue.enqueue()
+            prevNode.parker.park()
+            parkingQueue.dequeue()
+            owningThread.value = currentThreadId
+            holdCount.incrementAndGet()
+            return
+        }
+    }
+
+    fun unlock() {
+        val currentThreadId = currentThreadId()
+        val currentOwnerId = owningThread.value
+        if (currentThreadId != currentOwnerId) throw IllegalStateException("Thread is not holding the lock")
+
+        // dec hold count
+        val newHoldCount = holdCount.decrementAndGet()
+        if (newHoldCount > 0) return
+        if (newHoldCount < 0) throw IllegalStateException("Thread unlocked more than it locked")
+
+        // Lock is released by decrementing (only if decremented to 0)
+        val currentState = state.decrementAndGet()
+        if (currentState == 0) return
+
+        // If waiters wake up the first in line. The woken up thread will dequeue the node.
+        if (currentState > 0) {
+            val nextParker = parkingQueue.getHead()
+            nextParker.parker.unpark()
+            return
+        }
+    }
+
+    fun isLocked(): Boolean {
+        return state.value > 0
+    }
+
+    fun tryLock(): Boolean {
+        val currentThreadId = currentThreadId()
+        if (holdCount.value > 0 && owningThread.value == currentThreadId || state.compareAndSet(0, 1)) {
+            owningThread.value = currentThreadId
+            holdCount.incrementAndGet()
+            return true
+        }
+        return false
+    }
+
+    // Based on Micheal-Scott Queue
+    private class ParkingQueue {
+        private val head: AtomicRef<Node>
+        private val tail: AtomicRef<Node>
+
+        init {
+            val first = Node()
+            head = atomic(first)
+            tail = atomic(first)
+        }
+
+        fun getHead(): Node {
+            return head.value
+        }
+
+        fun enqueue(): Node {
+            while (true) {
+                val node = Node()
+                val curTail = tail.value
+                if (curTail.next.compareAndSet(null, node)) {
+                    tail.compareAndSet(curTail, node)
+                    return curTail
+                }
+                else tail.compareAndSet(curTail, curTail.next.value!!)
+            }
+        }
+
+        fun dequeue() {
+            while (true) {
+                val currentHead = head.value
+                val currentHeadNext = currentHead.next.value ?: throw IllegalStateException("Dequeing parker but already empty, should not be possible")
+                if (head.compareAndSet(currentHead, currentHeadNext)) return
+            }
+        }
+
+    }
+
+    private class Node {
+        val parker = ThreadParker()
+        val next = atomic<Node?>(null)
+    }
+}

atomicfu/src/concurrentMain/kotlin/kotlinx/atomicfu/parking/KThread.kt

@@ -33,4 +33,6 @@ expect class Parker private constructor() {
          */
         fun unpark(kThread: KThread): Unit
     }
-}
+}
+
+internal expect fun currentThreadId(): Long 

atomicfu/src/concurrentTest/kotlin/kotlinx/atomicfu/locks/NativeMutexTest.kt

@@ -0,0 +1,90 @@
+package kotlinx.atomicfu.locks
+
+import kotlinx.atomicfu.parking.sleepMills
+import kotlinx.atomicfu.parking.testThread
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+class NativeMutexTest {
+    
+    
+    @Test
+    fun testNativeMutexSlow() {
+        val mutex = NativeMutex()
+        val resultList = mutableListOf<String>()
+
+        val fut1 = testThread {  
+            repeat(30) { i ->
+                mutex.lock()
+                resultList.add("a$i")
+                sleepMills(100)
+                resultList.add("a$i")
+                mutex.unlock()
+            }
+        }
+
+        val fut2 = testThread {
+            repeat(30) { i ->
+                mutex.lock()
+                resultList.add("b$i")
+                sleepMills(100)
+                resultList.add("b$i")
+                mutex.unlock()
+            }
+        }
+
+        repeat(30) { i ->
+            mutex.lock()
+            resultList.add("c$i")
+            sleepMills(100)
+            resultList.add("c$i")
+            mutex.unlock()
+        }
+        fut1.join()
+        fut2.join()
+        
+        resultList.filterIndexed { i, _ -> i % 2 == 0 }
+            .zip(resultList.filterIndexed {i, _ -> i % 2 == 1}) { a, b ->
+            assertEquals(a, b)
+        }
+    }
+
+    @Test
+    fun testNativeMutexFast() {
+        val mutex = Mutex()
+        val resultList = mutableListOf<String>()
+
+        val fut1 = testThread {
+            repeat(30000) { i ->
+                mutex.lock()
+                resultList.add("a$i")
+                resultList.add("a$i")
+                mutex.unlock()
+            }
+        }
+
+        val fut2 = testThread {
+            repeat(30000) { i -> 
+                mutex.lock()
+                resultList.add("b$i")
+                resultList.add("b$i")
+                mutex.unlock()
+            }
+        }
+
+        repeat(30000) { i ->
+            mutex.lock()
+            resultList.add("c$i")
+            resultList.add("c$i")
+            mutex.unlock()
+        }
+        fut1.join()
+        fut2.join()
+        
+        resultList
+            .filterIndexed { i, _ -> i % 2 == 0 }
+            .zip(resultList.filterIndexed {i, _ -> i % 2 == 1}) { a, b ->
+            assertEquals(a, b)
+        }
+    }
+}

atomicfu/src/concurrentTest/kotlin/kotlinx/atomicfu/locks/ReentrancyTests.kt

@@ -0,0 +1,28 @@
+package kotlinx.atomicfu.locks
+
+import kotlin.test.Test
+import kotlin.test.assertFails
+
+class ReentrancyTests {
+    
+    @Test
+    fun reentrantTestSuccess() {
+        val lock = NativeMutex()
+        lock.lock()
+        lock.lock()
+        lock.unlock()
+        lock.unlock()
+    }
+    
+    @Test
+    fun reentrantTestFail() {
+        val lock = NativeMutex()
+        lock.lock()
+        lock.lock()
+        lock.unlock()
+        lock.unlock()
+        assertFails {
+            lock.unlock()
+        }
+    }
+}