diff --git a/backend.native/tests/build.gradle b/backend.native/tests/build.gradle index f8859963266..3bf4c76e38d 100644 --- a/backend.native/tests/build.gradle +++ b/backend.native/tests/build.gradle @@ -873,7 +873,8 @@ task worker4(type: KonanLocalTest) { source = "runtime/workers/worker4.kt" } -task worker5(type: KonanLocalTest) { +// This tests changes main thread worker queue state, so better be executed alone. +standaloneTest("worker5") { enabled = (project.testTarget != 'wasm32') // Workers need pthreads. goldValue = "Got 3\nOK\n" source = "runtime/workers/worker5.kt" diff --git a/backend.native/tests/runtime/workers/worker11.kt b/backend.native/tests/runtime/workers/worker11.kt index 533a15773e6..85e136e8cc9 100644 --- a/backend.native/tests/runtime/workers/worker11.kt +++ b/backend.native/tests/runtime/workers/worker11.kt @@ -98,7 +98,7 @@ val counters = Array(COUNT) { AtomicInt(0) } val futures = Array(workers.size) { workerIndex -> workers[workerIndex].execute(TransferMode.SAFE, { null }) { // Here we processed termination request. - assertEquals(false, Worker.current!!.processQueue()) + assertEquals(false, Worker.current.processQueue()) } } workers.forEach { diff --git a/backend.native/tests/runtime/workers/worker5.kt b/backend.native/tests/runtime/workers/worker5.kt index 16e825c1b45..7463eb026b1 100644 --- a/backend.native/tests/runtime/workers/worker5.kt +++ b/backend.native/tests/runtime/workers/worker5.kt @@ -3,13 +3,11 @@ * that can be found in the LICENSE file. */ -package runtime.workers.worker5 - import kotlin.test.* import kotlin.native.concurrent.* -@Test fun runTest() { +@Test fun runTest0() { val worker = Worker.start() val future = worker.execute(TransferMode.SAFE, { "zzz" }) { input -> input.length @@ -19,4 +17,36 @@ import kotlin.native.concurrent.* } worker.requestTermination().result println("OK") +} + +var done = false + +@Test fun runTest1() { + val worker = Worker.current + done = false + // Here we request execution of the operation on the current worker. + worker.executeAfter(0, { + done = true + }.freeze()) + while (!done) + worker.processQueue() +} + +// Ensure that termination of current worker on main thread doesn't lead to problems. +@Test fun runTest2() { + val worker = Worker.current + val future = worker.requestTermination(false) + worker.processQueue() + assertEquals(future.state, FutureState.COMPUTED) + future.consume {} + // After termination request this worker is no longer addressable. + assertFailsWith { worker.executeAfter(0, { + println("BUG!") + }.freeze()) } +} + +fun main() { + runTest0() + runTest1() + runTest2() } \ No newline at end of file diff --git a/runtime/src/main/cpp/Runtime.cpp b/runtime/src/main/cpp/Runtime.cpp index f47128c2999..a492ce3af5d 100644 --- a/runtime/src/main/cpp/Runtime.cpp +++ b/runtime/src/main/cpp/Runtime.cpp @@ -21,15 +21,17 @@ #include "Memory.h" #include "Porting.h" #include "Runtime.h" +#include "Worker.h" struct RuntimeState { MemoryState* memoryState; + Worker* worker; volatile int executionStatus; }; typedef void (*Initializer)(int initialize); struct InitNode { - Initializer init; + Initializer init; InitNode* next; }; @@ -89,6 +91,7 @@ RuntimeState* initRuntime() { RuntimeCheck(!isValidRuntime(), "No active runtimes allowed"); ::runtimeState = result; result->memoryState = InitMemory(); + result->worker = WorkerInit(true); bool firstRuntime = atomicAdd(&aliveRuntimesCount, 1) == 1; // Keep global variables in state as well. if (firstRuntime) { @@ -106,6 +109,7 @@ void deinitRuntime(RuntimeState* state) { InitOrDeinitGlobalVariables(DEINIT_THREAD_LOCAL_GLOBALS); if (lastRuntime) InitOrDeinitGlobalVariables(DEINIT_GLOBALS); + WorkerDeinit(state->worker); DeinitMemory(state->memoryState); konanDestructInstance(state); } @@ -160,6 +164,7 @@ RuntimeState* Kotlin_suspendRuntime() { auto result = ::runtimeState; RuntimeCheck(updateStatusIf(result, RUNNING, SUSPENDED), "Cannot transition state to SUSPENDED for suspend"); result->memoryState = SuspendMemory(); + result->worker = WorkerSuspend(); ::runtimeState = kInvalidRuntime; return result; } @@ -169,6 +174,7 @@ void Kotlin_resumeRuntime(RuntimeState* state) { RuntimeCheck(updateStatusIf(state, SUSPENDED, RUNNING), "Cannot transition state to RUNNING for resume"); ::runtimeState = state; ResumeMemory(state->memoryState); + WorkerResume(state->worker); } RuntimeState* RUNTIME_USED Kotlin_getRuntime() { diff --git a/runtime/src/main/cpp/Worker.cpp b/runtime/src/main/cpp/Worker.cpp index 1cb1ea06aa4..51d46aae49d 100644 --- a/runtime/src/main/cpp/Worker.cpp +++ b/runtime/src/main/cpp/Worker.cpp @@ -33,6 +33,7 @@ #include "Memory.h" #include "Runtime.h" #include "Types.h" +#include "Worker.h" extern "C" { @@ -42,9 +43,11 @@ OBJ_GETTER(WorkerLaunchpad, KRef); } // extern "C" +#if WITH_WORKERS + namespace { -#if WITH_WORKERS +class Future; enum { INVALID = 0, @@ -66,7 +69,89 @@ enum JobKind { JOB_EXECUTE_AFTER = 3 }; -THREAD_LOCAL_VARIABLE KInt g_currentWorkerId = 0; +struct Job { + 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; + +} // namespace + +class Worker { + public: + Worker(KInt id, bool errorReporting) : id_(id), errorReporting_(errorReporting), terminated_(false) { + pthread_mutex_init(&lock_, nullptr); + pthread_cond_init(&cond_, nullptr); + } + + ~Worker(); + + void putJob(Job job, bool toFront); + void putDelayedJob(Job job); + + bool waitDelayed(bool blocking); + + Job getJob(bool blocking); + + KLong checkDelayedLocked(); + + void waitForQueueLocked(); + + JobKind processQueueElement(bool blocking); + + KInt id() const { return id_; } + + bool errorReporting() const { return errorReporting_; } + + 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_; + bool terminated_; +}; + +#else // WITH_WORKERS +class Worker { + KInt id; +}; + +#endif // WITH_WORKERS + +namespace { + +#if WITH_WORKERS + +THREAD_LOCAL_VARIABLE Worker* g_worker = nullptr; KNativePtr transfer(ObjHolder* holder, KInt mode) { void* result = CreateStablePointer(holder->obj()); @@ -146,163 +231,6 @@ class Future { pthread_cond_t cond_; }; -struct Job { - 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), terminated_(false) { - pthread_mutex_init(&lock_, nullptr); - pthread_cond_init(&cond_, nullptr); - } - - ~Worker() { - // Cleanup jobs in the queue. - for (auto job : queue_) { - 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; - } - case JOB_NONE: { - RuntimeCheck(false, "Cannot be in queue"); - break; - } - } - } - - for (auto job : delayed_) { - RuntimeAssert(job.kind == JOB_EXECUTE_AFTER, "Must be delayed"); - DisposeStablePointer(job.executeAfter.operation); - } - - pthread_mutex_destroy(&lock_); - pthread_cond_destroy(&cond_); - } - - void putJob(Job job, bool toFront) { - Locker locker(&lock_); - if (toFront) - queue_.push_front(job); - else - queue_.push_back(job); - pthread_cond_signal(&cond_); - } - - void putDelayedJob(Job job) { - Locker locker(&lock_); - delayed_.insert(job); - pthread_cond_signal(&cond_); - } - - bool waitDelayed(bool blocking) { - Locker locker(&lock_); - if (delayed_.size() == 0) return false; - if (blocking) - waitForQueueLocked(); - return true; - } - - Job getJob(bool blocking) { - Locker locker(&lock_); - RuntimeAssert(!terminated_, "Must not be terminated"); - if (queue_.size() == 0 && !blocking) return Job { .kind = JOB_NONE }; - 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_); - } - } - } - - JobKind processQueueElement(bool blocking); - - KInt id() const { return id_; } - - bool errorReporting() const { return errorReporting_; } - - 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_; - bool terminated_; -}; class State { public: @@ -336,6 +264,18 @@ class State { workers_.erase(it); } + void destroyWorkerUnlocked(Worker* worker) { + { + Locker locker(&lock_); + auto id = worker->id(); + auto it = workers_.find(id); + if (it != workers_.end()) { + workers_.erase(it); + } + } + konanDestructInstance(worker); + } + Future* addJobToWorkerUnlocked( KInt id, KNativePtr jobFunction, KNativePtr jobArgument, bool toFront, KInt transferMode) { Future* future = nullptr; @@ -372,7 +312,9 @@ class State { Locker locker(&lock_); auto it = workers_.find(id); - if (it == workers_.end()) return false; + if (it == workers_.end()) { + return false; + } worker = it->second; Job job; job.kind = JOB_EXECUTE_AFTER; @@ -389,15 +331,8 @@ class State { // Returns `true` if something was indeed processed. bool processQueueUnlocked(KInt id) { // Can only process queue of the current worker. - if (id != g_currentWorkerId) ThrowWorkerInvalidState(); - Worker* worker = nullptr; - { - Locker locker(&lock_); - auto it = workers_.find(id); - if (it == workers_.end()) return false; - worker = it->second; - } - JobKind kind = worker->processQueueElement(false); + if (::g_worker == nullptr || id != ::g_worker->id()) ThrowWorkerInvalidState(); + JobKind kind = ::g_worker->processQueueElement(false); return kind != JOB_NONE && kind != JOB_TERMINATE; } @@ -415,6 +350,7 @@ class State { auto it = futures_.find(id); if (it == futures_.end()) ThrowWorkerInvalidState(); future = it->second; + } KRef result = future->consumeResultUnlocked(OBJ_RESULT); @@ -499,10 +435,10 @@ void Future::storeResultUnlocked(KNativePtr result, bool ok) { Locker locker(&lock_); state_ = ok ? COMPUTED : THROWN; result_ = result; - // Beware here: although manual clearly says that pthread_cond_signal() could be called outside + // Beware here: although manual clearly says that pthread_cond_broadcast() could be called outside // of the taken lock, it's not on macOS (as of 10.13.1). If moved outside of the lock, // some notifications are missing. - pthread_cond_signal(&cond_); + pthread_cond_broadcast(&cond_); } theState()->signalAnyFuture(); } @@ -512,7 +448,7 @@ void Future::cancelUnlocked() { Locker locker(&lock_); state_ = CANCELLED; result_ = nullptr; - pthread_cond_signal(&cond_); + pthread_cond_broadcast(&cond_); } theState()->signalAnyFuture(); } @@ -520,77 +456,16 @@ void Future::cancelUnlocked() { // Defined in RuntimeUtils.kt. extern "C" void ReportUnhandledException(KRef e); -JobKind Worker::processQueueElement(bool blocking) { - ObjHolder argumentHolder; - ObjHolder resultHolder; - if (terminated_) return JOB_TERMINATE; - Job job = getJob(blocking); - switch (job.kind) { - case JOB_NONE: { - break; - } - case JOB_TERMINATE: { - if (job.terminationRequest.waitDelayed) { - if (waitDelayed(blocking)) { - putJob(job, false); - return JOB_NONE; - } - } - terminated_ = true; - // Termination request, remove the worker and notify the future. - theState()->removeWorkerUnlocked(id()); - job.terminationRequest.future->storeResultUnlocked(nullptr, true); - break; - } - case JOB_EXECUTE_AFTER: { - ObjHolder operationHolder, dummyHolder; - KRef obj = DerefStablePointer(job.executeAfter.operation, operationHolder.slot()); - try { - WorkerLaunchpad(obj, dummyHolder.slot()); - } catch (ExceptionObjHolder& e) { - if (errorReporting()) - ReportUnhandledException(e.obj()); - } - DisposeStablePointer(job.executeAfter.operation); - break; - } - case JOB_REGULAR: { - KRef argument = AdoptStablePointer(job.regularJob.argument, argumentHolder.slot()); - KNativePtr result = nullptr; - bool ok = true; - try { - job.regularJob.function(argument, resultHolder.slot()); - argumentHolder.clear(); - // Transfer the result. - result = transfer(&resultHolder, job.regularJob.transferMode); - } catch (ExceptionObjHolder& e) { - ok = false; - if (errorReporting()) - ReportUnhandledException(e.obj()); - } - // Notify the future. - job.regularJob.future->storeResultUnlocked(result, ok); - break; - } - default: { - RuntimeCheck(false, "Must be exhaustive"); - } - } - return job.kind; -} - void* workerRoutine(void* argument) { Worker* worker = reinterpret_cast(argument); - g_currentWorkerId = worker->id(); + WorkerResume(worker); Kotlin_initRuntimeIfNeeded(); do { if (worker->processQueueElement(true) == JOB_TERMINATE) break; } while (true); - konanDestructInstance(worker); - return nullptr; } @@ -603,7 +478,8 @@ KInt startWorker(KBoolean errorReporting) { } KInt currentWorker() { - return g_currentWorkerId; + if (g_worker == nullptr) ThrowWorkerInvalidState(); + return ::g_worker->id(); } KInt execute(KInt id, KInt transferMode, KRef producer, KNativePtr jobFunction) { @@ -726,6 +602,204 @@ KNativePtr detachObjectGraphInternal(KInt transferMode, KRef producer) { } // namespace +Worker* WorkerInit(KBoolean errorReporting) { +#if WITH_WORKERS + if (::g_worker != nullptr) return ::g_worker; + Worker* worker = theState()->addWorkerUnlocked(errorReporting != 0); + ::g_worker = worker; + return worker; +#else + return nullptr; +#endif // WITH_WORKERS +} + +void WorkerDeinit(Worker* worker) { +#if WITH_WORKERS + ::g_worker = nullptr; + theState()->destroyWorkerUnlocked(worker); +#endif // WITH_WORKERS +} + +Worker* WorkerSuspend() { +#if WITH_WORKERS + auto* result = ::g_worker; + ::g_worker = nullptr; + return result; +#else + return nullptr; +#endif // WITH_WORKERS +} + +void WorkerResume(Worker* worker) { +#if WITH_WORKERS + ::g_worker = worker; +#endif // WITH_WORKERS +} + +#if WITH_WORKERS + +Worker::~Worker() { + // Cleanup jobs in the queue. + for (auto job : queue_) { + 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; + } + case JOB_NONE: { + RuntimeCheck(false, "Cannot be in queue"); + break; + } + } + } + + for (auto job : delayed_) { + RuntimeAssert(job.kind == JOB_EXECUTE_AFTER, "Must be delayed"); + DisposeStablePointer(job.executeAfter.operation); + } + + pthread_mutex_destroy(&lock_); + pthread_cond_destroy(&cond_); +} + +void Worker::putJob(Job job, bool toFront) { + Locker locker(&lock_); + if (toFront) + queue_.push_front(job); + else + queue_.push_back(job); + pthread_cond_signal(&cond_); +} + +void Worker::putDelayedJob(Job job) { + Locker locker(&lock_); + delayed_.insert(job); + pthread_cond_signal(&cond_); +} + +bool Worker::waitDelayed(bool blocking) { + Locker locker(&lock_); + if (delayed_.size() == 0) return false; + if (blocking) waitForQueueLocked(); + return true; +} + +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(); + auto result = queue_.front(); + queue_.pop_front(); + return result; +} + +KLong Worker::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 Worker::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_); + } + } +} + +JobKind Worker::processQueueElement(bool blocking) { + ObjHolder argumentHolder; + ObjHolder resultHolder; + if (terminated_) return JOB_TERMINATE; + Job job = getJob(blocking); + switch (job.kind) { + case JOB_NONE: { + break; + } + case JOB_TERMINATE: { + if (job.terminationRequest.waitDelayed) { + if (waitDelayed(blocking)) { + putJob(job, false); + return JOB_NONE; + } + } + terminated_ = true; + // Termination request, remove the worker and notify the future. + theState()->removeWorkerUnlocked(id()); + job.terminationRequest.future->storeResultUnlocked(nullptr, true); + break; + } + case JOB_EXECUTE_AFTER: { + ObjHolder operationHolder, dummyHolder; + KRef obj = DerefStablePointer(job.executeAfter.operation, operationHolder.slot()); + try { + WorkerLaunchpad(obj, dummyHolder.slot()); + } catch (ExceptionObjHolder& e) { + if (errorReporting()) + ReportUnhandledException(e.obj()); + } + DisposeStablePointer(job.executeAfter.operation); + break; + } + case JOB_REGULAR: { + KRef argument = AdoptStablePointer(job.regularJob.argument, argumentHolder.slot()); + KNativePtr result = nullptr; + bool ok = true; + try { + job.regularJob.function(argument, resultHolder.slot()); + argumentHolder.clear(); + // Transfer the result. + result = transfer(&resultHolder, job.regularJob.transferMode); + } catch (ExceptionObjHolder& e) { + ok = false; + if (errorReporting()) + ReportUnhandledException(e.obj()); + } + // Notify the future. + job.regularJob.future->storeResultUnlocked(result, ok); + break; + } + default: { + RuntimeCheck(false, "Must be exhaustive"); + } + } + return job.kind; +} + +#endif // WITH_WORKERS + extern "C" { KInt Kotlin_Worker_startInternal(KBoolean noErrorReporting) { diff --git a/runtime/src/main/cpp/Worker.h b/runtime/src/main/cpp/Worker.h new file mode 100644 index 00000000000..885c9bc7a9a --- /dev/null +++ b/runtime/src/main/cpp/Worker.h @@ -0,0 +1,15 @@ +#ifndef RUNTIME_WORKER_H +#define RUNTIME_WORKER_H + +#include "Common.h" +#include "Types.h" + +class Worker; + +Worker* WorkerInit(KBoolean errorReporting); +void WorkerDeinit(Worker* worker); + +Worker* WorkerSuspend(); +void WorkerResume(Worker* worker); + +#endif // RUNTIME_WORKER_H \ No newline at end of file diff --git a/runtime/src/main/kotlin/kotlin/native/concurrent/Worker.kt b/runtime/src/main/kotlin/kotlin/native/concurrent/Worker.kt index bb1c1a0c666..81fc891031a 100644 --- a/runtime/src/main/kotlin/kotlin/native/concurrent/Worker.kt +++ b/runtime/src/main/kotlin/kotlin/native/concurrent/Worker.kt @@ -39,14 +39,11 @@ public inline class Worker @PublishedApi internal constructor(val id: Int) { public fun start(errorReporting: Boolean = true): Worker = Worker(startInternal(errorReporting)) /** - * Return the current worker, if known, null otherwise. null value will be returned in the main thread - * or platform thread without an associated worker, non-null - if called inside worker started with - * [Worker.start]. + * 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. */ - public val current: Worker? get() { - val id = currentInternal() - return if (id != 0) Worker(id) else null - } + public val current: Worker get() = Worker(currentInternal()) /** * Create worker object from a C pointer. @@ -92,7 +89,6 @@ 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. @@ -106,7 +102,6 @@ public inline class Worker @PublishedApi internal constructor(val id: Int) { executeAfterInternal(id, operation, afterMicroseconds) } - /** * 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