From 2a54f1c6105f00264edf2af3a86138a9de9e598a Mon Sep 17 00:00:00 2001 From: Svyatoslav Scherbina Date: Fri, 18 Jun 2021 13:51:21 +0000 Subject: [PATCH] Improve exception messages for kotlin.native.concurrent.* operations --- .../backend.native/tests/build.gradle | 5 + .../workers/worker_exception_messages.kt | 103 ++++++++++++++++++ kotlin-native/runtime/src/main/cpp/Worker.cpp | 27 +++-- .../kotlin/native/concurrent/Internal.kt | 20 +++- .../test_support/cpp/CompilerGenerated.cpp | 14 ++- 5 files changed, 154 insertions(+), 15 deletions(-) create mode 100644 kotlin-native/backend.native/tests/runtime/workers/worker_exception_messages.kt diff --git a/kotlin-native/backend.native/tests/build.gradle b/kotlin-native/backend.native/tests/build.gradle index 02674f73b52..9fba27cb847 100644 --- a/kotlin-native/backend.native/tests/build.gradle +++ b/kotlin-native/backend.native/tests/build.gradle @@ -970,6 +970,11 @@ task worker11(type: KonanLocalTest) { source = "runtime/workers/worker11.kt" } +task worker_exception_messages(type: KonanLocalTest) { + enabled = (project.testTarget != 'wasm32') // Workers need pthreads. + source = "runtime/workers/worker_exception_messages.kt" +} + standaloneTest("worker_threadlocal_no_leak") { disabled = (project.testTarget == 'wasm32') // Needs pthreads. source = "runtime/workers/worker_threadlocal_no_leak.kt" diff --git a/kotlin-native/backend.native/tests/runtime/workers/worker_exception_messages.kt b/kotlin-native/backend.native/tests/runtime/workers/worker_exception_messages.kt new file mode 100644 index 00000000000..d5b450c0823 --- /dev/null +++ b/kotlin-native/backend.native/tests/runtime/workers/worker_exception_messages.kt @@ -0,0 +1,103 @@ +package runtime.workers.worker_exception_messages + +import kotlin.test.* + +import kotlin.native.concurrent.* + +@Test +fun checkArgumentTransferFailed(): Unit = withWorker { + if (Platform.memoryModel == MemoryModel.EXPERIMENTAL) return // Transfer is no-op in this case. + + val argument = Any() + val exception = assertFailsWith { + execute(TransferMode.SAFE, { argument }) { + } + } + assertEquals("Unable to transfer object: it is still owned elsewhere", exception.message) +} + +@Test +fun checkDetachedObjectGraphTransferFailed() { + if (Platform.memoryModel == MemoryModel.EXPERIMENTAL) return // Transfer is no-op in this case. + + val obj = Any() + val exception = assertFailsWith { + DetachedObjectGraph { obj } + } + assertEquals("Unable to transfer object: it is still owned elsewhere", exception.message) +} + +@Test +fun checkProcessQueueOnWrongThread(): Unit = withWorker { + val exception = assertFailsWith { + processQueue() + } + assertEquals("Worker is not current or already terminated", exception.message) +} + +@Test +fun checkParkOnWrongThread(): Unit = withWorker { + val exception = assertFailsWith { + park(1L) + } + assertEquals("Worker is not current or already terminated", exception.message) +} + +@Test +fun checkFutureConsumedTwice(): Unit = withWorker { + val future = execute(TransferMode.SAFE, {}) { + 42 + } + assertEquals(42, future.result) + val exception = assertFailsWith { + future.result + } + assertEquals("Future is in an invalid state", exception.message) +} + +@Test +fun checkTerminatedWorkerName() { + val worker = Worker.start(name = "WorkerName") + assertEquals("WorkerName", worker.name) + worker.requestTermination().result + + val exception = assertFailsWith { + worker.name + } + assertEquals("Worker is already terminated", exception.message) +} + +@Test +fun checkTerminatedWorkerExecute() { + val worker = Worker.start() + worker.execute(TransferMode.SAFE, {}, {}).result + worker.requestTermination().result + + val exception = assertFailsWith { + worker.execute(TransferMode.SAFE, {}, {}).result + } + assertEquals("Worker is already terminated", exception.message) +} + +@Test +fun checkTerminatedWorkerExecuteAfter() { + val worker = Worker.start() + worker.executeAfter(0L, {}.freeze()) + worker.requestTermination().result + + val exception = assertFailsWith { + worker.executeAfter(0L, {}.freeze()) + } + assertEquals("Worker is already terminated", exception.message) +} + +@Test +fun checkTerminatedWorkerRequestTermination() { + val worker = Worker.start() + worker.requestTermination().result + + val exception = assertFailsWith { + worker.requestTermination() + } + assertEquals("Worker is already terminated", exception.message) +} diff --git a/kotlin-native/runtime/src/main/cpp/Worker.cpp b/kotlin-native/runtime/src/main/cpp/Worker.cpp index 75bc87a8f60..6e1643072e8 100644 --- a/kotlin-native/runtime/src/main/cpp/Worker.cpp +++ b/kotlin-native/runtime/src/main/cpp/Worker.cpp @@ -40,7 +40,10 @@ using namespace kotlin; extern "C" { -RUNTIME_NORETURN void ThrowWorkerInvalidState(); +RUNTIME_NORETURN void ThrowWorkerAlreadyTerminated(); +RUNTIME_NORETURN void ThrowWrongWorkerOrAlreadyTerminated(); +RUNTIME_NORETURN void ThrowCannotTransferOwnership(); +RUNTIME_NORETURN void ThrowFutureInvalidState(); RUNTIME_NORETURN void ThrowWorkerUnsupported(); OBJ_GETTER(WorkerLaunchpad, KRef); @@ -202,7 +205,7 @@ KNativePtr transfer(ObjHolder* holder, KInt mode) { void* result = CreateStablePointer(holder->obj()); if (!ClearSubgraphReferences(holder->obj(), mode == CHECKED)) { DisposeStablePointer(result); - ThrowWorkerInvalidState(); + ThrowCannotTransferOwnership(); } holder->clear(); return result; @@ -430,14 +433,14 @@ class State { // Returns `true` if something was indeed processed. bool processQueueUnlocked(KInt id) { // Can only process queue of the current worker. - if (::g_worker == nullptr || id != ::g_worker->id()) ThrowWorkerInvalidState(); + if (::g_worker == nullptr || id != ::g_worker->id()) ThrowWrongWorkerOrAlreadyTerminated(); JobKind kind = ::g_worker->processQueueElement(false); 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(); + if (::g_worker == nullptr || id != ::g_worker->id()) ThrowWrongWorkerOrAlreadyTerminated(); return ::g_worker->park(timeoutMicroseconds, process); } @@ -453,7 +456,11 @@ class State { { Locker locker(&lock_); auto it = futures_.find(id); - if (it == futures_.end()) ThrowWorkerInvalidState(); + if (it == futures_.end()) { + // Caller checks [stateOfFutureUnlocked] first, so this code is reachable + // only when trying to consume future twice concurrently. + ThrowFutureInvalidState(); + } future = it->second; } @@ -477,7 +484,7 @@ class State { Locker locker(&lock_); auto it = workers_.find(id); if (it == workers_.end()) { - ThrowWorkerInvalidState(); + ThrowWorkerAlreadyTerminated(); } DerefStablePointer(it->second->name(), nameHolder.slot()); } @@ -643,7 +650,7 @@ KInt startWorker(KBoolean errorReporting, KRef customName) { } KInt currentWorker() { - if (g_worker == nullptr) ThrowWorkerInvalidState(); + if (g_worker == nullptr) ThrowWorkerAlreadyTerminated(); return ::g_worker->id(); } @@ -652,13 +659,13 @@ KInt execute(KInt id, KInt transferMode, KRef producer, KNativePtr jobFunction) WorkerLaunchpad(producer, holder.slot()); KNativePtr jobArgument = transfer(&holder, transferMode); Future* future = theState()->addJobToWorkerUnlocked(id, jobFunction, jobArgument, false, transferMode); - if (future == nullptr) ThrowWorkerInvalidState(); + if (future == nullptr) ThrowWorkerAlreadyTerminated(); return future->id(); } void executeAfter(KInt id, KRef job, KLong afterMicroseconds) { if (!theState()->executeJobAfterInWorkerUnlocked(id, job, afterMicroseconds)) - ThrowWorkerInvalidState(); + ThrowWorkerAlreadyTerminated(); } KBoolean processQueue(KInt id) { @@ -684,7 +691,7 @@ OBJ_GETTER(getWorkerName, KInt id) { KInt requestTermination(KInt id, KBoolean processScheduledJobs) { Future* future = theState()->addJobToWorkerUnlocked( id, nullptr, nullptr, /* toFront = */ !processScheduledJobs, UNCHECKED); - if (future == nullptr) ThrowWorkerInvalidState(); + if (future == nullptr) ThrowWorkerAlreadyTerminated(); return future->id(); } diff --git a/kotlin-native/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt b/kotlin-native/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt index 58ee42f71fc..302e01a2418 100644 --- a/kotlin-native/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt +++ b/kotlin-native/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt @@ -60,12 +60,24 @@ external internal fun parkInternal(id: Int, timeoutMicroseconds: Long, process: external internal fun getWorkerNameInternal(id: Int): String? @ExportForCppRuntime -internal fun ThrowWorkerUnsupported(): Unit = - throw UnsupportedOperationException("Workers are not supported") +internal fun ThrowWorkerAlreadyTerminated(): Unit = + throw IllegalStateException("Worker is already terminated") @ExportForCppRuntime -internal fun ThrowWorkerInvalidState(): Unit = - throw IllegalStateException("Illegal transfer state") +internal fun ThrowWrongWorkerOrAlreadyTerminated(): Unit = + throw IllegalStateException("Worker is not current or already terminated") + +@ExportForCppRuntime +internal fun ThrowCannotTransferOwnership(): Unit = + throw IllegalStateException("Unable to transfer object: it is still owned elsewhere") + +@ExportForCppRuntime +internal fun ThrowFutureInvalidState(): Unit = + throw IllegalStateException("Future is in an invalid state") + +@ExportForCppRuntime +internal fun ThrowWorkerUnsupported(): Unit = + throw UnsupportedOperationException("Workers are not supported") @ExportForCppRuntime internal fun WorkerLaunchpad(function: () -> Any?) = function() diff --git a/kotlin-native/runtime/src/test_support/cpp/CompilerGenerated.cpp b/kotlin-native/runtime/src/test_support/cpp/CompilerGenerated.cpp index 03f5c9455b0..2776aaa55ca 100644 --- a/kotlin-native/runtime/src/test_support/cpp/CompilerGenerated.cpp +++ b/kotlin-native/runtime/src/test_support/cpp/CompilerGenerated.cpp @@ -113,7 +113,19 @@ RUNTIME_NORETURN OBJ_GETTER(WorkerLaunchpad, KRef) { throw std::runtime_error("Not implemented for tests"); } -void RUNTIME_NORETURN ThrowWorkerInvalidState() { +void RUNTIME_NORETURN ThrowWorkerAlreadyTerminated() { + throw std::runtime_error("Not implemented for tests"); +} + +void RUNTIME_NORETURN ThrowWrongWorkerOrAlreadyTerminated() { + throw std::runtime_error("Not implemented for tests"); +} + +void RUNTIME_NORETURN ThrowCannotTransferOwnership() { + throw std::runtime_error("Not implemented for tests"); +} + +void RUNTIME_NORETURN ThrowFutureInvalidState() { throw std::runtime_error("Not implemented for tests"); }