Pthreads condvar wrapper and improvements in mutex wrapper

modules/build-script/src/main/kotlin/ru/landgrafhomyak/bgtu/networks0/build_script/Dependencies.kt

@@ -2,4 +2,5 @@ package ru.landgrafhomyak.bgtu.networks0.build_script
 
 object Dependencies {
 	val kotlin_atomicfu = "org.jetbrains.kotlinx:atomicfu:${Versions.kotlin_atomicfu}"
+	val kotlin_coroutines_core = "org.jetbrains.kotlinx:kotlinx-coroutines-core:${Versions.kotlin_coroutines_core}"
 }

modules/build-script/src/main/kotlin/ru/landgrafhomyak/bgtu/networks0/build_script/Versions.kt

@@ -3,4 +3,5 @@ package ru.landgrafhomyak.bgtu.networks0.build_script
 @Suppress( "MayBeConstant")
 object Versions {
 	val kotlin_atomicfu = "0.27.0"
+	val kotlin_coroutines_core = "1.10.1"
 }

modules/build-script/src/main/kotlin/ru/landgrafhomyak/bgtu/networks0/build_script/warnings.kt

@@ -0,0 +1,26 @@
+package ru.landgrafhomyak.bgtu.networks0.build_script
+
+import org.jetbrains.kotlin.gradle.dsl.KotlinMultiplatformExtension
+import org.jetbrains.kotlin.gradle.plugin.KotlinTarget
+import ru.landgrafhomyak.kotlin.kmp_gradle_build_helper.configureAllCompilersOptions
+
+fun KotlinTarget.suppressWarnings(vararg warningNames: String) {
+	this.compilations.configureEach c@{
+		this@c.compileTaskProvider.configure t@{
+			this.compilerOptions a@{
+				warningNames.map { w -> "-Xsuppress-warning=${w}" }.let(this@a.freeCompilerArgs::addAll)
+			}
+		}
+	}
+}
+
+fun KotlinMultiplatformExtension.suppressWarnings(vararg warningNames: String) {
+	this.targets.configureEach t@{ this@t.suppressWarnings(*warningNames) }
+}
+
+
+fun KotlinMultiplatformExtension.configureWarnings() {
+	this.configureAllCompilersOptions a@{ this@a.freeCompilerArgs.addAll("-Wextra", "-Xexpect-actual-classes") }
+	this.suppressWarnings("JoinDeclarationAndAssignment", "CanBeVal", "ConvertSecondaryConstructorToPrimary", "FunctionName", "PropertyName", "PrivatePropertyName")
+}
+

modules/low-level/c-interop-utilities/build.gradle.kts

