From d9263e0fc271540d55825d6de2d96215b4de5c0f Mon Sep 17 00:00:00 2001 From: Nikolay Igotti Date: Mon, 28 Jan 2019 15:38:53 +0300 Subject: [PATCH] Async continuation API. (#2580) --- samples/objc/src/objcMain/kotlin/Async.kt | 104 +++++++++++++++++++++ samples/objc/src/objcMain/kotlin/Window.kt | 59 +++--------- 2 files changed, 116 insertions(+), 47 deletions(-) create mode 100644 samples/objc/src/objcMain/kotlin/Async.kt diff --git a/samples/objc/src/objcMain/kotlin/Async.kt b/samples/objc/src/objcMain/kotlin/Async.kt new file mode 100644 index 00000000000..78517f043d7 --- /dev/null +++ b/samples/objc/src/objcMain/kotlin/Async.kt @@ -0,0 +1,104 @@ +package sample.objc + +import kotlinx.cinterop.staticCFunction +import platform.Foundation.NSOperationQueue +import platform.Foundation.NSThread +import platform.darwin.dispatch_async_f +import platform.darwin.dispatch_get_main_queue +import platform.darwin.dispatch_sync_f +import kotlin.native.concurrent.* +import kotlin.test.assertNotNull + +inline fun executeAsync(queue: NSOperationQueue, crossinline producerConsumer: () -> Pair Unit>) { + dispatch_async_f(queue.underlyingQueue, DetachedObjectGraph { + producerConsumer() + }.asCPointer(), staticCFunction { it -> + val result = DetachedObjectGraph Unit>>(it).attach() + result.second(result.first) + }) +} + +inline fun mainContinuation(singleShot: Boolean = true, noinline block: () -> Unit) = Continuation0( + block, staticCFunction { invokerArg -> + if (NSThread.isMainThread()) { + invokerArg!!.callContinuation0() + } else { + dispatch_sync_f(dispatch_get_main_queue(), invokerArg, staticCFunction { args -> + args!!.callContinuation0() + }) + } +}, singleShot) + +inline fun mainContinuation(singleShot: Boolean = true, noinline block: (T1) -> Unit) = Continuation1( + block, staticCFunction { invokerArg -> + if (NSThread.isMainThread()) { + invokerArg!!.callContinuation1() + } else { + dispatch_sync_f(dispatch_get_main_queue(), invokerArg, staticCFunction { args -> + args!!.callContinuation1() + }) + } +}, singleShot) + +inline fun mainContinuation(singleShot: Boolean = true, noinline block: (T1, T2) -> Unit) = Continuation2( + block, staticCFunction { invokerArg -> + if (NSThread.isMainThread()) { + invokerArg!!.callContinuation2() + } else { + dispatch_sync_f(dispatch_get_main_queue(), invokerArg, staticCFunction { args -> + args!!.callContinuation2() + }) + } +}, singleShot) + +// This object allows to create frozen Kotlin continuations suitable for execution on other threads/queues. +// It takes frozen operation and any after call, and creates lambda which could be used to run operation +// anywhere and provide result to `after` callback. +@ThreadLocal +object Continuator { + val map = mutableMapOf>() + + fun wrap(operation: () -> Unit, after: () -> Unit): () -> Unit { + assert(NSThread.isMainThread()) + assert(operation.isFrozen) + val id = Any().freeze() + map[id] = Pair(0, after) + return { + initRuntimeIfNeeded() + operation() + executeAsync(NSOperationQueue.mainQueue) { + Pair(id, { id: Any -> Continuator.execute(id) }) + } + }.freeze() + } + + fun

wrap(operation: () -> P, block: (P) -> Unit): () -> Unit { + assert(NSThread.isMainThread()) + assert(operation.isFrozen) + val id = Any().freeze() + map[id] = Pair(1, block) + return { + initRuntimeIfNeeded() + // Note, that operation here must return detachable value (for example, frozen). + executeAsync(NSOperationQueue.mainQueue) { + Pair(Pair(id, operation()), { it: Pair -> + Continuator.execute(it.first, it.second) + }) + } + }.freeze() + } + + fun execute(id: Any) { + val countAndBlock = map.remove(id) + assertNotNull(countAndBlock) + assert(countAndBlock.first == 0) + (countAndBlock.second as Function0)() + } + + fun

execute(id: Any, parameter: P) { + val countAndBlock = map.remove(id) + assertNotNull(countAndBlock) + assert(countAndBlock.first == 1) + (countAndBlock.second as Function1)(parameter) + } +} diff --git a/samples/objc/src/objcMain/kotlin/Window.kt b/samples/objc/src/objcMain/kotlin/Window.kt index 5f671d0523c..d2a217df6ed 100644 --- a/samples/objc/src/objcMain/kotlin/Window.kt +++ b/samples/objc/src/objcMain/kotlin/Window.kt @@ -10,55 +10,11 @@ import platform.AppKit.* import platform.Contacts.CNContactStore import platform.Contacts.CNEntityType import platform.Foundation.* -import platform.darwin.NSObject -import platform.darwin.dispatch_async_f -import platform.darwin.dispatch_get_main_queue -import platform.darwin.dispatch_sync_f +import platform.darwin.* +import platform.posix.QOS_CLASS_BACKGROUND import platform.posix.memcpy import kotlin.native.concurrent.* - -inline fun executeAsync(queue: NSOperationQueue, crossinline producerConsumer: () -> Pair Unit>) { - dispatch_async_f(queue.underlyingQueue, DetachedObjectGraph { - producerConsumer() - }.asCPointer(), staticCFunction { it -> - val result = DetachedObjectGraph Unit>>(it).attach() - result.second(result.first) - }) -} - -inline fun mainContinuation(singleShot: Boolean = true, noinline block: () -> Unit) = Continuation0( - block, staticCFunction { invokerArg -> - if (NSThread.isMainThread()) { - invokerArg!!.callContinuation0() - } else { - dispatch_sync_f(dispatch_get_main_queue(), invokerArg, staticCFunction { args -> - args!!.callContinuation0() - }) - } - }, singleShot) - -inline fun mainContinuation(singleShot: Boolean = true, noinline block: (T1) -> Unit) = Continuation1( - block, staticCFunction { invokerArg -> - if (NSThread.isMainThread()) { - invokerArg!!.callContinuation1() - } else { - dispatch_sync_f(dispatch_get_main_queue(), invokerArg, staticCFunction { args -> - args!!.callContinuation1() - }) - } -}, singleShot) - -inline fun mainContinuation(singleShot: Boolean = true, noinline block: (T1, T2) -> Unit) = Continuation2( - block, staticCFunction { invokerArg -> - if (NSThread.isMainThread()) { - invokerArg!!.callContinuation2() - } else { - dispatch_sync_f(dispatch_get_main_queue(), invokerArg, staticCFunction { args -> - args!!.callContinuation2() - }) - } -}, singleShot) - +import kotlin.test.assertNotNull data class QueryResult(val json: Map?, val error: String?) @@ -89,6 +45,7 @@ private fun runApp() { app.run() } + class Controller : NSObject() { private var index = 1 private val httpDelegate = HttpDelegate() @@ -99,6 +56,14 @@ class Controller : NSObject() { appDelegate.contentText.string = "Another load in progress..." return } + + // Here we call continuator service to ensure we can access mutable state from continuation. + + dispatch_async(dispatch_get_global_queue(QOS_CLASS_BACKGROUND.convert(), 0), + Continuator.wrap({ println("In queue ${dispatch_get_current_queue()}")}.freeze()) { + println("After in queue ${dispatch_get_current_queue()}: $index") + }) + appDelegate.canClick = false // Fetch URL in the background on the button click. httpDelegate.fetchUrl("https://jsonplaceholder.typicode.com/todos/${index++}")