From b8df05ed495964b4ffcc08511ec47e7cb998bc88 Mon Sep 17 00:00:00 2001 From: Nikolay Igotti Date: Fri, 25 Oct 2019 14:53:47 +0300 Subject: [PATCH] Implement park() and name APIs for Worker. (#3493) --- .../tests/runtime/workers/worker4.kt | 90 ++++++++++- .../tests/runtime/workers/worker9.kt | 2 +- runtime/src/main/cpp/Worker.cpp | 140 ++++++++++++++---- .../kotlin/native/concurrent/Internal.kt | 8 +- .../kotlin/kotlin/native/concurrent/Worker.kt | 81 ++++++++-- 5 files changed, 273 insertions(+), 48 deletions(-) diff --git a/backend.native/tests/runtime/workers/worker4.kt b/backend.native/tests/runtime/workers/worker4.kt index 44c30b86a30..99fece9eef3 100644 --- a/backend.native/tests/runtime/workers/worker4.kt +++ b/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 { + 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) + } } \ No newline at end of file diff --git a/backend.native/tests/runtime/workers/worker9.kt b/backend.native/tests/runtime/workers/worker9.kt index 67819215e0e..247780f5e8a 100644 --- a/backend.native/tests/runtime/workers/worker9.kt +++ b/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}") diff --git a/runtime/src/main/cpp/Worker.cpp b/runtime/src/main/cpp/Worker.cpp index 467f831ff24..5a115c47bee 100644 --- a/runtime/src/main/cpp/Worker.cpp +++ b/runtime/src/main/cpp/Worker.cpp @@ -15,7 +15,7 @@ */ #ifndef KONAN_NO_THREADS -# define WITH_WORKERS 1 +#define WITH_WORKERS 1 #endif #include @@ -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 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 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(nextWorkerId(), errorReporting); + Worker* worker = konanConstructInstance(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); } diff --git a/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt b/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt index e034245fa18..5f715c9a0af 100644 --- a/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt +++ b/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt @@ -30,7 +30,7 @@ internal fun executeImpl(worker: Worker, mode: TransferMode, producer: () -> Any Future(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") diff --git a/runtime/src/main/kotlin/kotlin/native/concurrent/Worker.kt b/runtime/src/main/kotlin/kotlin/native/concurrent/Worker.kt index 81fc891031a..fc007caa942 100644 --- a/runtime/src/main/kotlin/kotlin/native/concurrent/Worker.kt +++ b/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 = Future(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 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 + } } \ No newline at end of file