@@ -1,3 +1,4 @@
+import ru.landgrafhomyak.bgtu.networks0.build_script.configureWarnings
 import ru.landgrafhomyak.bgtu.networks0.build_script.setupHierarchy
 
 plugins {
@@ -10,6 +11,8 @@ repositories {
 
 
 kotlin {
+	configureWarnings()
+
 	mingwX64()
 	linuxX64()
 	linuxArm64()

modules/low-level/multithreading/build.gradle.kts

@@ -1,3 +1,5 @@
+import ru.landgrafhomyak.bgtu.networks0.build_script.Dependencies
+import ru.landgrafhomyak.bgtu.networks0.build_script.configureWarnings
 import ru.landgrafhomyak.bgtu.networks0.build_script.setupHierarchy
 
 plugins {
@@ -10,6 +12,8 @@ repositories {
 
 
 kotlin {
+	configureWarnings()
+
 	mingwX64()
 	linuxX64()
 	linuxArm64()
@@ -18,6 +22,12 @@ kotlin {
 	sourceSets {
 		setupHierarchy()
 
+		commonMain {
+			dependencies {
+				implementation(Dependencies.kotlin_atomicfu)
+			}
+		}
+
 		nativeMain {
 			dependencies {
 				implementation(project(":modules:low-level:c-interop-utilities"))

modules/low-level/multithreading/src/commonMain/kotlin/ru/landgrafhomyak/bgtu/networks0/low_level/multithreading/TCondition.kt

@@ -0,0 +1,9 @@
+package ru.landgrafhomyak.bgtu.networks0.low_level.multithreading
+
+expect class TCondition : AutoCloseable {
+	constructor()
+
+	fun await(mutex: TMutex)
+	fun signalOne()
+	fun signalAll()
+}

modules/low-level/multithreading/src/commonMain/kotlin/ru/landgrafhomyak/bgtu/networks0/low_level/multithreading/utilities.kt

@@ -0,0 +1,52 @@
+package ru.landgrafhomyak.bgtu.networks0.low_level.multithreading
+
+import kotlinx.atomicfu.AtomicLong
+
+internal fun AtomicLong.compareAndExchange(expected: Long, newValue: Long): Long {
+	while (true) {
+		val old = this.value
+		if (old != expected) return old
+		if (this.compareAndSet(old, newValue)) return old
+	}
+}
+
+internal inline fun <R> _safeAutoClose(
+	finally: () -> Unit,
+	action: () -> R
+): R = _safeAutoClose(onAbort = finally, onSuccess = finally, onCrossReturn = finally, action = action)
+
+internal inline fun <R> _safeAutoClose(
+	onAbort: () -> Unit = {},
+	onSuccess: () -> Unit = {},
+	action: () -> R
+): R = _safeAutoClose(onAbort = onAbort, onSuccess = onSuccess, onCrossReturn = onSuccess, action = action)
+
+internal inline fun <R> _safeAutoClose(
+	onAbort: () -> Unit = {},
+	onSuccess: () -> Unit = {},
+	onCrossReturn: () -> Unit = {},
+	action: () -> R
+): R {
+	val ret: R
+	var wasError = false
+	var crossReturned = true
+	try {
+		ret = action()
+		crossReturned = false
+	} catch (e1: Throwable) {
+		wasError = true
+		try {
+			onAbort()
+		} catch (e2: Throwable) {
+			e1.addSuppressed(e2)
+		}
+		throw e1
+	} finally {
+		if (!wasError) {
+			if (crossReturned)
+				onCrossReturn()
+			onSuccess()
+		}
+	}
+	return ret
+}

modules/low-level/multithreading/src/posixMain/kotlin/ru/landgrafhomyak/bgtu/networks0/low_level/multithreading/TCondition.kt

@@ -0,0 +1,84 @@
+package ru.landgrafhomyak.bgtu.networks0.low_level.multithreading
+
+import kotlinx.atomicfu.AtomicLong
+import kotlinx.atomicfu.atomic
+import kotlinx.atomicfu.update
+import kotlinx.cinterop.CPointer
+import kotlinx.cinterop.ExperimentalForeignApi
+import kotlinx.cinterop.alloc
+import kotlinx.cinterop.free
+import kotlinx.cinterop.nativeHeap
+import kotlinx.cinterop.ptr
+import platform.posix.pthread_cond_t
+import platform.posix.pthread_cond_init
+import platform.posix.pthread_cond_destroy
+import platform.posix.pthread_cond_wait
+import platform.posix.pthread_cond_broadcast
+import ru.landgrafhomyak.bgtu.networks0.low_level.c_interop_utilities.PosixUtilities
+import ru.landgrafhomyak.bgtu.networks0.low_level.multithreading.TMutex.RefcntAccess
+
+
+@OptIn(ExperimentalForeignApi::class)
+actual class TCondition : AutoCloseable {
+	val _descriptor: CPointer<pthread_cond_t>
+	private var _refcnt: AtomicLong = atomic(0L)
+
+	actual constructor() {
+		this._descriptor = nativeHeap.alloc<pthread_cond_t>().ptr
+		var err = pthread_cond_init(this._descriptor, null)
+		if (err != 0)
+			PosixUtilities.throwErrno { d -> RuntimeException("Failed to initialize pthreads condition: $d") }
+	}
+
+
+	private fun _throwClosed(): Nothing =
+		throw IllegalStateException("Pthreads condition was destroyed")
+
+	private inline fun <R> _withRefcnt(protectedBlock: () -> R): R {
+		this._refcnt.update { o ->
+			if (o < 0) this._throwClosed()
+			return@update o + 1
+		}
+		_safeAutoClose(finally = { this._refcnt.update(Long::dec) }) {
+			return protectedBlock()
+		}
+	}
+
+	internal fun _decref() {
+		this._refcnt.update(Long::dec)
+	}
+
+	@OptIn(TMutex.RefcntAccess::class)
+	actual fun await(mutex: TMutex) = this._withRefcnt {
+		mutex._incref()
+		_safeAutoClose(finally = { mutex._decref() }) {
+			var err = pthread_cond_wait(this._descriptor, mutex._descriptor)
+			if (err != 0)
+				PosixUtilities.throwErrno(err) { d -> RuntimeException("Failed to wait on pthreads condition: $d") }
+		}
+	}
+
+	actual fun signalAll() = this._withRefcnt {
+		var err = pthread_cond_broadcast(this._descriptor)
+		if (err != 0)
+			PosixUtilities.throwErrno(err) { d -> RuntimeException("Failed to signal pthreads condition: $d") }
+	}
+
+	actual fun signalOne() = this._withRefcnt {
+		var err = pthread_cond_broadcast(this._descriptor)
+		if (err != 0)
+			PosixUtilities.throwErrno(err) { d -> RuntimeException("Failed to signal pthreads condition: $d") }
+	}
+
+	override fun close() {
+		val state = this._refcnt.compareAndExchange(0, -1)
+		when {
+			state > 0 -> throw IllegalStateException("There are waiting threads on this pthreads condition")
+			state < 0 -> this._throwClosed()
+		}
+		var err = pthread_cond_destroy(this._descriptor)
+		if (err != 0)
+			PosixUtilities.throwErrno(err) { d -> RuntimeException("Failed to destroy pthreads condition: $d") }
+		nativeHeap.free(this._descriptor)
+	}
+}

modules/low-level/multithreading/src/posixMain/kotlin/ru/landgrafhomyak/bgtu/networks0/low_level/multithreading/TMutex.kt

@@ -1,50 +1,79 @@
 package ru.landgrafhomyak.bgtu.networks0.low_level.multithreading
 
+import kotlinx.atomicfu.AtomicLong
+import kotlinx.atomicfu.atomic
+import kotlinx.atomicfu.update
 import kotlinx.cinterop.CPointer
 import kotlinx.cinterop.ExperimentalForeignApi
-import platform.posix.free
+import kotlinx.cinterop.alloc
+import kotlinx.cinterop.free
+import kotlinx.cinterop.nativeHeap
+import kotlinx.cinterop.ptr
 import platform.posix.pthread_mutex_destroy
 import platform.posix.pthread_mutex_init
 import platform.posix.pthread_mutex_lock
 import platform.posix.pthread_mutex_t
 import platform.posix.pthread_mutex_unlock
-import ru.landgrafhomyak.bgtu.networks0.low_level.c_interop_utilities.CUtilities
 import ru.landgrafhomyak.bgtu.networks0.low_level.c_interop_utilities.PosixUtilities
 
-@Suppress("JoinDeclarationAndAssignment", "CanBeVal", "ConvertSecondaryConstructorToPrimary")
 @OptIn(ExperimentalForeignApi::class)
 actual class TMutex : AutoCloseable {
-	private var isClosed: Boolean
+	private var _refcnt: AtomicLong = atomic(0L)
-	private val descriptor: CPointer<pthread_mutex_t>
+	internal val _descriptor: CPointer<pthread_mutex_t>
 
 	actual constructor() {
-		this.descriptor = CUtilities.newOrOom<pthread_mutex_t>()
+		this._descriptor = nativeHeap.alloc<pthread_mutex_t>().ptr
-		var err = pthread_mutex_init(this.descriptor, null)
+		var err = pthread_mutex_init(this._descriptor, null)
 		if (err != 0)
 			PosixUtilities.throwErrno(err) { d -> RuntimeException("Failed to create pthreads mutex: $d") }
-		this.isClosed = false
+	}
+
+	private fun _throwClosed(): Nothing =
+		throw IllegalStateException("Pthreads mutex was destroyed")
+
+	@RequiresOptIn
+	annotation class RefcntAccess
+
+	@RefcntAccess
+	internal fun _incref() {
+		this._refcnt.update { o ->
+			if (o < 0) this._throwClosed()
+			return@update o + 1
+		}
+	}
+
+	@RefcntAccess
+	internal fun _decref() {
+		this._refcnt.update(Long::dec)
 	}
 
 	actual fun lock() {
-		if (this.isClosed) throw IllegalStateException("Mutex was destroyed")
+		this._refcnt.update { o ->
-		var err = pthread_mutex_lock(this.descriptor)
+			if (o < 0) this._throwClosed()
+			return@update o + 1
+		}
+		var err = pthread_mutex_lock(this._descriptor)
 		if (err != 0)
 			PosixUtilities.throwErrno(err) { d -> RuntimeException("Failed to lock pthreads mutex: $d") }
 	}
 
 	actual fun unlock() {
-		if (this.isClosed) throw IllegalStateException("Mutex was destroyed")
+		if (this._refcnt.value < 0) this._throwClosed()
-		var err = pthread_mutex_unlock(this.descriptor)
+		var err = pthread_mutex_unlock(this._descriptor)
 		if (err != 0)
 			PosixUtilities.throwErrno(err) { d -> RuntimeException("Failed to unlock pthreads mutex: $d") }
+		this._refcnt.update(Long::dec)
 	}
 
 	override fun close() {
-		if (this.isClosed) return
+		val state = this._refcnt.compareAndExchange(0, -1)
-		this.isClosed = true
+		when {
-		var err = pthread_mutex_destroy(this.descriptor)
+			state > 0 -> throw IllegalStateException("There are waiting threads on this pthreads mutex")
+			state < 0 -> this._throwClosed()
+		}
+		var err = pthread_mutex_destroy(this._descriptor)
 		if (err != 0)
 			PosixUtilities.throwErrno(err) { d -> RuntimeException("Failed to destroy pthreads mutex: $d") }
-		free(this.descriptor)
+		nativeHeap.free(this._descriptor)
 	}
 }