API for object passing between platform threads. (#1270)

backend.native/tests/build.gradle

@@ -594,6 +594,12 @@ task worker7(type: RunKonanTest) {
     source = "runtime/workers/worker7.kt"
 }
 
+task worker8(type: RunKonanTest) {
+    disabled = (project.testTarget == 'wasm32') // Workers need pthreads.
+    goldValue = "SharedData(string=Hello, int=10, member=SharedDataMember(double=0.1))\nGot kotlin.Unit\nOK\n"
+    source = "runtime/workers/worker8.kt"
+}
+
 task superFunCall(type: RunKonanTest) {
     goldValue = "<fun:C><fun:C1>\n<fun:C><fun:C3>\n"
     source = "codegen/basics/superFunCall.kt"

backend.native/tests/runtime/workers/worker8.kt

@@ -0,0 +1,27 @@
+package runtime.workers.worker8
+
+import kotlin.test.*
+
+import konan.worker.*
+
+data class SharedDataMember(val double: Double)
+
+data class SharedData(val string: String, val int: Int, val member: SharedDataMember)
+
+@Test fun runTest() {
+    val worker = startWorker()
+    // Here we do rather strange thing. To test object detach API we detach object graph,
+    // pass C pointer as a value to worker, where we manually reattached passed value.
+    val future = worker.schedule(TransferMode.CHECKED, {
+        detachObjectGraph { SharedData("Hello", 10, SharedDataMember(0.1)) }
+    } ) {
+        inputC ->
+        val input = attachObjectGraph<SharedData>(inputC)
+        println(input)
+    }
+    future.consume {
+        result -> println("Got $result")
+    }
+    worker.requestTermination().consume { _ -> }
+    println("OK")
+}

runtime/src/main/cpp/Assert.h

@@ -17,7 +17,9 @@
 #ifndef RUNTIME_ASSERT_H
 #define RUNTIME_ASSERT_H
 
-void RuntimeAssertFailed(const char* location, const char* message);
+#include "Common.h"
+
+RUNTIME_NORETURN void RuntimeAssertFailed(const char* location, const char* message);
 
 #define STRINGIFY(x) #x
 #define TOSTRING(x) STRINGIFY(x)

runtime/src/main/cpp/Common.h

@@ -18,6 +18,7 @@
 #define RUNTIME_COMMON_H
 
 #define RUNTIME_NOTHROW __attribute__((nothrow))
+#define RUNTIME_NORETURN __attribute__((noreturn))
 #define RUNTIME_CONST __attribute__((const))
 #define RUNTIME_PURE __attribute__((pure))
 #define RUNTIME_USED __attribute__((used))

runtime/src/main/cpp/Exceptions.h

@@ -34,22 +34,22 @@ void SetKonanTerminateHandler();
 // The functions below are implemented in Kotlin (at package konan.internal).
 
 // Throws null pointer exception. Context is evaluated from caller's address.
-void ThrowNullPointerException();
+void RUNTIME_NORETURN ThrowNullPointerException();
 // Throws array index out of bounds exception.
 // Context is evaluated from caller's address.
-void ThrowArrayIndexOutOfBoundsException();
+void RUNTIME_NORETURN ThrowArrayIndexOutOfBoundsException();
 // Throws class cast exception.
-void ThrowClassCastException();
+void RUNTIME_NORETURN ThrowClassCastException();
 // Throws arithmetic exception.
-void ThrowArithmeticException();
+void RUNTIME_NORETURN ThrowArithmeticException();
 // Throws number format exception.
-void ThrowNumberFormatException();
+void RUNTIME_NORETURN ThrowNumberFormatException();
 // Throws out of memory error.
-void ThrowOutOfMemoryError();
+void RUNTIME_NORETURN ThrowOutOfMemoryError();
 // Throws not implemented error.
-void ThrowNotImplementedError();
+void RUNTIME_NORETURN ThrowNotImplementedError();
 // Throws illegal character conversion exception (used in UTF8/UTF16 conversions).
-void ThrowIllegalCharacterConversionException();
+void RUNTIME_NORETURN ThrowIllegalCharacterConversionException();
 // Prints out mesage of Throwable.
 void PrintThrowable(KRef);
 

runtime/src/main/cpp/Math.cpp

@@ -248,9 +248,13 @@ KLong Kotlin_math_absl(KLong x) { return llabs(x); }
 
 // kotlin.math isn't supported for WASM.
 namespace {
-    void NotImplemented() { ThrowNotImplementedError(); }
+
+RUNTIME_NORETURN void NotImplemented() {
+  ThrowNotImplementedError();
 }
 
+}  // namespace
+
 KDouble Kotlin_math_sin(KDouble x) { NotImplemented(); }
 KDouble Kotlin_math_cos(KDouble x) { NotImplemented(); }
 KDouble Kotlin_math_tan(KDouble x) { NotImplemented(); }

runtime/src/main/cpp/Memory.cpp

@@ -1312,7 +1312,6 @@ bool ClearSubgraphReferences(ObjHeader* root, bool checked) {
 #if USE_GC
   if (root != nullptr) {
     auto state = memoryState;
-
     auto container = root->container();
 
     ContainerHeaderSet visited;

runtime/src/main/cpp/Worker.cpp

@@ -460,6 +460,19 @@ KInt versionToken() {
   return theState()->versionToken();
 }
 
+OBJ_GETTER(attachObjectGraphInternal, KNativePtr stable) {
+  RETURN_RESULT_OF(AdoptStablePointer, stable);
+}
+
+KNativePtr detachObjectGraphInternal(KInt transferMode, KRef producer) {
+   KRef ref = nullptr;
+   WorkerLaunchpad(producer, &ref);
+   if (ref != nullptr) {
+     return transfer(ref, transferMode);
+   } else
+     return nullptr;
+}
+
 #else
 
 KInt startWorker() {
@@ -502,6 +515,16 @@ KInt versionToken() {
   return 0;
 }
 
+OBJ_GETTER(attachObjectGraphInternal, KNativePtr stable) {
+  ThrowWorkerUnsupported();
+  return nullptr;
+}
+
+KNativePtr detachObjectGraphInternal(KInt transferMode, KRef producer) {
+   ThrowWorkerUnsupported();
+   return nullptr;
+}
+
 #endif  // WITH_WORKERS
 
 }  // namespace
@@ -540,5 +563,12 @@ KInt Kotlin_Worker_versionToken() {
   return versionToken();
 }
 
+OBJ_GETTER(Kotlin_Worker_attachObjectGraphInternal, KNativePtr stable) {
+  RETURN_RESULT_OF(attachObjectGraphInternal, stable);
+}
+
+KNativePtr Kotlin_Worker_detachObjectGraphInternal(KInt transferMode, KRef producer) {
+  return detachObjectGraphInternal(transferMode, producer);
+}
 
 }  // extern "C"

runtime/src/main/kotlin/konan/worker/Future.kt

@@ -0,0 +1,111 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package konan.worker
+
+import konan.SymbolName
+import konan.internal.ExportForCppRuntime
+
+/**
+ * Unique identifier of the future. Futures can be used from other workers.
+ */
+typealias FutureId = Int
+
+/**
+ * State of the future object.
+ */
+enum class FutureState(val value: Int) {
+    INVALID(0),
+    // Future is scheduled for execution.
+    SCHEDULED(1),
+    // Future result is computed.
+    COMPUTED(2),
+    // Future is cancelled.
+    CANCELLED(3)
+}
+
+/**
+ * Class representing abstract computation, whose result may become available in the future.
+ */
+// TODO: make me value class!
+class Future<T> internal constructor(val id: FutureId) {
+    /**
+     * Blocks execution until the future is ready.
+     */
+    inline fun <R> consume(code: (T) -> R) =
+            when (state) {
+                FutureState.SCHEDULED, FutureState.COMPUTED -> {
+                    val value = @Suppress("UNCHECKED_CAST", "NON_PUBLIC_CALL_FROM_PUBLIC_INLINE") (consumeFuture(id) as T)
+                    code(value)
+                }
+                FutureState.INVALID ->
+                    throw IllegalStateException("Future is in an invalid state: $state")
+                FutureState.CANCELLED ->
+                    throw IllegalStateException("Future is cancelled")
+            }
+
+    fun result(): T  = consume { it -> it }
+
+    val state: FutureState
+        get() = FutureState.values()[stateOfFuture(id)]
+
+    override fun equals(other: Any?) = (other is Future<*>) && (id == other.id)
+
+    override fun hashCode() = id
+}
+
+/**
+ * Wait for availability of futures in the collection. Returns set with all futures which have
+ * value available for the consumption.
+ */
+fun <T> Collection<Future<T>>.waitForMultipleFutures(millis: Int): Set<Future<T>> {
+    val result = mutableSetOf<Future<T>>()
+
+    while (true) {
+        val versionToken = versionToken()
+        for (future in this) {
+            if (future.state == FutureState.COMPUTED) {
+                result += future
+            }
+        }
+        if (result.isNotEmpty()) return result
+
+        if (waitForAnyFuture(versionToken, millis)) break
+    }
+
+    for (future in this) {
+        if (future.state == FutureState.COMPUTED) {
+            result += future
+        }
+    }
+
+    return result
+}
+
+// Private APIs.
+@SymbolName("Kotlin_Worker_stateOfFuture")
+external internal fun stateOfFuture(id: FutureId): Int
+
+@SymbolName("Kotlin_Worker_consumeFuture")
+@kotlin.internal.InlineExposed
+external internal fun consumeFuture(id: FutureId): Any?
+
+@SymbolName("Kotlin_Worker_waitForAnyFuture")
+external internal fun waitForAnyFuture(versionToken: Int, millis: Int): Boolean
+
+@SymbolName("Kotlin_Worker_versionToken")
+external internal fun versionToken(): Int
+

runtime/src/main/kotlin/konan/worker/ObjectTransfer.kt

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package konan.worker
+
+import kotlinx.cinterop.*
+
+/**
+ *  Object Transfer Basics.
+ *
+ *  Objects can be passed between threads in one of two possible modes.
+ *
+ *    - CHECKED - object subgraph is checked to be not reachable by other globals or locals, and passed
+ *      if so, otherwise an exception is thrown
+ *    - UNCHECKED - object is blindly passed to another worker, if there are references
+ *      left in the passing worker - it may lead to crash or program malfunction
+ *
+ *   Checked mode checks if object is no longer used in passing worker, using memory-management
+ *  specific algorithm (ARC implementation relies on trial deletion on object graph rooted in
+ *  passed object), and throws IllegalStateException if object graph rooted in transferred object
+ *  is reachable by some other means,
+ *
+ *   Unchecked mode, intended for most performance crititcal operations, where object graph ownership
+ *  is expected to be correct (such as application debugged earlier in CHECKED mode), just transfers
+ *  ownership without further checks.
+ *
+ *   Note, that for some cases cycle collection need to be done to ensure that dead cycles do not affect
+ *  reachability of passed object graph. See `konan.internal.GC.collect()`.
+ *
+ */
+enum class TransferMode(val value: Int) {
+    CHECKED(0),
+    UNCHECKED(1) // USE UNCHECKED MODE ONLY IF ABSOLUTELY SURE WHAT YOU'RE DOING!!!
+}
+
+/**
+ * Creates verbatim *shallow* copy of passed object, use carefully to create disjoint object graph.
+ */
+inline fun <reified T> T.shallowCopy(): T = shallowCopyInternal(this) as T
+
+/**
+ * Creates verbatim *deep* copy of passed object's graph, use *VERY* carefully to create disjoint object graph.
+ * Note that this function could potentially duplicate a lot of objects.
+ */
+inline fun <reified T> T.deepCopy(): T = TODO()
+
+/**
+ * Creates stable pointer to object, ensuring associated object subgraph is disjoint in specified mode
+ * ([TransferMode.CHECKED] by default).
+ * It could be stored to C variable or passed to another thread, where it could be retrieved with [attachObjectGraph].
+ */
+inline fun <reified T> detachObjectGraph(mode: TransferMode = TransferMode.CHECKED, noinline producer: () -> T): COpaquePointer? =
+        detachObjectGraphInternal(mode.value, producer as () -> Any?)
+
+/**
+ * Attaches previously detached with [detachObjectGraph] object subgraph.
+ * Please note, that once object graph is attached, the stable pointer does not have sense anymore,
+ * and shall be discarded.
+ */
+inline fun <reified T> attachObjectGraph(stable: COpaquePointer?): T =
+        attachObjectGraphInternal(stable) as T
+
+// Private APIs.
+@PublishedApi
+@SymbolName("Kotlin_Worker_shallowCopyInternal")
+external internal fun shallowCopyInternal(value: Any?): Any?
+@PublishedApi
+@SymbolName("Kotlin_Worker_detachObjectGraphInternal")
+external internal fun detachObjectGraphInternal(mode: Int, producer: () -> Any?): COpaquePointer?
+@PublishedApi
+@SymbolName("Kotlin_Worker_attachObjectGraphInternal")
+external internal fun attachObjectGraphInternal(stable: COpaquePointer?): Any?
+

runtime/src/main/kotlin/konan/worker/Worker.kt

@@ -31,85 +31,11 @@ import kotlinx.cinterop.*
  * workers as needed.
  */
 
-/**
- * State of the future object.
- */
-enum class FutureState(val value: Int) {
-    INVALID(0),
-    // Future is scheduled for execution.
-    SCHEDULED(1),
-    // Future result is computed.
-    COMPUTED(2),
-    // Future is cancelled.
-    CANCELLED(3)
-}
-
-/**
- *  Object Transfer Basics.
- *
- *  Objects can be passed between threads in one of two possible modes.
- *
- *    - CHECKED - object subgraph is checked to be not reachable by other globals or locals, and passed
- *      if so, otherwise an exception is thrown
- *    - UNCHECKED - object is blindly passed to another worker, if there are references
- *      left in the passing worker - it may lead to crash or program malfunction
- *
- *   Checked mode checks if object is no longer used in passing worker, using memory-management
- *  specific algorithm (ARC implementation relies on trial deletion on object graph rooted in
- *  passed object), and throws IllegalStateException if object graph rooted in transferred object
- *  is reachable by some other means,
- *
- *   Unchecked mode, intended for most performance crititcal operations, where object graph ownership
- *  is expected to be correct (such as application debugged earlier in CHECKED mode), just transfers
- *  ownership without further checks.
- *
- *   Note, that for some cases cycle collection need to be done to ensure that dead cycles do not affect
- *  reachability of passed object graph. See `konan.internal.GC.collect()`.
- *
- */
-enum class TransferMode(val value: Int) {
-    CHECKED(0),
-    UNCHECKED(1) // USE UNCHECKED MODE ONLY IF ABSOLUTELY SURE WHAT YOU'RE DOING!!!
-}
-
 /**
  * Unique identifier of the worker. Workers can be used from other workers.
  */
 typealias WorkerId = Int
-/**
- * Unique identifier of the future. Futures can be used from other workers.
- */
-typealias FutureId = Int
 
-/**
- * Class representing abstract computation, whose result may become available in the future.
- */
-// TODO: make me value class!
-class Future<T> internal constructor(val id: FutureId) {
-    /**
-     * Blocks execution until the future is ready.
-     */
-    inline fun <R> consume(code: (T) -> R) =
-        when (state) {
-            FutureState.SCHEDULED, FutureState.COMPUTED -> {
-                val value = @Suppress("UNCHECKED_CAST", "NON_PUBLIC_CALL_FROM_PUBLIC_INLINE") (consumeFuture(id) as T)
-                code(value)
-            }
-            FutureState.INVALID ->
-                throw IllegalStateException("Future is in an invalid state: $state")
-            FutureState.CANCELLED ->
-                throw IllegalStateException("Future is cancelled")
-        }
-
-    fun result(): T  = consume { it -> it }
-
-    val state: FutureState
-        get() = FutureState.values()[stateOfFuture(id)]
-
-    override fun equals(other: Any?) = (other is Future<*>) && (id == other.id)
-
-    override fun hashCode() = id
-}
 
 /**
  * Class representing worker.
@@ -154,46 +80,7 @@ class Worker(val id: WorkerId) {
  */
 fun startWorker(): Worker = Worker(startInternal())
 
-/**
+// Private APIs.
- * Wait for availability of futures in the collection. Returns set with all futures which have
- * value available for the consumption.
- */
-fun <T> Collection<Future<T>>.waitForMultipleFutures(millis: Int): Set<Future<T>> {
-    val result = mutableSetOf<Future<T>>()
-
-    while (true) {
-        val versionToken = versionToken()
-        for (future in this) {
-            if (future.state == FutureState.COMPUTED) {
-                result += future
-            }
-        }
-        if (result.isNotEmpty()) return result
-
-        if (waitForAnyFuture(versionToken, millis)) break
-    }
-
-    for (future in this) {
-        if (future.state == FutureState.COMPUTED) {
-            result += future
-        }
-    }
-
-    return result
-}
-
-/**
- * Creates verbatim *shallow* copy of passed object, use carefully to create disjoint object graph.
- */
-fun <T> T.shallowCopy(): T = @Suppress("UNCHECKED_CAST") (shallowCopyInternal(this) as T)
-
-/**
- * Creates verbatim *deep* copy of passed object's graph, use *VERY* carefully to create disjoint object graph.
- * Note that this function could potentially duplicate a lot of objects.
- */
-fun <T> T.deepCopy(): T = TODO()
-
-// Implementation details.
 @konan.internal.ExportForCompiler
 internal fun scheduleImpl(worker: Worker, mode: TransferMode, producer: () -> Any?,
                           job: CPointer<CFunction<*>>): Future<Any?> =
@@ -209,22 +96,6 @@ external internal fun requestTerminationInternal(id: WorkerId, processScheduledJ
 external internal fun scheduleInternal(
         id: WorkerId, mode: Int, producer: () -> Any?, job: CPointer<CFunction<*>>): FutureId
 
-@SymbolName("Kotlin_Worker_shallowCopyInternal")
-external internal fun shallowCopyInternal(value: Any?): Any?
-
-@SymbolName("Kotlin_Worker_stateOfFuture")
-external internal fun stateOfFuture(id: FutureId): Int
-
-@SymbolName("Kotlin_Worker_consumeFuture")
-@kotlin.internal.InlineExposed
-external internal fun consumeFuture(id: FutureId): Any?
-
-@SymbolName("Kotlin_Worker_waitForAnyFuture")
-external internal fun waitForAnyFuture(versionToken: Int, millis: Int): Boolean
-
-@SymbolName("Kotlin_Worker_versionToken")
-external internal fun versionToken(): Int
-
 @ExportForCppRuntime
 internal fun ThrowWorkerUnsupported(): Unit =
         throw UnsupportedOperationException("Workers are not supported")

samples/globalState/src/main/c_interop/global.def

@@ -5,6 +5,7 @@ typedef struct {
   int x;
   float f;
   char* string;
+  void* kotlinObject;
 } SharedData;
 
 SharedData sharedData;

samples/globalState/src/main/kotlin/Global.kt

@@ -25,8 +25,15 @@ inline fun Int.ensureUnixCallResult(op: String, predicate: (Int) -> Boolean = {
     return this
 }
 
-fun dumpShared(prefix: String) =
+data class SharedDataMember(val double: Double)
-        println("$prefix: ${pthread_self().rawValue} x=${sharedData.x} f=${sharedData.f} s=${sharedData.string!!.toKString()}")
+
+data class SharedData(val string: String, val int: Int, val member: SharedDataMember)
+
+fun dumpShared(prefix: String): Unit {
+    println("""
+            $prefix: ${pthread_self().rawValue} x=${sharedData.x} f=${sharedData.f} s=${sharedData.string!!.toKString()}
+            """.trimIndent())
+}
 
 fun main(args: Array<String>) {
     // Arena owning all native allocs.
@@ -36,6 +43,9 @@ fun main(args: Array<String>) {
     sharedData.x = 239
     sharedData.f = 0.5f
     sharedData.string = "Hello Kotlin!".cstr.getPointer(arena)
+    sharedData.kotlinObject = konan.worker.detachObjectGraph {
+        SharedData("A string", 42, SharedDataMember(2.39))
+    }
     dumpShared("thread1")
 
     // Start a new thread, that sees the variable.
@@ -43,12 +53,15 @@ fun main(args: Array<String>) {
     memScoped {
         val thread = alloc<pthread_tVar>()
         pthread_create(thread.ptr, null, staticCFunction {
-            _ ->
+            argC ->
             initRuntimeIfNeeded()
             dumpShared("thread2")
+            val kotlinObject = konan.worker.attachObjectGraph<SharedData>(sharedData.kotlinObject)
+            val arg = konan.worker.attachObjectGraph<SharedDataMember>(argC)
+            println("thread arg is $arg Kotlin object is $kotlinObject")
             // Workaround for compiler issue.
             null as COpaquePointer?
-        }, null).ensureUnixCallResult("pthread_create")
+        }, konan.worker.detachObjectGraph { SharedDataMember(3.14)} ).ensureUnixCallResult("pthread_create")
         pthread_join(thread.value, null).ensureUnixCallResult("pthread_join")
     }