Improve exception messages for kotlin.native.concurrent.* operations

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"

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<IllegalStateException> {
+        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<IllegalStateException> {
+        DetachedObjectGraph { obj }
+    }
+    assertEquals("Unable to transfer object: it is still owned elsewhere", exception.message)
+}
+
+@Test
+fun checkProcessQueueOnWrongThread(): Unit = withWorker {
+    val exception = assertFailsWith<IllegalStateException> {
+        processQueue()
+    }
+    assertEquals("Worker is not current or already terminated", exception.message)
+}
+
+@Test
+fun checkParkOnWrongThread(): Unit = withWorker {
+    val exception = assertFailsWith<IllegalStateException> {
+        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<IllegalStateException> {
+        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<IllegalStateException> {
+        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<IllegalStateException> {
+        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<IllegalStateException> {
+        worker.executeAfter(0L, {}.freeze())
+    }
+    assertEquals("Worker is already terminated", exception.message)
+}
+
+@Test
+fun checkTerminatedWorkerRequestTermination() {
+    val worker = Worker.start()
+    worker.requestTermination().result
+
+    val exception = assertFailsWith<IllegalStateException> {
+        worker.requestTermination()
+    }
+    assertEquals("Worker is already terminated", exception.message)
+}

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();
 }
 

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()

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");
 }