Implement park() and name APIs for Worker. (#3493)

backend.native/tests/runtime/workers/worker4.kt

@@ -9,14 +9,88 @@ import kotlin.test.*
 
 import kotlin.native.concurrent.*
 
-@Test fun runTest() {
-    val worker = Worker.start()
-    val future = worker.execute(TransferMode.SAFE, { 41 }) {
-        input -> input + 1
+@Test fun runTest1() {
+    withWorker {
+        val future = execute(TransferMode.SAFE, { 41 }) { input ->
+            input + 1
+        }
+        future.consume { result ->
+            println("Got $result")
+        }
     }
-    future.consume {
-        result -> println("Got $result")
-    }
-    worker.requestTermination().result
     println("OK")
+}
+
+@Test fun runTest2() {
+    withWorker {
+        val counter = AtomicInt(0)
+
+        executeAfter(0, {
+            assertTrue(Worker.current.park(10_000_000, false))
+            assertEquals(counter.value, 0)
+            assertTrue(Worker.current.processQueue())
+            assertEquals(1, counter.value)
+            // Let main proceed.
+            counter.increment()  // counter becomes 2 here.
+            assertTrue(Worker.current.park(10_000_000, true))
+            assertEquals(3, counter.value)
+        }.freeze())
+
+        executeAfter(0, {
+            counter.increment()
+        }.freeze())
+
+        while (counter.value < 2) {
+            Worker.current.park(1_000)
+        }
+
+        executeAfter(0, {
+            counter.increment()
+        }.freeze())
+
+        while (counter.value == 2) {
+            Worker.current.park(1_000)
+        }
+    }
+}
+
+@Test fun runTest3() {
+    val worker = Worker.start(name = "Lumberjack")
+    val counter = AtomicInt(0)
+    worker.executeAfter(0, {
+        assertEquals("Lumberjack", Worker.current.name)
+        counter.increment()
+    }.freeze())
+
+    while (counter.value == 0) {
+        Worker.current.park(1_000)
+    }
+    assertEquals("Lumberjack", worker.name)
+    worker.requestTermination().result
+    assertFailsWith<IllegalStateException> {
+        println(worker.name)
+    }
+}
+
+@Test fun runTest4() {
+    val counter = AtomicInt(0)
+    Worker.current.executeAfter(10_000, {
+        counter.increment()
+    }.freeze())
+    assertTrue(Worker.current.park(1_000_000, process = true))
+    assertEquals(1, counter.value)
+}
+
+@Test fun runTest5() {
+    val main = Worker.current
+    val counter = AtomicInt(0)
+    withWorker {
+        executeAfter(1000, {
+            main.executeAfter(1, {
+                counter.increment()
+            }.freeze())
+        }.freeze())
+        assertTrue(main.park(1000L * 1000 * 1000, process = true))
+        assertEquals(1, counter.value)
+    }
 }

backend.native/tests/runtime/workers/worker9.kt

@@ -29,7 +29,7 @@ fun withLock(op: () -> Unit) {
 @Test fun runTest2() {
     val worker = Worker.start()
     val future = worker.execute(TransferMode.SAFE, {}) {
-        val me = Worker.current!!
+        val me = Worker.current
         var x = 1
         me.executeAfter (20000) {
             println("second ${++x}")

runtime/src/main/cpp/Worker.cpp

@@ -15,7 +15,7 @@
  */
 
 #ifndef KONAN_NO_THREADS
-# define WITH_WORKERS 1
+#define WITH_WORKERS 1
 #endif
 
 #include <stdlib.h>
@@ -64,8 +64,10 @@ enum {
 
 enum JobKind {
   JOB_NONE = 0,
-  JOB_REGULAR = 1,
-  JOB_TERMINATE = 2,
+  JOB_TERMINATE = 1,
+  // Order is important in sense that all job kinds after this one is considered
+  // processed for APIs returning request process status.
+  JOB_REGULAR = 2,
   JOB_EXECUTE_AFTER = 3
 };
 
@@ -104,7 +106,8 @@ typedef KStdOrderedSet<Job, JobCompare> DelayedJobSet;
 
 class Worker {
  public:
-  Worker(KInt id, bool errorReporting) : id_(id), errorReporting_(errorReporting), terminated_(false) {
+  Worker(KInt id, bool errorReporting, KRef customName) : id_(id), errorReporting_(errorReporting), terminated_(false) {
+    name_ = customName != nullptr ? CreateStablePointer(customName) : nullptr;
     pthread_mutex_init(&lock_, nullptr);
     pthread_cond_init(&cond_, nullptr);
   }
@@ -120,18 +123,24 @@ class Worker {
 
   KLong checkDelayedLocked();
 
-  void waitForQueueLocked();
+  bool waitForQueueLocked(KLong timeoutMicroseconds, KLong* remaining);
 
   JobKind processQueueElement(bool blocking);
 
+  bool park(KLong timeoutMicroseconds, bool process);
+
   KInt id() const { return id_; }
 
   bool errorReporting() const { return errorReporting_; }
 
+  KNativePtr name() const { return name_; }
+
  private:
   KInt id_;
   KStdDeque<Job> queue_;
   DelayedJobSet delayed_;
+  // Stable pointer with worker's name.
+  KNativePtr name_;
   // Lock and condition for waiting on the queue.
   pthread_mutex_t lock_;
   pthread_cond_t cond_;
@@ -249,9 +258,10 @@ class State {
     pthread_cond_destroy(&cond_);
   }
 
-  Worker* addWorkerUnlocked(bool errorReporting) {
+  Worker* addWorkerUnlocked(bool errorReporting, KRef customName) {
     Locker locker(&lock_);
-    Worker* worker = konanConstructInstance<Worker>(nextWorkerId(), errorReporting);
+    Worker* worker = konanConstructInstance<Worker>(nextWorkerId(), errorReporting,
+        customName);
     if (worker == nullptr) return nullptr;
     workers_[worker->id()] = worker;
     return worker;
@@ -336,6 +346,12 @@ class State {
     return kind != JOB_NONE && kind != JOB_TERMINATE;
   }
 
+  bool parkUnlocked(KInt id, KLong timeoutMicroseconds, KBoolean process) {
+      // Can only park current worker.
+      if (::g_worker == nullptr || id != ::g_worker->id()) ThrowWorkerInvalidState();
+      return ::g_worker->park(timeoutMicroseconds, process);
+  }
+
   KInt stateOfFutureUnlocked(KInt id) {
     Locker locker(&lock_);
     auto it = futures_.find(id);
@@ -367,6 +383,20 @@ class State {
     return result;
   }
 
+  OBJ_GETTER(getWorkerNameUnlocked, KInt id) {
+    Worker* worker = nullptr;
+    ObjHolder nameHolder;
+    {
+      Locker locker(&lock_);
+      auto it = workers_.find(id);
+      if (it == workers_.end()) {
+        ThrowWorkerInvalidState();
+      }
+      DerefStablePointer(it->second->name(), nameHolder.slot());
+    }
+    RETURN_OBJ(nameHolder.obj());
+  }
+
   KBoolean waitForAnyFuture(KInt version, KInt millis) {
     Locker locker(&lock_);
     if (version != currentVersion_) return false;
@@ -473,8 +503,8 @@ void* workerRoutine(void* argument) {
   return nullptr;
 }
 
-KInt startWorker(KBoolean errorReporting) {
-  Worker* worker = theState()->addWorkerUnlocked(errorReporting != 0);
+KInt startWorker(KBoolean errorReporting, KRef customName) {
+  Worker* worker = theState()->addWorkerUnlocked(errorReporting != 0, customName);
   if (worker == nullptr) return -1;
   pthread_t thread = 0;
   pthread_create(&thread, nullptr, workerRoutine, worker);
@@ -505,6 +535,10 @@ KBoolean processQueue(KInt id) {
    return theState()->processQueueUnlocked(id);
 }
 
+KBoolean park(KInt id, KLong timeoutMicroseconds, KBoolean process) {
+   return theState()->parkUnlocked(id, timeoutMicroseconds, process);
+}
+
 KInt stateOfFuture(KInt id) {
   return theState()->stateOfFutureUnlocked(id);
 }
@@ -513,6 +547,10 @@ OBJ_GETTER(consumeFuture, KInt id) {
   RETURN_RESULT_OF(theState()->consumeFutureUnlocked, id);
 }
 
+OBJ_GETTER(getWorkerName, KInt id) {
+  RETURN_RESULT_OF(theState()->getWorkerNameUnlocked, id);
+}
+
 KInt requestTermination(KInt id, KBoolean processScheduledJobs) {
   Future* future = theState()->addJobToWorkerUnlocked(
       id, nullptr, nullptr, /* toFront = */ !processScheduledJobs, UNCHECKED);
@@ -544,19 +582,16 @@ KNativePtr detachObjectGraphInternal(KInt transferMode, KRef producer) {
 
 #else
 
-KInt startWorker(KBoolean errorReporting) {
+KInt startWorker(KBoolean errorReporting, KRef customName) {
   ThrowWorkerUnsupported();
-  return -1;
 }
 
 KInt stateOfFuture(KInt id) {
   ThrowWorkerUnsupported();
-  return 0;
 }
 
 KInt execute(KInt id, KInt transferMode, KRef producer, KNativePtr jobFunction) {
   ThrowWorkerUnsupported();
-  return 0;
 }
 
 void executeAfter(KInt id, KRef job, KLong afterMicroseconds) {
@@ -567,39 +602,40 @@ KBoolean processQueue(KInt id) {
   ThrowWorkerUnsupported();
 }
 
+KBoolean park(KInt id, KLong timeoutMicroseconds, KBoolean process) {
+   ThrowWorkerUnsupported();
+}
+
 KInt currentWorker() {
   ThrowWorkerUnsupported();
-  return 0;
 }
 
 OBJ_GETTER(consumeFuture, KInt id) {
   ThrowWorkerUnsupported();
-  RETURN_OBJ(nullptr);
+}
+
+OBJ_GETTER(getWorkerName, KInt id) {
+  ThrowWorkerUnsupported();
 }
 
 KInt requestTermination(KInt id, KBoolean processScheduledJobs) {
   ThrowWorkerUnsupported();
-  return -1;
 }
 
 KBoolean waitForAnyFuture(KInt versionToken, KInt millis) {
   ThrowWorkerUnsupported();
-  return false;
 }
 
 KInt versionToken() {
   ThrowWorkerUnsupported();
-  return 0;
 }
 
 OBJ_GETTER(attachObjectGraphInternal, KNativePtr stable) {
   ThrowWorkerUnsupported();
-  return nullptr;
 }
 
 KNativePtr detachObjectGraphInternal(KInt transferMode, KRef producer) {
    ThrowWorkerUnsupported();
-   return nullptr;
 }
 
 #endif  // WITH_WORKERS
@@ -609,7 +645,7 @@ KNativePtr detachObjectGraphInternal(KInt transferMode, KRef producer) {
 Worker* WorkerInit(KBoolean errorReporting) {
 #if WITH_WORKERS
   if (::g_worker != nullptr) return ::g_worker;
-  Worker* worker = theState()->addWorkerUnlocked(errorReporting != 0);
+  Worker* worker = theState()->addWorkerUnlocked(errorReporting != 0, nullptr);
   ::g_worker = worker;
   return worker;
 #else
@@ -672,6 +708,8 @@ Worker::~Worker() {
     DisposeStablePointer(job.executeAfter.operation);
   }
 
+  if (name_ != nullptr) DisposeStablePointer(name_);
+
   pthread_mutex_destroy(&lock_);
   pthread_cond_destroy(&cond_);
 }
@@ -694,7 +732,7 @@ void Worker::putDelayedJob(Job job) {
 bool Worker::waitDelayed(bool blocking) {
   Locker locker(&lock_);
   if (delayed_.size() == 0) return false;
-  if (blocking) waitForQueueLocked();
+  if (blocking) waitForQueueLocked(-1, nullptr);
   return true;
 }
 
@@ -702,7 +740,7 @@ Job Worker::getJob(bool blocking) {
   Locker locker(&lock_);
   RuntimeAssert(!terminated_, "Must not be terminated");
   if (queue_.size() == 0 && !blocking) return Job { .kind = JOB_NONE };
-  waitForQueueLocked();
+  waitForQueueLocked(-1, nullptr);
   auto result = queue_.front();
   queue_.pop_front();
   return result;
@@ -725,10 +763,15 @@ KLong Worker::checkDelayedLocked() {
   }
 }
 
-void Worker::waitForQueueLocked() {
+bool Worker::waitForQueueLocked(KLong timeoutMicroseconds, KLong* remaining) {
   while (queue_.size() == 0) {
     KLong closestToRun = checkDelayedLocked();
-    if (closestToRun == 0) continue;
+    if (closestToRun == 0) {
+        continue;
+    }
+    if (timeoutMicroseconds >= 0) {
+        closestToRun = timeoutMicroseconds < closestToRun || closestToRun < 0 ? timeoutMicroseconds : closestToRun;
+    }
     if (closestToRun > 0) {
       struct timeval tv;
       struct timespec ts;
@@ -737,10 +780,45 @@ void Worker::waitForQueueLocked() {
       ts.tv_nsec = (tv.tv_usec * 1000LL + nsDelta) % 1000000000LL;
       ts.tv_sec = (tv.tv_sec * 1000000000LL + nsDelta) / 1000000000LL;
       pthread_cond_timedwait(&cond_, &lock_, &ts);
+      if (remaining) {
+        struct timeval tvAfter;
+        gettimeofday(&tvAfter, nullptr);
+        KLong usBefore = tv.tv_sec * 1000000LL + tv.tv_usec;
+        KLong usAfter = tvAfter.tv_sec * 1000000LL + tvAfter.tv_usec;
+        *remaining = timeoutMicroseconds - (usAfter - usBefore);
+      }
     } else {
       pthread_cond_wait(&cond_, &lock_);
+      if (remaining) *remaining = 0;
+    }
+    if (timeoutMicroseconds >= 0) return queue_.size() != 0;
+  }
+  return true;
+}
+
+bool Worker::park(KLong timeoutMicroseconds, bool process) {
+  {
+    Locker locker(&lock_);
+    if (terminated_) {
+      return false;
+    }
+    auto arrived = false;
+    KLong remaining = timeoutMicroseconds;
+    do {
+      arrived = waitForQueueLocked(remaining, &remaining);
+    } while (remaining > 0 && !arrived);
+    if (!process) {
+      return arrived;
+    }
+    if (!arrived) {
+      return false;
     }
   }
+  int processed = 0;
+  while (processQueueElement(false) >= JOB_REGULAR) {
+    processed++;
+  }
+  return processed > 0;
 }
 
 JobKind Worker::processQueueElement(bool blocking) {
@@ -806,8 +884,8 @@ JobKind Worker::processQueueElement(bool blocking) {
 
 extern "C" {
 
-KInt Kotlin_Worker_startInternal(KBoolean noErrorReporting) {
-  return startWorker(noErrorReporting);
+KInt Kotlin_Worker_startInternal(KBoolean noErrorReporting, KRef customName) {
+  return startWorker(noErrorReporting, customName);
 }
 
 KInt Kotlin_Worker_currentInternal() {
@@ -830,6 +908,14 @@ KBoolean Kotlin_Worker_processQueueInternal(KInt id) {
   return processQueue(id);
 }
 
+KBoolean Kotlin_Worker_parkInternal(KInt id, KLong timeoutMicroseconds, KBoolean process) {
+  return park(id, timeoutMicroseconds, process);
+}
+
+OBJ_GETTER(Kotlin_Worker_getNameInternal, KInt id) {
+  RETURN_RESULT_OF(getWorkerName, id);
+}
+
 KInt Kotlin_Worker_stateOfFuture(KInt id) {
   return stateOfFuture(id);
 }

runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt

@@ -30,7 +30,7 @@ internal fun executeImpl(worker: Worker, mode: TransferMode, producer: () -> Any
         Future<Any?>(executeInternal(worker.id, mode.value, producer, job))
 
 @SymbolName("Kotlin_Worker_startInternal")
-external internal fun startInternal(errorReporting: Boolean): Int
+external internal fun startInternal(errorReporting: Boolean, name: String?): Int
 
 @SymbolName("Kotlin_Worker_currentInternal")
 external internal fun currentInternal(): Int
@@ -48,6 +48,12 @@ external internal fun executeAfterInternal(id: Int, operation: () -> Unit, after
 @SymbolName("Kotlin_Worker_processQueueInternal")
 external internal fun processQueueInternal(id: Int): Boolean
 
+@SymbolName("Kotlin_Worker_parkInternal")
+external internal fun parkInternal(id: Int, timeoutMicroseconds: Long, process: Boolean): Boolean
+
+@SymbolName("Kotlin_Worker_getNameInternal")
+external internal fun getWorkerNameInternal(id: Int): String?
+
 @ExportForCppRuntime
 internal fun ThrowWorkerUnsupported(): Unit =
         throw UnsupportedOperationException("Workers are not supported")

runtime/src/main/kotlin/kotlin/native/concurrent/Worker.kt

@@ -35,22 +35,26 @@ public inline class Worker @PublishedApi internal constructor(val id: Int) {
          * better to use non-blocking IO combined with more lightweight coroutines.
          *
          * @param errorReporting controls if an uncaught exceptions in the worker will be printed out
+         * @param name defines the optional name of this worker, if none - default naming is used.
+         * @return worker object, usable across multiple concurrent contexts.
          */
-        public fun start(errorReporting: Boolean = true): Worker = Worker(startInternal(errorReporting))
+        public fun start(errorReporting: Boolean = true, name: String? = null): Worker
+                = Worker(startInternal(errorReporting, name))
 
         /**
          * Return the current worker. Worker context is accessible to any valid Kotlin context,
          * but only actual active worker produced with [Worker.start] automatically processes execution requests.
          * For other situations [processQueue] must be called explicitly to process request queue.
+         * @return current worker object, usable across multiple concurrent contexts.
          */
         public val current: Worker get() = Worker(currentInternal())
 
         /**
          * Create worker object from a C pointer.
          *
-         *  @param pointer value returned earlier by [Worker.asCPointer]
+         * @param pointer value returned earlier by [Worker.asCPointer]
          */
-        public fun fromCPointer(pointer: COpaquePointer?) =
+        public fun fromCPointer(pointer: COpaquePointer?): Worker =
                 if (pointer != null) Worker(pointer.toLong().toInt()) else throw IllegalArgumentException()
     }
 
@@ -61,7 +65,7 @@ public inline class Worker @PublishedApi internal constructor(val id: Int) {
      * or terminate immediately. If there are jobs to be execucted with [executeAfter] their execution
      * is awaited for.
      */
-    public fun requestTermination(processScheduledJobs: Boolean = true) =
+    public fun requestTermination(processScheduledJobs: Boolean = true): Future<Unit> =
             Future<Unit>(requestTerminationInternal(id, processScheduledJobs))
 
     /**
@@ -77,7 +81,7 @@ public inline class Worker @PublishedApi internal constructor(val id: Int) {
      * so `kotlin.native.internal.GC.collect()` could be called in the end of `producer` and `job`
      * if garbage cyclic structures or other uncollected objects refer to the value being transferred.
      *
-     * @return the future with the computation result of [job]
+     * @return the future with the computation result of [job].
      */
     @Suppress("UNUSED_PARAMETER")
     @TypedIntrinsic(IntrinsicType.WORKER_EXECUTE)
@@ -92,8 +96,10 @@ public inline class Worker @PublishedApi internal constructor(val id: Int) {
     /**
      * Plan job for further execution in the worker. [operation] parameter must be either frozen, or execution to be
      * planned on the current worker. Otherwise [IllegalStateException] will be thrown.
-     * [afterMicroseconds] defines after how many microseconds delay execution shall happen, 0 means immediately,
-     * on negative values [IllegalArgumentException] is thrown.
+     *
+     * @param afterMicroseconds defines after how many microseconds delay execution shall happen, 0 means immediately,
+     * @throws [IllegalArgumentException] on negative values of [afterMicroseconds].
+     * @throws [IllegalStateException] if [operation] parameter is not frozen and worker is not current.
      */
     public fun executeAfter(afterMicroseconds: Long = 0, operation: () -> Unit): Unit {
         val current = currentInternal()
@@ -103,21 +109,74 @@ public inline class Worker @PublishedApi internal constructor(val id: Int) {
     }
 
     /**
-     * Process pending job(s) on the queue of this worker, returns `true` if something was processed
-     * and `false` otherwise. Note that jobs scheduled with [executeAfter] using non-zero timeout are
+     * Process pending job(s) on the queue of this worker.
+     * Note that jobs scheduled with [executeAfter] using non-zero timeout are
      * not processed this way. If termination request arrives while processing the queue via this API,
      * worker is marked as terminated and will exit once the current request is done with.
+     *
+     * @throws [IllegalStateException] if this request is executed on non-current [Worker].
+     * @return `true` if request(s) was processed and `false` otherwise.
      */
     public fun processQueue(): Boolean = processQueueInternal(id)
 
     /**
-     * String representation of this worker.
+     * Park execution of the current worker until a new request arrives or timeout specified in
+     * [timeoutMicroseconds] elapsed. If [process] is true, pending queue elements are processed,
+     * including delayed requests. Note that multiple requests could be processed this way.
+     *
+     * @param timeoutMicroseconds defines how long to park worker if no requests arrive, waits forever if -1.
+     * @param process defines if arrived request(s) shall be processed.
+     * @return if [process] is `true`: if request(s) was processed `true` and `false` otherwise.
+     *   if [process] is `false`:` true` if request(s) has arrived and `false` if timeout happens.
+     * @throws [IllegalStateException] if this request is executed on non-current [Worker].
+     * @throws [IllegalArgumentException] if timeout value is incorrect.
      */
-    override public fun toString(): String = "worker $id"
+    public fun park(timeoutMicroseconds: Long, process: Boolean = false): Boolean {
+        if (timeoutMicroseconds < -1) throw IllegalArgumentException()
+        return parkInternal(id, timeoutMicroseconds, process)
+    }
+
+    /**
+     * Name of the worker, as specified in [Worker.start] or "worker $id" by default,
+     *
+     * @throws [IllegalStateException] if this request is executed on an invalid worker.
+     */
+    public val name: String
+        get() {
+            val customName = getWorkerNameInternal(id)
+            return if (customName == null) "worker $id" else customName
+        }
+
+    /**
+     * String representation of the worker.
+     */
+    override public fun toString(): String = "Worker $name"
 
     /**
      * Convert worker to a COpaquePointer value that could be passed via native void* pointer.
      * Can be used as an argument of [Worker.fromCPointer].
+     *
+     * @return worker identifier as C pointer.
      */
     public fun asCPointer() : COpaquePointer? = id.toLong().toCPointer()
+}
+
+/**
+ * Executes [block] with new [Worker] as resource, by starting the new worker, calling provided [block]
+ * (in current context) with newly started worker as [this] and terminating worker after the block completes.
+ * Note that this operation is pretty heavyweight, use preconfigured worker or worker pool if need to
+ * execute it frequently.
+ *
+ * @param name of the started worker.
+ * @param errorReporting controls if uncaught errors in worker to be reported.
+ * @param block to be executed.
+ * @return value returned by the block.
+ */
+public inline fun <R> withWorker(name: String? = null, errorReporting: Boolean = true, block: Worker.() -> R): R {
+    val worker = Worker.start(errorReporting, name)
+    try {
+        return worker.block()
+    } finally {
+        worker.requestTermination().result
+    }
 }