From afe0780d28813e1843a7dc4d8dd513009eeff8b2 Mon Sep 17 00:00:00 2001 From: Nikolay Igotti Date: Mon, 13 Nov 2017 16:42:00 +0300 Subject: [PATCH] Improve workers API, fix lost sync. (#1029) --- runtime/src/main/cpp/Worker.cpp | 24 +++++++++++++------ .../src/main/kotlin/konan/worker/Worker.kt | 19 ++++----------- 2 files changed, 22 insertions(+), 21 deletions(-) diff --git a/runtime/src/main/cpp/Worker.cpp b/runtime/src/main/cpp/Worker.cpp index 3869849ee66..659462ee7d5 100644 --- a/runtime/src/main/cpp/Worker.cpp +++ b/runtime/src/main/cpp/Worker.cpp @@ -158,9 +158,9 @@ 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_); } @@ -340,8 +340,11 @@ void Future::storeResultUnlocked(KNativePtr result) { Locker locker(&lock_); state_ = COMPUTED; result_ = result; + // Beware here: although manual clearly says that pthread_cond_signal() could be called outside + // of the taken lock, it's not on OSX (as of 10.13.1). If moved outside of the lock, + // some notifications gets missed. + pthread_cond_signal(&cond_); } - pthread_cond_signal(&cond_); theState()->signalAnyFuture(); } @@ -350,11 +353,13 @@ void Future::cancelUnlocked() { Locker locker(&lock_); state_ = CANCELLED; result_ = nullptr; + pthread_cond_signal(&cond_); } - pthread_cond_signal(&cond_); theState()->signalAnyFuture(); } +// Defined in RuntimeUtils.kt. +extern "C" void ReportUnhandledException(KRef e); void* workerRoutine(void* argument) { Worker* worker = reinterpret_cast(argument); @@ -374,9 +379,14 @@ void* workerRoutine(void* argument) { // so we don't use ObjHolder. // It is so, as ownership is transferred. KRef resultRef = nullptr; - job.function(argument, &resultRef); - // Transfer the result. - KNativePtr result = transfer(resultRef, job.transferMode); + KNativePtr result = nullptr; + try { + job.function(argument, &resultRef); + // Transfer the result. + result = transfer(resultRef, job.transferMode); + } catch (ObjHolder& e) { + ReportUnhandledException(e.obj()); + } // Notify the future. job.future->storeResultUnlocked(result); } diff --git a/runtime/src/main/kotlin/konan/worker/Worker.kt b/runtime/src/main/kotlin/konan/worker/Worker.kt index 6c4ff3613c7..57954011f5c 100644 --- a/runtime/src/main/kotlin/konan/worker/Worker.kt +++ b/runtime/src/main/kotlin/konan/worker/Worker.kt @@ -89,7 +89,7 @@ class Future internal constructor(val id: FutureId) { /** * Blocks execution until the future is ready. */ - fun consume(code: (T) -> Unit) { + fun consume(code: (T) -> R) = when (state) { FutureState.SCHEDULED, FutureState.COMPUTED -> { val value = @Suppress("UNCHECKED_CAST") (consumeFuture(id) as T) @@ -100,18 +100,13 @@ class Future internal constructor(val id: FutureId) { FutureState.CANCELLED -> throw IllegalStateException("Future is cancelled") } - } val state: FutureState get() = FutureState.values()[stateOfFuture(id)] - override fun equals(other: Any?): Boolean { - return (other is Future<*>) && (id == other.id) - } + override fun equals(other: Any?) = (other is Future<*>) && (id == other.id) - override fun hashCode(): Int { - return id - } + override fun hashCode() = id } /** @@ -145,13 +140,9 @@ class Worker(val id: WorkerId) { */ throw RuntimeException("Shall not be called directly") - override fun equals(other: Any?): Boolean { - return (other is Worker) && (id == other.id) - } + override fun equals(other: Any?) = (other is Worker) && (id == other.id) - override fun hashCode(): Int { - return id - } + override fun hashCode() = id } /**