From 89c2fb21c4c46f8aa01a9e6b314763006526727d Mon Sep 17 00:00:00 2001 From: Nikolay Igotti Date: Tue, 14 May 2019 14:52:38 +0300 Subject: [PATCH] Worker.executeAfter() API for scheduling jobs on worker. (#2971) --- backend.native/tests/build.gradle | 2 +- .../tests/runtime/workers/worker9.kt | 37 ++- runtime/src/main/cpp/Types.h | 3 + runtime/src/main/cpp/Worker.cpp | 233 +++++++++++++++--- .../kotlin/native/concurrent/Internal.kt | 3 + .../kotlin/kotlin/native/concurrent/Worker.kt | 17 +- 6 files changed, 257 insertions(+), 38 deletions(-) diff --git a/backend.native/tests/build.gradle b/backend.native/tests/build.gradle index d4aec0d9925..3324a581788 100644 --- a/backend.native/tests/build.gradle +++ b/backend.native/tests/build.gradle @@ -762,7 +762,7 @@ task worker8(type: RunKonanTest) { task worker9(type: RunKonanTest) { disabled = (project.testTarget == 'wasm32') // Workers need pthreads. - goldValue = "zzz\n42\nOK\n" + goldValue = "zzz\n42\nOK\nfirst 2\nsecond 3\nfrozen OK\n" source = "runtime/workers/worker9.kt" } diff --git a/backend.native/tests/runtime/workers/worker9.kt b/backend.native/tests/runtime/workers/worker9.kt index 45c97533543..cd99c5a594a 100644 --- a/backend.native/tests/runtime/workers/worker9.kt +++ b/backend.native/tests/runtime/workers/worker9.kt @@ -9,7 +9,7 @@ import kotlin.test.* import kotlin.native.concurrent.* -@Test fun runTest() { +@Test fun runTest1() { withLock { println("zzz") } val worker = Worker.start() val future = worker.execute(TransferMode.SAFE, {}) { @@ -24,4 +24,39 @@ import kotlin.native.concurrent.* fun withLock(op: () -> Unit) { op() +} + +@Test fun runTest2() { + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, {}) { + val me = Worker.current!! + var x = 1 + me.executeAfter (20000) { + println("second ${++x}") + } + me.executeAfter(10000) { + println("first ${++x}") + } + } + worker.requestTermination().result +} + +@Test fun runTest3() { + val worker = Worker.start() + assertFailsWith { + worker.executeAfter { + println("shall not happen") + } + } + assertFailsWith { + worker.executeAfter(-1, { + println("shall not happen") + }.freeze()) + } + + worker.executeAfter(0, { + println("frozen OK") + }.freeze()) + + worker.requestTermination().result } \ No newline at end of file diff --git a/runtime/src/main/cpp/Types.h b/runtime/src/main/cpp/Types.h index 9512f898395..88b61c8bfee 100644 --- a/runtime/src/main/cpp/Types.h +++ b/runtime/src/main/cpp/Types.h @@ -26,6 +26,7 @@ #include #include +#include #include #include #include @@ -69,6 +70,8 @@ template using KStdUnorderedSet = std::unordered_set, std::equal_to, KonanAllocator>; +template +using KStdOrderedSet = std::set>; template using KStdVector = std::vector>; diff --git a/runtime/src/main/cpp/Worker.cpp b/runtime/src/main/cpp/Worker.cpp index e6fb7a08dba..5077f64706b 100644 --- a/runtime/src/main/cpp/Worker.cpp +++ b/runtime/src/main/cpp/Worker.cpp @@ -59,6 +59,12 @@ enum { UNCHECKED = 1 }; +enum JobKind { + JOB_REGULAR = 1, + JOB_TERMINATE, + JOB_EXECUTE_AFTER +}; + THREAD_LOCAL_VARIABLE KInt g_currentWorkerId = 0; KNativePtr transfer(ObjHolder* holder, KInt mode) { @@ -92,10 +98,20 @@ class Future { } ~Future() { + clear(); pthread_mutex_destroy(&lock_); pthread_cond_destroy(&cond_); } + void clear() { + Locker locker(&lock_); + if (result_ != nullptr) { + // No one cared to consume result - dispose it. + DisposeStablePointer(result_); + result_ = nullptr; + } + } + OBJ_GETTER0(consumeResultUnlocked) { Locker locker(&lock_); while (state_ == SCHEDULED) { @@ -130,12 +146,36 @@ class Future { }; struct Job { - KRef (*function)(KRef, ObjHeader**); - KNativePtr argument; - Future* future; - KInt transferMode; + enum JobKind kind; + union { + struct { + KRef (*function)(KRef, ObjHeader**); + KNativePtr argument; + Future* future; + KInt transferMode; + } regularJob; + + struct { + Future* future; + bool waitDelayed; + } terminationRequest; + + struct { + KNativePtr operation; + KLong whenExecute; + } executeAfter; + }; }; +struct JobCompare { + bool operator() (const Job& lhs, const Job& rhs) const { + RuntimeAssert(lhs.kind == JOB_EXECUTE_AFTER && rhs.kind == JOB_EXECUTE_AFTER, "Must be delayed jobs"); + return lhs.executeAfter.whenExecute < rhs.executeAfter.whenExecute; + } +}; + +typedef KStdOrderedSet DelayedJobSet; + class Worker { public: Worker(KInt id, bool errorReporting) : id_(id), errorReporting_(errorReporting) { @@ -144,10 +184,29 @@ class Worker { } ~Worker() { - // Cleanup jobs in queue. + // Cleanup jobs in the queue. for (auto job : queue_) { - DisposeStablePointer(job.argument); - job.future->cancelUnlocked(); + switch (job.kind) { + case JOB_REGULAR: + DisposeStablePointer(job.regularJob.argument); + job.regularJob.future->cancelUnlocked(); + break; + case JOB_EXECUTE_AFTER: { + // TODO: what do we do here? Shall we execute them? + DisposeStablePointer(job.executeAfter.operation); + break; + } + case JOB_TERMINATE: { + // TODO: any more processing here? + job.terminationRequest.future->cancelUnlocked(); + break; + } + } + } + + for (auto job : delayed_) { + RuntimeAssert(job.kind == JOB_EXECUTE_AFTER, "Must be delayed"); + DisposeStablePointer(job.executeAfter.operation); } pthread_mutex_destroy(&lock_); @@ -157,22 +216,68 @@ class Worker { void putJob(Job job, bool toFront) { Locker locker(&lock_); if (toFront) - queue_.push_front(job); + queue_.push_front(job); else - queue_.push_back(job); + queue_.push_back(job); pthread_cond_signal(&cond_); } + void putDelayedJob(Job job) { + Locker locker(&lock_); + delayed_.insert(job); + pthread_cond_signal(&cond_); + } + + bool waitDelayed() { + Locker locker(&lock_); + if (delayed_.size() == 0) return false; + waitForQueueLocked(); + return true; + } + Job getJob() { Locker locker(&lock_); - while (queue_.size() == 0) { - pthread_cond_wait(&cond_, &lock_); - } + waitForQueueLocked(); auto result = queue_.front(); queue_.pop_front(); return result; } + KLong checkDelayedLocked() { + if (delayed_.size() == 0) { + return -1; + } + auto it = delayed_.begin(); + auto job = *it; + RuntimeAssert(job.kind == JOB_EXECUTE_AFTER, "Must be delayed job"); + auto now = konan::getTimeMicros(); + if (job.executeAfter.whenExecute <= now) { + delayed_.erase(it); + queue_.push_back(job); + return 0; + } else { + return job.executeAfter.whenExecute - now; + } + } + + void waitForQueueLocked() { + while (queue_.size() == 0) { + KLong closestToRun = checkDelayedLocked(); + if (closestToRun == 0) continue; + if (closestToRun > 0) { + struct timeval tv; + struct timespec ts; + gettimeofday(&tv, nullptr); + KLong nsDelta = closestToRun * 1000LL; + ts.tv_nsec = (tv.tv_usec * 1000LL + nsDelta) % 1000000000LL; + ts.tv_sec = (tv.tv_sec * 1000000000LL + nsDelta) / 1000000000LL; + pthread_cond_timedwait(&cond_, &lock_, &ts); + } else { + pthread_cond_wait(&cond_, &lock_); + } + } + } + KInt id() const { return id_; } bool errorReporting() const { return errorReporting_; } @@ -180,9 +285,11 @@ class Worker { private: KInt id_; KStdDeque queue_; + DelayedJobSet delayed_; // Lock and condition for waiting on the queue. pthread_mutex_t lock_; pthread_cond_t cond_; + // If errors to be reported on console. bool errorReporting_; }; @@ -222,28 +329,52 @@ class State { KInt id, KNativePtr jobFunction, KNativePtr jobArgument, bool toFront, KInt transferMode) { Future* future = nullptr; Worker* worker = nullptr; - { - Locker locker(&lock_); + Locker locker(&lock_); - auto it = workers_.find(id); - if (it == workers_.end()) return nullptr; - worker = it->second; + auto it = workers_.find(id); + if (it == workers_.end()) return nullptr; + worker = it->second; - future = konanConstructInstance(nextFutureId()); - futures_[future->id()] = future; - } + future = konanConstructInstance(nextFutureId()); + futures_[future->id()] = future; Job job; - job.function = reinterpret_cast(jobFunction); - job.argument = jobArgument; - job.future = future; - job.transferMode = transferMode; + if (jobFunction == nullptr) { + job.kind = JOB_TERMINATE; + job.terminationRequest.future = future; + job.terminationRequest.waitDelayed = !toFront; + } else { + job.kind = JOB_REGULAR; + job.regularJob.function = reinterpret_cast(jobFunction); + job.regularJob.argument = jobArgument; + job.regularJob.future = future; + job.regularJob.transferMode = transferMode; + } worker->putJob(job, toFront); return future; } + bool executeJobAfterInWorkerUnlocked(KInt id, KRef operation, KLong afterMicroseconds) { + Worker* worker = nullptr; + Locker locker(&lock_); + + auto it = workers_.find(id); + if (it == workers_.end()) return false; + worker = it->second; + Job job; + job.kind = JOB_EXECUTE_AFTER; + job.executeAfter.operation = CreateStablePointer(operation); + if (afterMicroseconds == 0) { + worker->putJob(job, false); + } else { + job.executeAfter.whenExecute = konan::getTimeMicros() + afterMicroseconds; + worker->putDelayedJob(job); + } + return true; + } + KInt stateOfFutureUnlocked(KInt id) { Locker locker(&lock_); auto it = futures_.find(id); @@ -285,7 +416,7 @@ class State { struct timeval tv; struct timespec ts; gettimeofday(&tv, nullptr); - KLong nsDelta = millis * 1000000LL; + KLong nsDelta = millis * 1000LL * 1000LL; ts.tv_nsec = (tv.tv_usec * 1000LL + nsDelta) % 1000000000LL; ts.tv_sec = (tv.tv_sec * 1000000000LL + nsDelta) / 1000000000LL; pthread_cond_timedwait(&cond_, &lock_, &ts); @@ -374,27 +505,46 @@ void* workerRoutine(void* argument) { ObjHolder resultHolder; while (true) { Job job = worker->getJob(); - if (job.function == nullptr) { + if (job.kind == JOB_TERMINATE) { + if (job.terminationRequest.waitDelayed) { + if (worker->waitDelayed()) { + worker->putJob(job, false); + continue; + } + } // Termination request, notify the future. - job.future->storeResultUnlocked(nullptr, true); + job.terminationRequest.future->storeResultUnlocked(nullptr, true); theState()->removeWorkerUnlocked(worker->id()); break; } - KRef argument = AdoptStablePointer(job.argument, argumentHolder.slot()); + if (job.kind == JOB_EXECUTE_AFTER) { + ObjHolder operationHolder, dummyHolder; + KRef obj = DerefStablePointer(job.executeAfter.operation, operationHolder.slot()); + try { + WorkerLaunchpad(obj, dummyHolder.slot()); + } catch (ExceptionObjHolder& e) { + if (worker->errorReporting()) + ReportUnhandledException(e.obj()); + } + DisposeStablePointer(job.executeAfter.operation); + continue; + } + RuntimeAssert(job.kind == JOB_REGULAR, "Must be regular job"); + KRef argument = AdoptStablePointer(job.regularJob.argument, argumentHolder.slot()); KNativePtr result = nullptr; bool ok = true; try { - job.function(argument, resultHolder.slot()); + job.regularJob.function(argument, resultHolder.slot()); argumentHolder.clear(); // Transfer the result. - result = transfer(&resultHolder, job.transferMode); + result = transfer(&resultHolder, job.regularJob.transferMode); } catch (ExceptionObjHolder& e) { ok = false; if (worker->errorReporting()) ReportUnhandledException(e.obj()); } // Notify the future. - job.future->storeResultUnlocked(result, ok); + job.regularJob.future->storeResultUnlocked(result, ok); } } @@ -417,7 +567,7 @@ KInt currentWorker() { return g_currentWorkerId; } -KInt schedule(KInt id, KInt transferMode, KRef producer, KNativePtr jobFunction) { +KInt execute(KInt id, KInt transferMode, KRef producer, KNativePtr jobFunction) { Job job; ObjHolder holder; WorkerLaunchpad(producer, holder.slot()); @@ -427,6 +577,11 @@ KInt schedule(KInt id, KInt transferMode, KRef producer, KNativePtr jobFunction) return future->id(); } +void executeAfter(KInt id, KRef job, KLong afterMicroseconds) { + if (!theState()->executeJobAfterInWorkerUnlocked(id, job, afterMicroseconds)) + ThrowWorkerInvalidState(); +} + KInt stateOfFuture(KInt id) { return theState()->stateOfFutureUnlocked(id); } @@ -476,11 +631,15 @@ KInt stateOfFuture(KInt id) { return 0; } -KInt schedule(KInt id, KInt transferMode, KRef producer, KNativePtr jobFunction) { +KInt execute(KInt id, KInt transferMode, KRef producer, KNativePtr jobFunction) { ThrowWorkerUnsupported(); return 0; } +void executeAfter(KInt id, KRef job, KLong afterMicroseconds) { + ThrowWorkerUnsupported(); +} + KInt currentWorker() { ThrowWorkerUnsupported(); return 0; @@ -531,11 +690,15 @@ KInt Kotlin_Worker_currentInternal() { } KInt Kotlin_Worker_requestTerminationWorkerInternal(KInt id, KBoolean processScheduledJobs) { - return requestTermination(id, processScheduledJobs); + return requestTermination(id, processScheduledJobs); } KInt Kotlin_Worker_executeInternal(KInt id, KInt transferMode, KRef producer, KNativePtr job) { - return schedule(id, transferMode, producer, job); + return execute(id, transferMode, producer, job); +} + +void Kotlin_Worker_executeAfterInternal(KInt id, KRef job, KLong afterMicroseconds) { + executeAfter(id, job, afterMicroseconds); } KInt Kotlin_Worker_stateOfFuture(KInt id) { diff --git a/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt b/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt index eff42cc4484..e63afa7fdc5 100644 --- a/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt +++ b/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt @@ -42,6 +42,9 @@ external internal fun requestTerminationInternal(id: Int, processScheduledJobs: external internal fun executeInternal( id: Int, mode: Int, producer: () -> Any?, job: CPointer>): Int +@SymbolName("Kotlin_Worker_executeAfterInternal") +external internal fun executeAfterInternal(id: Int, operation: () -> Unit, afterMicroseconds: Long): Unit + @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 9c26470e7bb..26b6d961ff6 100644 --- a/runtime/src/main/kotlin/kotlin/native/concurrent/Worker.kt +++ b/runtime/src/main/kotlin/kotlin/native/concurrent/Worker.kt @@ -61,7 +61,8 @@ public inline class Worker @PublishedApi internal constructor(val id: Int) { * Requests termination of the worker. * * @param processScheduledJobs controls is we shall wait until all scheduled jobs processed, - * or terminate immediately. + * or terminate immediately. If there are jobs to be execucted with [executeAfter] their execution + * is awaited for. */ public fun requestTermination(processScheduledJobs: Boolean = true) = Future(requestTerminationInternal(id, processScheduledJobs)) @@ -88,6 +89,20 @@ public inline class Worker @PublishedApi internal constructor(val id: Int) { */ throw RuntimeException("Shall not be called directly") + + /** + * 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. + */ + public fun executeAfter(afterMicroseconds: Long = 0, operation: () -> Unit): Unit { + val current = currentInternal() + if (current != id && !operation.isFrozen) throw IllegalStateException("Job for another worker must be frozen") + if (afterMicroseconds < 0) throw IllegalArgumentException("Timeout parameter must be non-negative") + executeAfterInternal(id, operation, afterMicroseconds) + } + override public fun toString(): String = "worker $id" /**