Thread parking for Kotlin/Common

atomicfu/api/atomicfu.api

@@ -135,3 +135,11 @@ 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/ParkingSupport {
+	public static final field INSTANCE Lkotlinx/atomicfu/locks/ParkingSupport;
+	public final fun currentThreadHandle ()Ljava/lang/Thread;
+	public final fun park-LRDsOJo (J)V
+	public final fun parkUntil (Lkotlin/time/TimeMark;)V
+	public final fun unpark (Ljava/lang/Thread;)V
+}
+

atomicfu/build.gradle.kts

@@ -10,7 +10,33 @@ plugins {
     id("kotlin-multiplatform-publish-conventions")
 }
 
+// Support of all non-deprecated targets from the official tier list: https://kotlinlang.org/docs/native-target-support.html
 kotlin {
+    // Tier 1
+    macosX64()
+    macosArm64()
+    iosSimulatorArm64()
+    iosX64()
+
+    // Tier 2
+    linuxX64()
+    linuxArm64()
+    watchosSimulatorArm64()
+    watchosX64()
+    watchosArm32()
+    watchosArm64()
+    tvosSimulatorArm64()
+    tvosX64()
+    tvosArm64()
+    iosArm64()
+
+    // Tier 3
+    androidNativeArm32()
+    androidNativeArm64()
+    androidNativeX86()
+    androidNativeX64()
+    mingwX64()
+    watchosDeviceArm64()
 
     // JS -- always
     js(IR) {
@@ -33,6 +59,12 @@ kotlin {
     wasmWasi {
         nodejs()
     }
+
+    @Suppress("DEPRECATION") //https://github.com/Kotlin/kotlinx-atomicfu/issues/207
+    linuxArm32Hfp()
+
+    @OptIn(ExperimentalKotlinGradlePluginApi::class)
+    applyDefaultHierarchyTemplate()
 
     sourceSets {
         commonMain.dependencies {
@@ -93,87 +125,19 @@ kotlin {
                 implementation(libs.junit.junit)
             }
         }
-    }
-}
-
-// Support of all non-deprecated targets from the official tier list: https://kotlinlang.org/docs/native-target-support.html
-kotlin {
-    // Tier 1
-    macosX64()
-    macosArm64()
-    iosSimulatorArm64()
-    iosX64()
-
-    // Tier 2
-    linuxX64()
-    linuxArm64()
-    watchosSimulatorArm64()
-    watchosX64()
-    watchosArm32()
-    watchosArm64()
-    tvosSimulatorArm64()
-    tvosX64()
-    tvosArm64()
-    iosArm64()
-
-    // Tier 3
-    androidNativeArm32()
-    androidNativeArm64()
-    androidNativeX86()
-    androidNativeX64()
-    mingwX64()
-    watchosDeviceArm64()
-
-    @Suppress("DEPRECATION") //https://github.com/Kotlin/kotlinx-atomicfu/issues/207
-    linuxArm32Hfp()
-
-    @OptIn(ExperimentalKotlinGradlePluginApi::class)
-    applyDefaultHierarchyTemplate {
-        group("native") {
-            group("nativeUnixLike") {
-                withLinux()
-                withApple()
-            }
-        }
-        group("androidNative32Bit") {
-            withAndroidNativeX86()
-            withCompilations { compilation ->
-                (compilation.target as? KotlinNativeTarget)?.konanTarget?.name == "android_arm32"
-            }
-        }
-        group("androidNative64Bit") {
-            withAndroidNativeArm64()
-            withAndroidNativeX64()
-        }
-
-    }
-
-    sourceSets {
-        val nativeUnixLikeMain by getting {
-            kotlin.srcDir("src/nativeUnixLikeMain/kotlin")
-            dependsOn(nativeMain.get())
-        }
-
-        val androidNative32BitMain by getting {
-            kotlin.srcDir("src/androidNative32BitMain/kotlin")
-            dependsOn(nativeMain.get())
-        }
-
-        val androidNative64BitMain by getting {
-            kotlin.srcDir("src/androidNative64BitMain/kotlin")
-            dependsOn(nativeMain.get())
-        }
-
-        val androidNative32BitTest by getting {
-            kotlin.srcDir("src/androidNative32BitTest/kotlin")
-            dependsOn(nativeTest.get())
-        }
-
-        val androidNative64BitTest by getting {
-            kotlin.srcDir("src/androidNative64BitTest/kotlin")
-            dependsOn(nativeTest.get())
-        }
-
+
+        val concurrentMain by creating { dependsOn(commonMain.get()) }
+        val concurrentTest by creating { dependsOn(commonTest.get()) }
+
+        val jvmMain by getting { dependsOn(concurrentMain) }
+        val jvmTest by getting { dependsOn(concurrentTest) }
+
+        val nativeMain by getting { dependsOn(concurrentMain) }
+        val nativeTest by getting { dependsOn(concurrentTest) }
+
+        val nativeUnixLikeMain by creating { dependsOn(nativeMain) }
+        val appleMain by getting { dependsOn(nativeUnixLikeMain) }
+        val linuxMain by getting { dependsOn(nativeUnixLikeMain) }
     }
 
     // atomicfu-cinterop-interop.klib with an empty interop.def file will still be published for compatibility reasons (see KT-68411)
@@ -189,6 +153,7 @@ kotlin {
     }
 }
 
+
 val transformer: Configuration by configurations.creating
 
 dependencies {

atomicfu/src/androidNative32BitMain/kotlin/kotlinx/atomicfu/locks/NativeMutexNode.kt → atomicfu/src/androidNativeMain/kotlin/kotlinx/atomicfu/locks/NativeMutexNode.kt

@@ -4,6 +4,7 @@ import kotlinx.cinterop.*
 import platform.posix.*
 import kotlin.concurrent.Volatile
 
+@OptIn(UnsafeNumber::class)
 public actual class NativeMutexNode {
 
     @Volatile

atomicfu/src/androidNativeMain/kotlin/kotlinx/atomicfu/locks/PosixParkingDelegator.kt

@@ -0,0 +1,59 @@
+package kotlinx.atomicfu.locks
+
+import kotlinx.cinterop.*
+import kotlinx.cinterop.alloc
+import kotlinx.cinterop.pointed
+import kotlinx.cinterop.ptr
+import platform.posix.*
+
+@OptIn(ExperimentalForeignApi::class, UnsafeNumber::class)
+internal actual object ParkingDelegator {
+    actual fun createRef(): ParkingData {
+        val mut = nativeHeap.alloc<pthread_mutex_t>().ptr
+        val cond = nativeHeap.alloc<pthread_cond_t>().ptr
+        callAndVerify(0) { pthread_mutex_init(mut, null) }
+        callAndVerify(0) { pthread_cond_init(cond, null) }
+        return ParkingData(mut, cond)
+    }
+
+    actual inline fun wait(ref: ParkingData, shouldWait: () -> Boolean){
+        callAndVerify(0) { pthread_mutex_lock(ref.mut) }
+        if (shouldWait()) callAndVerify(0) { pthread_cond_wait(ref.cond, ref.mut) }
+        callAndVerify(0) { pthread_mutex_unlock(ref.mut) }
+    }
+
+    actual inline fun timedWait(ref: ParkingData, nanos: Long, shouldWait: () -> Boolean): Unit = memScoped {
+        val ts = alloc<timespec>().ptr
+
+        // Add nanos to current time
+        callAndVerify(0) { clock_gettime(CLOCK_REALTIME.convert(), ts) }
+        ts.pointed.tv_sec = ts.pointed.tv_sec.addNanosToSeconds(nanos)
+        ts.pointed.tv_nsec = (ts.pointed.tv_nsec + nanos % 1_000_000_000).convert()
+        //Fix overflow
+        if (ts.pointed.tv_nsec >= 1_000_000_000) {
+            ts.pointed.tv_sec = ts.pointed.tv_sec.addNanosToSeconds(1_000_000_000)
+            ts.pointed.tv_nsec -= 1_000_000_000
+        }
+        callAndVerify(0) { pthread_mutex_lock(ref.mut) }
+        if (shouldWait()) callAndVerify(0, ETIMEDOUT) { pthread_cond_timedwait(ref.cond, ref.mut, ts) }
+        callAndVerify(0) { pthread_mutex_unlock(ref.mut) }
+    }
+
+    actual fun wake(ref: ParkingData) {
+        callAndVerify(0) { pthread_mutex_lock(ref.mut) }
+        callAndVerify(0) { pthread_cond_signal(ref.cond) }
+        callAndVerify(0) { pthread_mutex_unlock(ref.mut) }
+    }
+
+    actual fun destroyRef(ref: ParkingData) {
+        callAndVerify(0) { pthread_mutex_destroy(ref.mut) }
+        callAndVerify(0) { pthread_cond_destroy(ref.cond) }
+        nativeHeap.free(ref.mut)
+        nativeHeap.free(ref.cond)
+    }
+
+    private inline fun callAndVerify(vararg expectedReturn: Int, block: () -> Int) =
+        callAndVerifyNative(*expectedReturn, getErrno = { errno }, block = block)
+
+}
+internal actual class ParkingData(val mut: CPointer<pthread_mutex_t>, val cond: CPointer<pthread_cond_t>)

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

@@ -0,0 +1,33 @@
+package kotlinx.atomicfu.locks
+
+/**
+ * Internal utility that delegates the thread suspending and resuming to pthread_cond_wait on native.
+ * On jvm delegates to LockSupport.Park.
+ */
+
+internal expect class ParkingData
+
+internal expect object ParkingDelegator {
+    fun createRef(): ParkingData
+    fun wait(ref: ParkingData, shouldWait: () -> Boolean) 
+    fun timedWait(ref: ParkingData, nanos: Long, shouldWait: () -> Boolean)
+    fun wake(ref: ParkingData)
+    fun destroyRef(ref: ParkingData)
+}
+
+/**
+ * Adds nano seconds to current time in seconds.
+ * Clamps for Int.
+ */
+internal fun Int.addNanosToSeconds(nanos: Long): Int = 
+    (this + nanos / 1_000_000_000).coerceIn(Int.MIN_VALUE.toLong(), Int.MAX_VALUE.toLong()).toInt()
+internal fun Long.addNanosToSeconds(nanos: Long): Long {
+
+    // Should never happen as this is checked in `ThreadParker`
+    check(nanos >= 0) { "Cannot wait for a negative number of nanoseconds" }
+    val result =  this + nanos / 1_000_000_000
+
+    // Overflow check: should never happen since this is very far into the future.
+    check(!(this xor result < 0 && this >= 0)) { "Nano seconds addition overflowed, current time in seconds is $this" }
+    return result
+}

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

@@ -0,0 +1,61 @@
+package kotlinx.atomicfu.locks
+
+import kotlin.time.Duration
+import kotlin.time.TimeMark
+
+/**
+ * Parking and unparking support for threads on Kotlin/Native and Kotlin/JVM. 
+ * Can be used as a building block to create locks and other synchronization primitives.
+ *
+ * A call to [ParkingSupport.park] or [ParkingSupport.parkUntil] will suspend the current thread.
+ * A suspended thread will wake up in one of the following four cases:
+ * - A call to [ParkingSupport.unpark] 
+ * - The given `timout` or `deadline` is exceeded
+ * - A spurious wakeup
+ * - (Only on JVM) The thread was interrupted
+ * 
+ * The caller is responsible for verifying the reason of wakeup and how to respond accordingly.
+ */
+expect object ParkingSupport {
+
+    /**
+     * Parks the current thread for [timout] duration.
+     * 
+     * Wakes up in the following cases:
+     * - A call to [ParkingSupport.unpark]
+     * - [timout] is exceeded
+     * - A spurious wakeup
+     * - (Only on JVM) The thread was interrupted 
+     */
+    fun park(timout: Duration)
+
+    /**
+     * Parks the current thread until [deadline] is reached.
+     *
+     * Wakes up in the following cases:
+     * - A call to [ParkingSupport.unpark]
+     * - [deadline] is exceeded
+     * - A spurious wakeup
+     * - (Only on JVM) The thread was interrupted
+     */
+    fun parkUntil(deadline: TimeMark)
+
+    /**
+     * Unparks the thread corresponding to [handle].
+     * If the corresponding thread was not parked the next park call will return immediately.
+     * A [ParkingHandle] can only be pre-unparked once.
+     */
+    fun unpark(handle: ParkingHandle)
+
+    /**
+     * Returns the [ParkingHandle] that can be used to unpark the current thread.
+     */
+    fun currentThreadHandle(): ParkingHandle
+}
+
+/**
+ * Is used to unpark the current thread.
+ * Can be obtained by calling [ParkingSupport.currentThreadHandle].
+ * Is required by [ParkingSupport.unpark].
+ */
+expect class ParkingHandle