From 6f844f15e014d2522d0a2db9c379f38c89336d05 Mon Sep 17 00:00:00 2001 From: Alexander Shabalin Date: Mon, 22 Aug 2022 10:41:14 +0000 Subject: [PATCH] [K/N] Worker API improvements ^KT-52429 Merge-request: KT-MR-6898 Merged-by: Alexander Shabalin --- .../backend.native/tests/build.gradle | 18 +++++++ .../interop/workerSignals/workerSignals.cpp | 31 +++++++++++ .../interop/workerSignals/workerSignals.def | 2 + .../interop/workerSignals/workerSignals.h | 13 +++++ .../interop/workerSignals/workerSignals.kt | 27 ++++++++++ .../runtime/workers/worker_list_workers.kt | 38 ++++++++++++++ kotlin-native/runtime/src/main/cpp/Worker.cpp | 52 +++++++++++++++++++ .../kotlin/native/concurrent/Internal.kt | 6 +++ .../kotlin/kotlin/native/concurrent/Worker.kt | 19 +++++++ 9 files changed, 206 insertions(+) create mode 100644 kotlin-native/backend.native/tests/interop/workerSignals/workerSignals.cpp create mode 100644 kotlin-native/backend.native/tests/interop/workerSignals/workerSignals.def create mode 100644 kotlin-native/backend.native/tests/interop/workerSignals/workerSignals.h create mode 100644 kotlin-native/backend.native/tests/interop/workerSignals/workerSignals.kt create mode 100644 kotlin-native/backend.native/tests/runtime/workers/worker_list_workers.kt diff --git a/kotlin-native/backend.native/tests/build.gradle b/kotlin-native/backend.native/tests/build.gradle index 290e0c7d779..44ea8e37307 100644 --- a/kotlin-native/backend.native/tests/build.gradle +++ b/kotlin-native/backend.native/tests/build.gradle @@ -1115,6 +1115,11 @@ standaloneTest("worker_threadlocal_no_leak") { source = "runtime/workers/worker_threadlocal_no_leak.kt" } +task worker_list_workers(type: KonanLocalTest) { + disabled = (project.testTarget == 'wasm32') // Needs pthreads. + source = "runtime/workers/worker_list_workers.kt" +} + standaloneTest("freeze0") { enabled = (project.testTarget != 'wasm32') && (cacheTesting == null) // No workers on WASM. useGoldenData = true @@ -4322,6 +4327,12 @@ createInterop("threadStates") { it.extraOpts "-Xcompile-source", "$projectDir/interop/threadStates/threadStates.cpp" } +createInterop("workerSignals") { + it.defFile "interop/workerSignals/workerSignals.def" + it.headers "$projectDir/interop/workerSignals/workerSignals.h" + it.extraOpts "-Xcompile-source", "$projectDir/interop/workerSignals/workerSignals.cpp" +} + if (PlatformInfo.isAppleTarget(project)) { createInterop("objcSmoke") { it.defFile 'interop/objc/objcSmoke.def' @@ -4827,6 +4838,13 @@ interopTest("interop_cppSkiaSignature") { UtilsKt.dependsOnPlatformLibs(it) } +interopTest("interop_workerSignals") { + disabled = (project.testTarget == 'wasm32') || // No interop for wasm yet. + (project.testTarget == 'mingw_x86' || project.testTarget == 'mingw_x64') // cross-thread signalling does not work on Windows + source = "interop/workerSignals/workerSignals.kt" + interop = "workerSignals" +} + /* TODO: This test isn't run automatically task interop_echo_server(type: RunInteropKonanTest) { diff --git a/kotlin-native/backend.native/tests/interop/workerSignals/workerSignals.cpp b/kotlin-native/backend.native/tests/interop/workerSignals/workerSignals.cpp new file mode 100644 index 00000000000..ad48a7953d9 --- /dev/null +++ b/kotlin-native/backend.native/tests/interop/workerSignals/workerSignals.cpp @@ -0,0 +1,31 @@ +#include "workerSignals.h" + +#include +#include +#include + +namespace { + +int pendingValue = 0; +thread_local int value = 0; + +void signalHandler(int signal) { + value = pendingValue; +} + +} // namespace + +extern "C" void setupSignalHandler(void) { + signal(SIGUSR1, &signalHandler); +} + +extern "C" void signalThread(uint64_t thread, int value) { + pendingValue = value; + auto t = reinterpret_cast(thread); + auto result = pthread_kill(t, SIGUSR1); + assert(result == 0); +} + +extern "C" int getValue(void) { + return value; +} diff --git a/kotlin-native/backend.native/tests/interop/workerSignals/workerSignals.def b/kotlin-native/backend.native/tests/interop/workerSignals/workerSignals.def new file mode 100644 index 00000000000..4bf83f1d95c --- /dev/null +++ b/kotlin-native/backend.native/tests/interop/workerSignals/workerSignals.def @@ -0,0 +1,2 @@ +language = C +headerFilter = **/workerSignals.h diff --git a/kotlin-native/backend.native/tests/interop/workerSignals/workerSignals.h b/kotlin-native/backend.native/tests/interop/workerSignals/workerSignals.h new file mode 100644 index 00000000000..0ce39b76e3f --- /dev/null +++ b/kotlin-native/backend.native/tests/interop/workerSignals/workerSignals.h @@ -0,0 +1,13 @@ +#include + +#ifdef __cplusplus +extern "C" { +#endif + +void setupSignalHandler(void); +void signalThread(uint64_t thread, int value); +int getValue(void); + +#ifdef __cplusplus +} +#endif diff --git a/kotlin-native/backend.native/tests/interop/workerSignals/workerSignals.kt b/kotlin-native/backend.native/tests/interop/workerSignals/workerSignals.kt new file mode 100644 index 00000000000..00395706fc9 --- /dev/null +++ b/kotlin-native/backend.native/tests/interop/workerSignals/workerSignals.kt @@ -0,0 +1,27 @@ +@file:OptIn(kotlin.ExperimentalStdlibApi::class) + +import kotlin.native.concurrent.* +import kotlin.test.* +import workerSignals.* + +const val defaultValue = 0 +const val newValue = 42 + +fun main() { + setupSignalHandler() + + withWorker { + val before = execute(TransferMode.SAFE, {}) { + getValue() + }.result + assertEquals(defaultValue, getValue()) + assertEquals(defaultValue, before) + + signalThread(platformThreadId, newValue) + val after = execute(TransferMode.SAFE, {}) { + getValue() + }.result + assertEquals(defaultValue, getValue()) + assertEquals(newValue, after) + } +} diff --git a/kotlin-native/backend.native/tests/runtime/workers/worker_list_workers.kt b/kotlin-native/backend.native/tests/runtime/workers/worker_list_workers.kt new file mode 100644 index 00000000000..1ac15f88b5e --- /dev/null +++ b/kotlin-native/backend.native/tests/runtime/workers/worker_list_workers.kt @@ -0,0 +1,38 @@ +@file:OptIn(kotlin.ExperimentalStdlibApi::class) + +package runtime.workers.worker_list_workers + +import kotlin.native.concurrent.* +import kotlin.test.* + +const val WORKER_COUNT = 10 + +@Test +fun getAllWorkers() { + val workers = Array(WORKER_COUNT) { Worker.start() } + + val expectedWorkers = listOf(Worker.current) + workers + + assertEquals(expectedWorkers.toSet(), Worker.activeWorkers.toSet()) + + workers.forEach { + it.requestTermination().result + } +} + +@Test +fun getActiveWorkers() { + val workers = Array(WORKER_COUNT) { Worker.start() } + + val expectedWorkers = mutableListOf(Worker.current) + (0 until WORKER_COUNT step 2).forEach { i -> + workers[i].requestTermination().result + expectedWorkers.add(workers[i + 1]) + } + + assertEquals(expectedWorkers.toSet(), Worker.activeWorkers.toSet()) + + (0 until WORKER_COUNT step 2).forEach { i -> + workers[i + 1].requestTermination().result + } +} diff --git a/kotlin-native/runtime/src/main/cpp/Worker.cpp b/kotlin-native/runtime/src/main/cpp/Worker.cpp index ba8b78576b6..fd9e6763011 100644 --- a/kotlin-native/runtime/src/main/cpp/Worker.cpp +++ b/kotlin-native/runtime/src/main/cpp/Worker.cpp @@ -30,6 +30,7 @@ #include "Exceptions.h" #include "KAssert.h" #include "Memory.h" +#include "Natives.h" #include "ObjCMMAPI.h" #include "Runtime.h" #include "Types.h" @@ -632,6 +633,33 @@ class State { } } + KULong getWorkerPlatformThreadIdUnlocked(KInt id) { + Locker locker(&lock_); + auto it = workers_.find(id); + if (it == workers_.end()) { + ThrowWorkerAlreadyTerminated(); + } + pthread_t thread = it->second->thread(); + static_assert(sizeof(pthread_t) <= sizeof(KULong), "Casting pthread_t to ULong will lose data"); + return reinterpret_cast(thread); + } + + OBJ_GETTER0(getActiveWorkers) { + std_support::vector workers; + { + Locker locker(&lock_); + + workers.reserve(workers_.size()); + for (auto [id, worker] : workers_) { + workers.push_back(id); + } + } + ObjHolder arrayHolder; + AllocArrayInstance(theIntArrayTypeInfo, workers.size(), arrayHolder.slot()); + std::copy(workers.begin(), workers.end(), IntArrayAddressOfElementAt(arrayHolder.obj()->array(), 0)); + RETURN_OBJ(arrayHolder.obj()); + } + private: pthread_mutex_t lock_; pthread_cond_t cond_; @@ -764,6 +792,14 @@ KNativePtr detachObjectGraphInternal(KInt transferMode, KRef producer) { } } +KULong platformThreadId(KInt id) { + return theState()->getWorkerPlatformThreadIdUnlocked(id); +} + +OBJ_GETTER0(activeWorkers) { + RETURN_RESULT_OF0(theState()->getActiveWorkers); +} + #else KInt startWorker(WorkerExceptionHandling exceptionHandling, KRef customName) { @@ -822,6 +858,14 @@ KNativePtr detachObjectGraphInternal(KInt transferMode, KRef producer) { ThrowWorkerUnsupported(); } +KULong platformThreadId(KInt id) { + ThrowWorkerUnsupported(); +} + +OBJ_GETTER0(activeWorkers) { + ThrowWorkerUnsupported(); +} + #endif // WITH_WORKERS } // namespace @@ -1224,4 +1268,12 @@ void Kotlin_Worker_waitTermination(KInt id) { WaitNativeWorkerTermination(id); } +KULong Kotlin_Worker_getPlatformThreadIdInternal(KInt id) { + return platformThreadId(id); +} + +OBJ_GETTER0(Kotlin_Worker_getActiveWorkersInternal) { + RETURN_RESULT_OF0(activeWorkers); +} + } // extern "C" diff --git a/kotlin-native/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt b/kotlin-native/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt index c20201c854f..1c1c43c5f64 100644 --- a/kotlin-native/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt +++ b/kotlin-native/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt @@ -128,3 +128,9 @@ external internal fun checkIfFrozen(ref: Any?) @InternalForKotlinNative @GCUnsafeCall("Kotlin_Worker_waitTermination") external public fun waitWorkerTermination(worker: Worker) + +@GCUnsafeCall("Kotlin_Worker_getPlatformThreadIdInternal") +external internal fun getPlatfromThreadIdInternal(id: Int): ULong + +@GCUnsafeCall("Kotlin_Worker_getActiveWorkersInternal") +external internal fun getActiveWorkersInternal(): IntArray diff --git a/kotlin-native/runtime/src/main/kotlin/kotlin/native/concurrent/Worker.kt b/kotlin-native/runtime/src/main/kotlin/kotlin/native/concurrent/Worker.kt index 2a481fe47ed..c0951d3a54f 100644 --- a/kotlin-native/runtime/src/main/kotlin/kotlin/native/concurrent/Worker.kt +++ b/kotlin-native/runtime/src/main/kotlin/kotlin/native/concurrent/Worker.kt @@ -60,6 +60,16 @@ public value class Worker @PublishedApi internal constructor(val id: Int) { @Deprecated("Use kotlinx.cinterop.StableRef instead", level = DeprecationLevel.WARNING) public fun fromCPointer(pointer: COpaquePointer?): Worker = if (pointer != null) Worker(pointer.toLong().toInt()) else throw IllegalArgumentException() + + /** + * Get a list of all unterminated workers. + * + * Thread safety: If some other thread calls [Worker.requestTermination] at the same time then this may + * return a [Worker] that's already terminated. + */ + @ExperimentalStdlibApi + public val activeWorkers: List + get() = getActiveWorkersInternal().map { Worker(it) } } /** @@ -170,6 +180,15 @@ public value class Worker @PublishedApi internal constructor(val id: Int) { */ @Deprecated("Use kotlinx.cinterop.StableRef instead", level = DeprecationLevel.WARNING) public fun asCPointer() : COpaquePointer? = id.toLong().toCPointer() + + /** + * Get platform thread id of the worker thread. + * + * Usually returns `pthread_t` casted to [ULong]. + */ + @ExperimentalStdlibApi + public val platformThreadId: ULong + get() = getPlatfromThreadIdInternal(id) } /**