Timeout semantics

Author: bbrockbernd

atomicfu/src/commonMain/kotlin/kotlinx/atomicfu/locks/Mutex.kt renamed to atomicfu/src/commonMain/kotlin/kotlinx/atomicfu/locks/SynchronousMutex.kt

@@ -3,6 +3,7 @@ package kotlinx.atomicfu.locks
 import kotlin.contracts.ExperimentalContracts
 import kotlin.contracts.InvocationKind
 import kotlin.contracts.contract
+import kotlin.time.Duration
 
 /**
  * Mutual exclusion for Kotlin Multiplatform.
@@ -27,12 +28,7 @@ import kotlin.contracts.contract
  * }
  * ```
  */
-expect class Mutex() {
-    /**
-     * Returns `true` if this mutex is locked.
-     */
-    fun isLocked(): Boolean
-
+expect class SynchronousMutex() {
     /**
      * Tries to lock this mutex, returning `false` if this mutex is already locked.
      *
@@ -41,6 +37,8 @@ expect class Mutex() {
      * lock acquisition.     
      */
     fun tryLock(): Boolean
+    
+    fun tryLock(timeout: Duration): Boolean
 
     /**
      * Locks the mutex, suspends the thread until the lock is acquired.
@@ -68,7 +66,7 @@ expect class Mutex() {
  * @return result of [block]
  */
 @OptIn(ExperimentalContracts::class)
-inline fun <T> Mutex.withLock(block: () -> T): T {
+inline fun <T> SynchronousMutex.withLock(block: () -> T): T {
     contract { callsInPlace(block, InvocationKind.EXACTLY_ONCE) }
     lock()
     return try {

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

@@ -3,9 +3,10 @@ package kotlinx.atomicfu.locks
 import kotlinx.atomicfu.AtomicRef
 import kotlinx.atomicfu.atomic
 import kotlin.time.Duration
+import kotlin.time.TimeSource
 
 internal class NativeMutex(
-    val park : () -> Unit = { ParkingSupport.park(Duration.INFINITE) },
+    val park : (Duration) -> Unit = { ParkingSupport.park(it) },
     val unpark : (ParkingHandle) -> Unit = ParkingSupport::unpark,
 ) {
     /**
@@ -38,16 +39,19 @@ internal class NativeMutex(
     private val owningThread = atomic<ParkingHandle?>(null)
     private val state = atomic(0)
     private val holdCount = atomic(0)
-
-
+    
     fun lock() {
+        tryLock(Duration.INFINITE)
+    }
+
+    fun tryLock(duration: Duration): Boolean {
         val currentParkingHandle = ParkingSupport.currentThreadHandle()
 
         // Has to be checked in this order!
         if (holdCount.value > 0 && currentParkingHandle == owningThread.value) {
             // Is reentring thread 
             holdCount.incrementAndGet()
-            return
+            return true
         }
 
         // Otherwise try acquire lock
@@ -56,20 +60,23 @@ internal class NativeMutex(
         if (newState == 1) {
             owningThread.value = currentParkingHandle
             holdCount.incrementAndGet()
-            return
+            return true
         }
 
         // 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.nodeWait()
+            // if timeout
+            if (!prevNode.nodeWait(duration)) return false
             parkingQueue.dequeue()
             owningThread.value = currentParkingHandle
             holdCount.incrementAndGet()
-            return
+            return true
         }
+        
+        return true
     }
 
     fun unlock() {
@@ -88,16 +95,18 @@ internal class NativeMutex(
 
         // If waiters wake up the first in line. The woken up thread will dequeue the node.
         if (currentState > 0) {
-            val nextParker = parkingQueue.getHead()
-            nextParker.nodeWake()
+            var nextParker = parkingQueue.getHead()
+            // If cancelled And there are other waiting nodes, go to next
+            while (!nextParker.nodeWake() && state.decrementAndGet() > 0) {
+                // We only dequeue here in case of timeoud out node.
+                // Dequeueing woken nodes can lead to issues when pre-unparked.
+                parkingQueue.dequeue()
+                nextParker = parkingQueue.getHead()
+            }
             return
         }
     }
 
-    fun isLocked(): Boolean {
-        return state.value > 0
-    }
-
     fun tryLock(): Boolean {
         val currentThreadId = ParkingSupport.currentThreadHandle()
         if (holdCount.value > 0 && owningThread.value == currentThreadId || state.compareAndSet(0, 1)) {
@@ -149,30 +158,42 @@ internal class NativeMutex(
         val parker = atomic<Any>(Empty)
         val next = atomic<Node?>(null)
 
-        fun nodeWait() {
+        fun nodeWait(duration: Duration): Boolean {
+            val deadline = TimeSource.Monotonic.markNow() + duration
             while (true) {
                 when (parker.value) {
-                    Empty -> if (parker.compareAndSet(Empty, ParkingSupport.currentThreadHandle())) park()
-                    is ParkingHandle -> park()
-                    Awoken -> return
+                    Empty -> if (parker.compareAndSet(Empty, ParkingSupport.currentThreadHandle())) {
+                        park(deadline - TimeSource.Monotonic.markNow())
+                        if (deadline < TimeSource.Monotonic.markNow()) 
+                            parker.compareAndSet(ParkingSupport.currentThreadHandle(), Cancelled)
+                    }
+                    is ParkingHandle -> {
+                        park(deadline - TimeSource.Monotonic.markNow())
+                        if (deadline < TimeSource.Monotonic.markNow())
+                            parker.compareAndSet(ParkingSupport.currentThreadHandle(), Cancelled)
+                    }
+                    Awoken -> return true
+                    Cancelled -> return false
                 }
             }
         }
 
-        fun nodeWake() {
+        fun nodeWake(): Boolean {
             while (true) {
                 when (val currentState = parker.value) {
-                    Empty -> if (parker.compareAndSet(Empty, Awoken)) return
+                    Empty -> if (parker.compareAndSet(Empty, Awoken)) return true
                     is ParkingHandle -> if (parker.compareAndSet(currentState, Awoken)) {
                         unpark(currentState)
-                        return
+                        return true
                     }
                     Awoken -> throw IllegalStateException("Node is already woken")
+                    Cancelled -> return false
                 }
             }
         }
     }
 
     private object Empty 
     private object Awoken
+    private object Cancelled
 }

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

@@ -0,0 +1,58 @@
+package kotlinx.atomicfu.locks
+
+import kotlinx.atomicfu.atomic
+import kotlin.random.Random
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.time.Duration.Companion.milliseconds
+
+class LockWithTimoutTests {
+
+    // Helper class with atomic counter in constructor
+    class AtomicCounter(initialValue: Int = 0) {
+        val counter = atomic(initialValue)
+
+        fun incrementAndGet(): Int = counter.incrementAndGet()
+        val value: Int get() = counter.value
+    }
+
+    @Test
+    fun timeoutLockStressTest() {
+        val mutex = SynchronousMutex()
+        val counter = AtomicCounter(0)
+        val targetCount = 1000
+        val threads = mutableListOf<TestThread>()
+
+        // Create 5 test threads
+        repeat(5) { threadId ->
+            val thread = testThread {
+                while (counter.value < targetCount) {
+                    // Try to acquire the lock with a timeout
+                    if (mutex.tryLock((Random.nextInt(1, 10)).milliseconds)) {
+                        try {
+                            // Increment the counter if lock was acquired
+                            if (counter.value < targetCount) {
+                                counter.incrementAndGet()
+                            }
+                            // Random sleep to increase variation
+                            sleepMillis(Random.nextInt(0, 5).toLong())
+                        } finally {
+                            mutex.unlock()
+                        }
+                    }
+
+                    // Random sleep between attempts to increase variation
+                    sleepMillis(Random.nextInt(0, 3).toLong())
+                }
+            }
+            threads.add(thread)
+        }
+
+        // Wait for all threads to complete
+        threads.forEach { it.join() }
+
+        // Verify the counter reached the target
+        assertEquals(targetCount, counter.value)
+    }
+
+}

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

@@ -49,7 +49,7 @@ class NativeMutexTest {
 
     @Test
     fun testNativeMutexFast() {
-        val mutex = Mutex()
+        val mutex = SynchronousMutex()
         val resultList = mutableListOf<String>()
 
         val fut1 = testThread {

atomicfu/src/jsAndWasmSharedMain/kotlin/kotlinx/atomicfu/locks/Mutex.jsAndWasmShared.kt

@@ -1,15 +1,17 @@
 package kotlinx.atomicfu.locks
 
+import kotlin.time.Duration
+
 /**
  * Part of multiplatform mutex.
  * Since this mutex will run in a single threaded environment, it doesn't provide any real synchronization.
  * 
  * It does keep track of reentrancy.
  */
-actual class Mutex {
+actual class SynchronousMutex {
     private var state = 0
-    actual fun isLocked(): Boolean = state != 0
     actual fun tryLock(): Boolean = true
+    actual fun tryLock(timeout: Duration): Boolean = true
     actual fun lock(): Unit { state++ }
     actual fun unlock(): Unit { if (state-- < 0) throw IllegalStateException("Mutex already unlocked") }
 }

atomicfu/src/jvmMain/kotlin/kotlinx/atomicfu/locks/Mutex.kt renamed to atomicfu/src/jvmMain/kotlin/kotlinx/atomicfu/locks/SynchronousMutex.kt

@@ -1,14 +1,17 @@
 package kotlinx.atomicfu.locks
 
+import java.util.concurrent.TimeUnit
+import kotlin.time.Duration
+
 /**
  * This mutex uses a [ReentrantLock].
  * 
  * [getReentrantLock] obtains the actual [ReentrantLock].
- * Construct with `Mutex(reentrantLock)` to create a [Mutex] that uses an existing instance of [ReentrantLock].
+ * Construct with `Mutex(reentrantLock)` to create a [SynchronousMutex] that uses an existing instance of [ReentrantLock].
  */
-actual class Mutex(private val reentrantLock: java.util.concurrent.locks.ReentrantLock) {
+actual class SynchronousMutex(private val reentrantLock: java.util.concurrent.locks.ReentrantLock) {
     actual constructor(): this(ReentrantLock())
-    actual fun isLocked(): Boolean = reentrantLock.isLocked
+    actual fun tryLock(timeout: Duration): Boolean = reentrantLock.tryLock(timeout.inWholeNanoseconds, TimeUnit.NANOSECONDS)
     actual fun tryLock(): Boolean = reentrantLock.tryLock()
     actual fun lock() = reentrantLock.lock()
     actual fun unlock() = reentrantLock.unlock()

atomicfu/src/jvmTest/kotlin/kotlinx/atomicfu/locks/NativeMutexTimeoutLincheckTest.kt

@@ -0,0 +1,63 @@
+package kotlinx.atomicfu.locks
+
+import org.jetbrains.kotlinx.lincheck.annotations.Operation
+import org.jetbrains.kotlinx.lincheck.check
+import org.jetbrains.kotlinx.lincheck.strategy.managed.modelchecking.ModelCheckingOptions
+import org.jetbrains.kotlinx.lincheck.util.LoggingLevel
+import org.junit.Test
+import java.util.concurrent.ConcurrentHashMap
+import kotlin.time.Duration.Companion.nanoseconds
+
+class NativeMutexTimeoutLincheckTest {
+    class Counter {
+        @Volatile
+        private var value = 0
+
+        fun inc(): Int = ++value
+        fun get() = value
+    }
+    private val counter = Counter()
+    private val localParkers = ConcurrentHashMap<ParkingHandle, ThreadParker>()
+
+    private val lock = NativeMutex(
+        park = { localParkers[ParkingSupport.currentThreadHandle()]!!.parkNanos(it.inWholeNanoseconds) },
+        unpark = { localParkers[it]!!.unpark() }
+    )
+
+    @Test
+    fun modelCheckingTest(): Unit = ModelCheckingOptions()
+        .iterations(2) // Change to 300 for exhaustive testing
+        .invocationsPerIteration(5_000)
+        .actorsBefore(1)
+        .threads(3)
+        .actorsPerThread(3)
+        .actorsAfter(0)
+        .hangingDetectionThreshold(100)
+        .logLevel(LoggingLevel.INFO)
+        .check(this::class.java)
+
+    @Operation
+    fun incNoTimout() {
+        localParkers.computeIfAbsent(ParkingSupport.currentThreadHandle()) { ThreadParker() }
+        lock.lock()
+        counter.inc()
+        lock.unlock()
+    }
+    
+    @Operation
+    fun incTimeout() {
+        localParkers.computeIfAbsent(ParkingSupport.currentThreadHandle()) { ThreadParker() }
+        if (lock.tryLock(0.nanoseconds)) {
+            counter.inc()
+            lock.unlock()
+        }
+    }
+
+    @Operation
+    fun get() {
+        localParkers.computeIfAbsent(ParkingSupport.currentThreadHandle()) { ThreadParker() }
+        lock.lock()
+        counter.get()
+        lock.unlock()
+    }
+}

atomicfu/src/nativeMain/kotlin/kotlinx/atomicfu/locks/Synchronized.kt

@@ -2,7 +2,7 @@ package kotlinx.atomicfu.locks
 
 public actual open class SynchronizedObject {
 
-    private val nativeMutex = Mutex()
+    private val nativeMutex = SynchronousMutex()
     public fun lock() = nativeMutex.lock()
     public fun tryLock(): Boolean = nativeMutex.tryLock()
     public fun unlock() = nativeMutex.unlock()

atomicfu/src/nativeMain/kotlin/kotlinx/atomicfu/locks/Mutex.kt renamed to atomicfu/src/nativeMain/kotlin/kotlinx/atomicfu/locks/SynchronousMutex.kt

@@ -1,9 +1,11 @@
 package kotlinx.atomicfu.locks
 
-actual class Mutex {
+import kotlin.time.Duration
+
+actual class SynchronousMutex {
     private val lock = NativeMutex()
-    actual fun isLocked() = lock.isLocked()
     actual fun tryLock() = lock.tryLock()
+    actual fun tryLock(timeout: Duration) = lock.tryLock(timeout)
     actual fun lock() = lock.lock()
     actual fun unlock() = lock.unlock()
 }