[K/N] Worker API improvements ^KT-52429

Merge-request: KT-MR-6898
Merged-by: Alexander Shabalin <Alexander.Shabalin@jetbrains.com>

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) {

kotlin-native/backend.native/tests/interop/workerSignals/workerSignals.cpp

@@ -0,0 +1,31 @@
+#include "workerSignals.h"
+
+#include <cassert>
+#include <pthread.h>
+#include <signal.h>
+
+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<pthread_t>(thread);
+    auto result = pthread_kill(t, SIGUSR1);
+    assert(result == 0);
+}
+
+extern "C" int getValue(void) {
+    return value;
+}

kotlin-native/backend.native/tests/interop/workerSignals/workerSignals.def

@@ -0,0 +1,2 @@
+language = C
+headerFilter = **/workerSignals.h

kotlin-native/backend.native/tests/interop/workerSignals/workerSignals.h

@@ -0,0 +1,13 @@
+#include <stdint.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+void setupSignalHandler(void);
+void signalThread(uint64_t thread, int value);
+int getValue(void);
+
+#ifdef __cplusplus
+}
+#endif

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)
+    }
+}

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
+    }
+}

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<KULong>(thread);
+  }
+
+  OBJ_GETTER0(getActiveWorkers) {
+      std_support::vector<KInt> 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"

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

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<Worker>
+            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)
 }
 
 /**