From 1ad50bc33f575e38400d495b8c86d7b8c31fc567 Mon Sep 17 00:00:00 2001 From: Nikolay Igotti Date: Wed, 21 Jun 2017 11:26:09 +0300 Subject: [PATCH] Workers draft (#655) --- .../kotlin/kotlinx/cinterop/NativeTypes.kt | 46 +- .../kotlin/backend/konan/InteropUtils.kt | 7 + .../jetbrains/kotlin/backend/konan/ir/Ir.kt | 1 + .../konan/lower/CallableReferenceLowering.kt | 61 +- .../backend/konan/lower/InteropLowering.kt | 26 + backend.native/tests/build.gradle | 15 + .../tests/runtime/workers/worker0.kt | 13 + .../tests/runtime/workers/worker1.kt | 26 + .../tests/runtime/workers/worker2.kt | 35 ++ runtime/src/main/cpp/Interop.cpp | 15 +- runtime/src/main/cpp/Memory.cpp | 71 ++- runtime/src/main/cpp/Memory.h | 12 + runtime/src/main/cpp/Types.h | 1 + runtime/src/main/cpp/Worker.cpp | 523 ++++++++++++++++++ runtime/src/main/kotlin/konan/Annotations.kt | 5 +- .../src/main/kotlin/konan/worker/Worker.kt | 243 ++++++++ 16 files changed, 1052 insertions(+), 48 deletions(-) create mode 100644 backend.native/tests/runtime/workers/worker0.kt create mode 100644 backend.native/tests/runtime/workers/worker1.kt create mode 100644 backend.native/tests/runtime/workers/worker2.kt create mode 100644 runtime/src/main/cpp/Worker.cpp create mode 100644 runtime/src/main/kotlin/konan/worker/Worker.kt diff --git a/Interop/Runtime/src/native/kotlin/kotlinx/cinterop/NativeTypes.kt b/Interop/Runtime/src/native/kotlin/kotlinx/cinterop/NativeTypes.kt index c31c5e13d1c..82c67d0334d 100644 --- a/Interop/Runtime/src/native/kotlin/kotlinx/cinterop/NativeTypes.kt +++ b/Interop/Runtime/src/native/kotlin/kotlinx/cinterop/NativeTypes.kt @@ -25,48 +25,48 @@ fun typeOf(): CVariable.Type = throw Error("typeOf() is called w internal fun CPointer<*>.cPointerToString() = "CPointer(raw=$rawValue)" -@Intrinsic external fun staticCFunction(function: () -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: () -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3, P4) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3, P4) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3, P4, P5) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3, P4, P5) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3, P4, P5, P6) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3, P4, P5, P6) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3, P4, P5, P6, P7) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3, P4, P5, P6, P7) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3, P4, P5, P6, P7, P8) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3, P4, P5, P6, P7, P8) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3, P4, P5, P6, P7, P8, P9) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3, P4, P5, P6, P7, P8, P9) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13, P14) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13, P14) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13, P14, P15) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13, P14, P15) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13, P14, P15, P16) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13, P14, P15, P16) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13, P14, P15, P16, P17) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13, P14, P15, P16, P17) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13, P14, P15, P16, P17, P18) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13, P14, P15, P16, P17, P18) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13, P14, P15, P16, P17, P18, P19) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13, P14, P15, P16, P17, P18, P19) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13, P14, P15, P16, P17, P18, P19, P20) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13, P14, P15, P16, P17, P18, P19, P20) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13, P14, P15, P16, P17, P18, P19, P20, P21) -> R): CPointer R>> +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13, P14, P15, P16, P17, P18, P19, P20, P21) -> R): CPointer R>> -@Intrinsic external fun staticCFunction(function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13, P14, P15, P16, P17, P18, P19, P20, P21, P22) -> R): CPointer R>> \ No newline at end of file +@Intrinsic external fun staticCFunction(@VolatileLambda function: (P1, P2, P3, P4, P5, P6, P7, P8, P9, P10, P11, P12, P13, P14, P15, P16, P17, P18, P19, P20, P21, P22) -> R): CPointer R>> \ No newline at end of file diff --git a/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/InteropUtils.kt b/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/InteropUtils.kt index 59d928b5d1f..33c3bbfa69e 100644 --- a/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/InteropUtils.kt +++ b/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/InteropUtils.kt @@ -90,6 +90,13 @@ internal class InteropBuiltIns(builtIns: KonanBuiltIns) { val staticCFunction = packageScope.getContributedFunctions("staticCFunction").toSet() + val workerPackageScope = builtIns.builtInsModule.getPackage(FqName("konan.worker")).memberScope + + val scheduleFunction = (workerPackageScope.getContributedClassifier("Worker") as ClassDescriptor). + unsubstitutedMemberScope.getContributedFunctions("schedule").single() + + val scheduleImplFunction = workerPackageScope.getContributedFunctions("scheduleImpl").single() + val signExtend = packageScope.getContributedFunctions("signExtend").single() val narrow = packageScope.getContributedFunctions("narrow").single() diff --git a/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/ir/Ir.kt b/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/ir/Ir.kt index 426ddd9be61..e028f5aed70 100644 --- a/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/ir/Ir.kt +++ b/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/ir/Ir.kt @@ -64,4 +64,5 @@ internal class KonanSymbols(context: Context, val symbolTable: SymbolTable): Sym }.toMap() + val scheduleImpl = symbolTable.referenceSimpleFunction(context.interopBuiltIns.scheduleImplFunction) } \ No newline at end of file diff --git a/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/lower/CallableReferenceLowering.kt b/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/lower/CallableReferenceLowering.kt index 4903f9fb67e..37d366e62a6 100644 --- a/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/lower/CallableReferenceLowering.kt +++ b/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/lower/CallableReferenceLowering.kt @@ -26,6 +26,9 @@ import org.jetbrains.kotlin.backend.common.lower.* import org.jetbrains.kotlin.backend.konan.Context import org.jetbrains.kotlin.backend.konan.descriptors.synthesizedName import org.jetbrains.kotlin.backend.common.ir.createFakeOverrideDescriptor +import org.jetbrains.kotlin.backend.common.ir.ir2string +import org.jetbrains.kotlin.backend.common.pop +import org.jetbrains.kotlin.backend.common.push import org.jetbrains.kotlin.backend.konan.llvm.functionName import org.jetbrains.kotlin.descriptors.* import org.jetbrains.kotlin.descriptors.annotations.Annotations @@ -33,15 +36,14 @@ import org.jetbrains.kotlin.descriptors.impl.ClassConstructorDescriptorImpl import org.jetbrains.kotlin.descriptors.impl.ClassDescriptorImpl import org.jetbrains.kotlin.descriptors.impl.SimpleFunctionDescriptorImpl import org.jetbrains.kotlin.incremental.components.NoLookupLocation +import org.jetbrains.kotlin.ir.IrElement +import org.jetbrains.kotlin.ir.IrStatement import org.jetbrains.kotlin.ir.builders.* import org.jetbrains.kotlin.ir.declarations.* import org.jetbrains.kotlin.ir.declarations.impl.IrClassImpl import org.jetbrains.kotlin.ir.declarations.impl.IrConstructorImpl import org.jetbrains.kotlin.ir.declarations.impl.IrFunctionImpl -import org.jetbrains.kotlin.ir.expressions.IrCall -import org.jetbrains.kotlin.ir.expressions.IrConstKind -import org.jetbrains.kotlin.ir.expressions.IrExpression -import org.jetbrains.kotlin.ir.expressions.IrFunctionReference +import org.jetbrains.kotlin.ir.expressions.* import org.jetbrains.kotlin.ir.expressions.impl.IrConstImpl import org.jetbrains.kotlin.ir.expressions.impl.IrDelegatingConstructorCallImpl import org.jetbrains.kotlin.ir.expressions.impl.IrInstanceInitializerCallImpl @@ -52,6 +54,7 @@ import org.jetbrains.kotlin.ir.symbols.IrValueParameterSymbol import org.jetbrains.kotlin.ir.symbols.impl.IrConstructorSymbolImpl import org.jetbrains.kotlin.ir.symbols.impl.IrSimpleFunctionSymbolImpl import org.jetbrains.kotlin.ir.util.* +import org.jetbrains.kotlin.ir.visitors.IrElementTransformerVoid import org.jetbrains.kotlin.ir.visitors.transformChildrenVoid import org.jetbrains.kotlin.name.FqName import org.jetbrains.kotlin.name.Name @@ -68,16 +71,56 @@ internal class CallableReferenceLowering(val context: Context): FileLoweringPass override fun lower(irFile: IrFile) { irFile.transformChildrenVoid(object: IrElementTransformerVoidWithContext() { - override fun visitCall(expression: IrCall): IrExpression { - if (expression.descriptor.original in context.interopBuiltIns.staticCFunction) { - return expression - } - return super.visitCall(expression) + private val stack = mutableListOf() + + override fun visitElement(element: IrElement): IrElement { + stack.push(element) + val result = super.visitElement(element) + stack.pop() + return result + } + + override fun visitExpression(expression: IrExpression): IrExpression { + stack.push(expression) + val result = super.visitExpression(expression) + stack.pop() + return result + } + + override fun visitDeclaration(declaration: IrDeclaration): IrStatement { + stack.push(declaration) + val result = super.visitDeclaration(declaration) + stack.pop() + return result + } + + override fun visitSpreadElement(spread: IrSpreadElement): IrSpreadElement { + stack.push(spread) + val result = super.visitSpreadElement(spread) + stack.pop() + return result } override fun visitFunctionReference(expression: IrFunctionReference): IrExpression { expression.transformChildrenVoid(this) + for (i in stack.size - 1 downTo 0) { + val cur = stack[i] + if (cur is IrBlock) + continue + if (cur !is IrCall) + break + val argument = if (i < stack.size - 1) stack[i + 1] else expression + val descriptor = cur.descriptor + val argumentDescriptor = descriptor.valueParameters.singleOrNull { + cur.getValueArgument(it.index) == argument + } + if (argumentDescriptor != null && argumentDescriptor.annotations.findAnnotation(FqName("konan.VolatileLambda")) != null) { + return expression + } + break + } + if (!expression.type.isFunctionOrKFunctionType) { // Not a subject of this lowering. return expression diff --git a/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/lower/InteropLowering.kt b/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/lower/InteropLowering.kt index bc6a4040b5f..d433aaf802d 100644 --- a/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/lower/InteropLowering.kt +++ b/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/lower/InteropLowering.kt @@ -179,6 +179,32 @@ private class InteropTransformer(val context: Context, val irFile: IrFile) : IrB typeArguments = null) } + interop.scheduleFunction -> { + val irCallableReference = unwrapStaticFunctionArgument(expression.getValueArgument(2)!!) + + if (irCallableReference == null || irCallableReference.getArguments().isNotEmpty()) { + context.reportCompilationError( + "${descriptor.fqNameSafe} must take an unbound, non-capturing function or lambda", + irFile, expression + ) + } + + val targetSymbol = irCallableReference.symbol + val target = targetSymbol.descriptor + val jobPointer = IrFunctionReferenceImpl( + builder.startOffset, builder.endOffset, + interop.cPointer.defaultType, + targetSymbol, target, + typeArguments = null) + + builder.irCall(symbols.scheduleImpl).apply { + putValueArgument(0, expression.dispatchReceiver) + putValueArgument(1, expression.getValueArgument(0)) + putValueArgument(2, expression.getValueArgument(1)) + putValueArgument(3, jobPointer) + } + } + interop.signExtend, interop.narrow -> { val integerTypePredicates = arrayOf( diff --git a/backend.native/tests/build.gradle b/backend.native/tests/build.gradle index 53d1196c8f7..2aa86c2a8a1 100644 --- a/backend.native/tests/build.gradle +++ b/backend.native/tests/build.gradle @@ -425,6 +425,21 @@ task empty_substring(type: RunKonanTest) { source = "runtime/basic/empty_substring.kt" } +task worker0(type: RunKonanTest) { + goldValue = "Got Input processed\nOK\n" + source = "runtime/workers/worker0.kt" +} + +task worker1(type: RunKonanTest) { + goldValue = "OK\n" + source = "runtime/workers/worker1.kt" +} + +task worker2(type: RunKonanTest) { + goldValue = "OK\n" + source = "runtime/workers/worker2.kt" +} + task superFunCall(type: RunKonanTest) { goldValue = "\n\n" source = "codegen/basics/superFunCall.kt" diff --git a/backend.native/tests/runtime/workers/worker0.kt b/backend.native/tests/runtime/workers/worker0.kt new file mode 100644 index 00000000000..223135e3090 --- /dev/null +++ b/backend.native/tests/runtime/workers/worker0.kt @@ -0,0 +1,13 @@ +import konan.worker.* + +fun main(args: Array) { + val worker = startWorker() + val future = worker.schedule(TransferMode.CHECKED, { "Input".shallowCopy()}) { + input -> input + " processed" + } + future.consume { + result -> println("Got $result") + } + worker.requestTermination().consume { _ -> } + println("OK") +} \ No newline at end of file diff --git a/backend.native/tests/runtime/workers/worker1.kt b/backend.native/tests/runtime/workers/worker1.kt new file mode 100644 index 00000000000..9bcbe4562c3 --- /dev/null +++ b/backend.native/tests/runtime/workers/worker1.kt @@ -0,0 +1,26 @@ +import konan.worker.* + +fun main(args: Array) { + val COUNT = 5 + val workers = Array(COUNT, { _ -> startWorker()}) + + for (attempt in 1 .. 3) { + val futures = Array(workers.size, + { i -> workers[i].schedule(TransferMode.CHECKED, { "$attempt: Input $i".shallowCopy() }) + { input -> input + " processed" } + }) + futures.forEachIndexed { index, future -> + future.consume { + result -> + if ("$attempt: Input $index processed" != result) { + println("Got unexpected $result") + throw Error(result) + } + } + } + } + workers.forEach { + it.requestTermination().consume { _ -> } + } + println("OK") +} \ No newline at end of file diff --git a/backend.native/tests/runtime/workers/worker2.kt b/backend.native/tests/runtime/workers/worker2.kt new file mode 100644 index 00000000000..57cb7f7bbf8 --- /dev/null +++ b/backend.native/tests/runtime/workers/worker2.kt @@ -0,0 +1,35 @@ +import konan.worker.* + +data class WorkerArgument(val intParam: Int, val stringParam: String) +data class WorkerResult(val intResult: Int, val stringResult: String) + +fun main(args: Array) { + val COUNT = 5 + val workers = Array(COUNT, { _ -> startWorker()}) + + for (attempt in 1 .. 3) { + val futures = Array(workers.size, { workerIndex -> workers[workerIndex].schedule(TransferMode.CHECKED, { + WorkerArgument(workerIndex, "attempt $attempt") }) { input -> + var sum = 0 + for (i in 0..input.intParam * 1000) { + sum += i + } + WorkerResult(sum, input.stringParam + " result") + } + }) + val futureSet = futures.toSet() + var consumed = 0 + while (consumed < futureSet.size) { + val ready = futureSet.waitForMultipleFutures(10000) + ready.forEach { + it.consume { result -> + if (result.stringResult != "attempt $attempt result") throw Error("Unexpected $result") + consumed++ } + } + } + } + workers.forEach { + it.requestTermination().consume { _ -> } + } + println("OK") +} \ No newline at end of file diff --git a/runtime/src/main/cpp/Interop.cpp b/runtime/src/main/cpp/Interop.cpp index a7090ef4aa6..10a267b52ab 100644 --- a/runtime/src/main/cpp/Interop.cpp +++ b/runtime/src/main/cpp/Interop.cpp @@ -73,19 +73,16 @@ void Kotlin_Interop_callWithVarargs(void* codePtr, void* returnValuePtr, FfiType ffi_call(&cif, (void (*)())codePtr, returnValuePtr, arguments); } -void* Kotlin_Interop_createStablePointer(KRef any) { - ::AddRef(any->container()); - return reinterpret_cast(any); +KNativePtr Kotlin_Interop_createStablePointer(KRef any) { + return CreateStablePointer(any); } -void Kotlin_Interop_disposeStablePointer(void* pointer) { - KRef ref = reinterpret_cast(pointer); - ::Release(ref->container()); +void Kotlin_Interop_disposeStablePointer(KNativePtr pointer) { + DisposeStablePointer(pointer); } -OBJ_GETTER(Kotlin_Interop_derefStablePointer, void* pointer) { - KRef ref = reinterpret_cast(pointer); - RETURN_OBJ(ref); +OBJ_GETTER(Kotlin_Interop_derefStablePointer, KNativePtr pointer) { + RETURN_RESULT_OF(DerefStablePointer, pointer); } } diff --git a/runtime/src/main/cpp/Memory.cpp b/runtime/src/main/cpp/Memory.cpp index 7c2a5b91eb6..f3a36c8e0d2 100644 --- a/runtime/src/main/cpp/Memory.cpp +++ b/runtime/src/main/cpp/Memory.cpp @@ -224,17 +224,17 @@ ContainerHeaderList collectMutableReferred(ContainerHeader* header) { for (int index = 0; index < typeInfo->objOffsetsCount_; index++) { ObjHeader** location = reinterpret_cast( reinterpret_cast(obj + 1) + typeInfo->objOffsets_[index]); - ObjHeader* obj = *location; - if (obj != nullptr && !isPermanent(obj->container())) { - result.push_back(obj->container()); + ObjHeader* ref = *location; + if (ref != nullptr && !isPermanent(ref->container())) { + result.push_back(ref->container()); } } if (typeInfo == theArrayTypeInfo) { ArrayHeader* array = obj->array(); for (int index = 0; index < array->count_; index++) { - ObjHeader* obj = *ArrayAddressOfElementAt(array, index); - if (obj != nullptr && !isPermanent(obj->container())) { - result.push_back(obj->container()); + ObjHeader* ref = *ArrayAddressOfElementAt(array, index); + if (ref != nullptr && !isPermanent(ref->container())) { + result.push_back(ref->container()); } } } @@ -258,6 +258,7 @@ void dumpWorker(const char* prefix, ContainerHeader* header, ContainerHeaderSet* void dumpReachable(const char* prefix, const ContainerHeaderSet* roots) { ContainerHeaderSet seen; for (auto container : *roots) { + fprintf(stderr, "%p is root\n", container); dumpWorker(prefix, container, &seen); } } @@ -590,6 +591,14 @@ void DeinitMemory(MemoryState* memoryState) { GarbageCollect(); delete memoryState->toFree; memoryState->toFree = nullptr; + +#if OPTIMIZE_GC + if (memoryState->toFreeCache != nullptr) { + freeMemory(memoryState->toFreeCache); + memoryState->toFreeCache = nullptr; + } +#endif + #endif // USE_GC if (memoryState->allocCount > 0) { @@ -857,5 +866,55 @@ KInt Kotlin_konan_internal_GC_getThreshold(KRef) { #endif } +KNativePtr CreateStablePointer(KRef any) { + if (any == nullptr) return nullptr; + ::AddRef(any->container()); + return reinterpret_cast(any); +} + +void DisposeStablePointer(KNativePtr pointer) { + if (pointer == nullptr) return; + KRef ref = reinterpret_cast(pointer); + ::Release(ref->container()); +} + +OBJ_GETTER(DerefStablePointer, KNativePtr pointer) { + KRef ref = reinterpret_cast(pointer); + RETURN_OBJ(ref); +} + +OBJ_GETTER(AdoptStablePointer, KNativePtr pointer) { + __sync_synchronize(); + KRef ref = reinterpret_cast(pointer); + // Somewhat hacky. + *OBJ_RESULT = ref; + return ref; +} + +bool ClearSubgraphReferences(ObjHeader* root, bool checked) { +#if USE_GC + if (root != nullptr) { + auto state = memoryState; + auto container = root->container(); + ContainerHeaderList todo; + ContainerHeaderSet subgraph; + todo.push_back(container); + while (todo.size() > 0) { + auto header = todo.back(); + todo.pop_back(); + if (subgraph.count(header) != 0) + continue; + subgraph.insert(header); + removeFreeable(state, header); + auto children = collectMutableReferred(header); + for (auto child : children) { + todo.push_back(child); + } + } + } +#endif // USE_GC + // TODO: perform trial deletion starting from this root, if in checked mode. + return true; +} } // extern "C" diff --git a/runtime/src/main/cpp/Memory.h b/runtime/src/main/cpp/Memory.h index f10daeff402..445c1112bf2 100644 --- a/runtime/src/main/cpp/Memory.h +++ b/runtime/src/main/cpp/Memory.h @@ -353,6 +353,18 @@ void ReleaseRefs(ObjHeader** start, int count) RUNTIME_NOTHROW; void LeaveFrame(ObjHeader** start, int count) RUNTIME_NOTHROW; // Collect garbage, which cannot be found by reference counting (cycles). void GarbageCollect() RUNTIME_NOTHROW; +// Clears object subgraph references from memory subsystem, and optionally +// checks if subgraph referenced by given root is disjoint from the rest of +// object graph, i.e. no external references exists. +bool ClearSubgraphReferences(ObjHeader* root, bool checked) RUNTIME_NOTHROW; +// Creates stable pointer out of the object. +void* CreateStablePointer(ObjHeader* obj) RUNTIME_NOTHROW; +// Disposes stable pointer to the object. +void DisposeStablePointer(void* pointer) RUNTIME_NOTHROW;; +// Translate stable pointer to object reference. +OBJ_GETTER(DerefStablePointer, void*) RUNTIME_NOTHROW; +// Move stable pointer ownership. +OBJ_GETTER(AdoptStablePointer, void*) RUNTIME_NOTHROW; #ifdef __cplusplus } diff --git a/runtime/src/main/cpp/Types.h b/runtime/src/main/cpp/Types.h index e7a314d43dc..f945f6c27b3 100644 --- a/runtime/src/main/cpp/Types.h +++ b/runtime/src/main/cpp/Types.h @@ -30,6 +30,7 @@ typedef int32_t KInt; typedef int64_t KLong; typedef float KFloat; typedef double KDouble; +typedef void* KNativePtr; typedef ObjHeader* KRef; typedef const ObjHeader* KConstRef; diff --git a/runtime/src/main/cpp/Worker.cpp b/runtime/src/main/cpp/Worker.cpp new file mode 100644 index 00000000000..406731c4166 --- /dev/null +++ b/runtime/src/main/cpp/Worker.cpp @@ -0,0 +1,523 @@ +/* + * Copyright 2010-2017 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. + */ +#define WITH_WORKERS 1 + +#include +#include +#include + +#if WITH_WORKERS +#include +#include + +#include +#include +#endif + +#include "Assert.h" +#include "Memory.h" +#include "Runtime.h" +#include "Types.h" + +extern "C" { + +void ThrowWorkerInvalidState(); +void ThrowWorkerUnsupported(); +OBJ_GETTER(WorkerLaunchpad, KRef); + +} // extern "C" + +namespace { + +#if WITH_WORKERS + +enum { + INVALID = 0, + SCHEDULED = 1, + COMPUTED = 2, + CANCELLED = 3 +}; + +enum { + CHECKED = 0, + UNCHECKED = 1 +}; + +KNativePtr transfer(KRef object, KInt mode) { + switch (mode) { + case CHECKED: + case UNCHECKED: + if (!ClearSubgraphReferences(object, mode == CHECKED)) { + ThrowWorkerInvalidState(); + return nullptr; + } + return object; + } + return nullptr; +} + +class Locker { + public: + explicit Locker(pthread_mutex_t* lock) : lock_(lock) { + pthread_mutex_lock(lock_); + } + ~Locker() { + pthread_mutex_unlock(lock_); + } + + private: + pthread_mutex_t* lock_; +}; + +class Future { + public: + Future(KInt id) : state_(SCHEDULED), id_(id) { + pthread_mutex_init(&lock_, nullptr); + pthread_cond_init(&cond_, nullptr); + } + + ~Future() { + pthread_mutex_destroy(&lock_); + pthread_cond_destroy(&cond_); + } + + OBJ_GETTER0(consumeResultUnlocked) { + Locker locker(&lock_); + while (state_ == SCHEDULED) { + pthread_cond_wait(&cond_, &lock_); + } + auto result = AdoptStablePointer(result_, OBJ_RESULT); + result_ = nullptr; + return result; + } + + void storeResultUnlocked(KNativePtr result); + + void cancelUnlocked(); + + // Those are called with the lock taken. + KInt state() const { return state_; } + KInt id() const { return id_; } + + private: + // State of future execution. + KInt state_; + // Integer id of the future. + KInt id_; + // Stable pointer with future's result. + KNativePtr result_; + // Lock and condition for waiting on the future. + pthread_mutex_t lock_; + pthread_cond_t cond_; +}; + +struct Job { + KRef (*function)(KRef, ObjHeader**); + KNativePtr argument; + Future* future; + KInt transferMode; +}; + +class Worker { + public: + Worker(KInt id) : id_(id) { + pthread_mutex_init(&lock_, nullptr); + pthread_cond_init(&cond_, nullptr); + } + + ~Worker() { + // Cleanup jobs in queue. + for (auto job : queue_) { + DisposeStablePointer(job.argument); + job.future->cancelUnlocked(); + } + + pthread_mutex_destroy(&lock_); + pthread_cond_destroy(&cond_); + } + + void putJob(Job job, bool toFront) { + Locker locker(&lock_); + if (toFront) + queue_.push_front(job); + else + queue_.push_back(job); + pthread_cond_signal(&cond_); + } + + Job getJob() { + Locker locker(&lock_); + while (queue_.size() == 0) { + pthread_cond_wait(&cond_, &lock_); + } + auto result = queue_.front(); + queue_.pop_front(); + return result; + } + + KInt id() const { return id_; } + + private: + KInt id_; + std::deque queue_; + // Lock and condition for waiting on the queue. + pthread_mutex_t lock_; + pthread_cond_t cond_; +}; + +class State { + public: + State() { + pthread_mutex_init(&lock_, nullptr); + pthread_cond_init(&cond_, nullptr); + + currentWorkerId_ = 1; + currentFutureId_ = 1; + currentVersion_ = 0; + } + + ~State() { + // TODO: some sanity check here? + pthread_mutex_destroy(&lock_); + pthread_cond_destroy(&cond_); + } + + Worker* addWorkerUnlocked() { + Locker locker(&lock_); + Worker* worker = new Worker(nextWorkerId()); + if (worker == nullptr) return nullptr; + workers_[worker->id()] = worker; + return worker; + } + + void removeWorkerUnlocked(KInt id) { + Locker locker(&lock_); + auto it = workers_.find(id); + if (it == workers_.end()) return; + workers_.erase(it); + } + + Future* addJobToWorkerUnlocked( + KInt id, KNativePtr jobFunction, KNativePtr jobArgument, bool toFront, KInt transferMode) { + Future* future = nullptr; + Worker* worker = nullptr; + { + Locker locker(&lock_); + + auto it = workers_.find(id); + if (it == workers_.end()) return nullptr; + worker = it->second; + + future = new Future(nextFutureId()); + futures_[future->id()] = future; + } + + Job job; + job.function = reinterpret_cast(jobFunction); + job.argument = jobArgument; + job.future = future; + job.transferMode = transferMode; + + worker->putJob(job, toFront); + + return future; + } + + KInt stateOfFutureUnlocked(KInt id) { + Locker locker(&lock_); + auto it = futures_.find(id); + if (it == futures_.end()) return INVALID; + return it->second->state(); + } + + OBJ_GETTER(consumeFutureUnlocked, KInt id) { + Future* future = nullptr; + { + Locker locker(&lock_); + auto it = futures_.find(id); + if (it == futures_.end()) ThrowWorkerInvalidState(); + future = it->second; + } + + KRef result = future->consumeResultUnlocked(OBJ_RESULT); + + { + Locker locker(&lock_); + auto it = futures_.find(id); + if (it != futures_.end()) { + futures_.erase(it); + delete future; + } + } + + return result; + } + + KBoolean waitForAnyFuture(KInt version, KInt millis) { + Locker locker(&lock_); + if (version != currentVersion_) return false; + + if (millis < 0) { + pthread_cond_wait(&cond_, &lock_); + return true; + } + struct timeval tv; + struct timespec ts; + gettimeofday(&tv, nullptr); + KLong nsDelta = millis * 1000000LL; + ts.tv_nsec = (tv.tv_usec * 1000LL + nsDelta) % 1000000000LL; + ts.tv_sec = (tv.tv_sec * 1000000000LL + nsDelta) / 1000000000LL; + pthread_cond_timedwait(&cond_, &lock_, &ts); + return true; + } + + void signalAnyFuture() { + { + Locker locker(&lock_); + currentVersion_++; + } + pthread_cond_broadcast(&cond_); + } + + KInt versionToken() { + Locker locker(&lock_); + return currentVersion_; + } + + // All those called with lock taken. + KInt nextWorkerId() { return currentWorkerId_++; } + KInt nextFutureId() { return currentFutureId_++; } + + private: + pthread_mutex_t lock_; + pthread_cond_t cond_; + std::unordered_map futures_; + std::unordered_map workers_; + KInt currentWorkerId_; + KInt currentFutureId_; + KInt currentVersion_; +}; + +State* theState() { + static State* state = nullptr; + + if (state != nullptr) { + return state; + } + + State* result = new State(); + + State* old = __sync_val_compare_and_swap(&state, nullptr, result); + if (old != nullptr) { + delete result; + // Someone else inited this data. + return old; + } + return state; +} + +void Future::storeResultUnlocked(KNativePtr result) { + { + Locker locker(&lock_); + state_ = COMPUTED; + result_ = result; + } + pthread_cond_signal(&cond_); + theState()->signalAnyFuture(); +} + +void Future::cancelUnlocked() { + { + Locker locker(&lock_); + state_ = CANCELLED; + result_ = nullptr; + } + pthread_cond_signal(&cond_); + theState()->signalAnyFuture(); +} + + +void* workerRoutine(void* argument) { + Worker* worker = reinterpret_cast(argument); + + RuntimeState* state = InitRuntime(); + while (true) { + Job job = worker->getJob(); + if (job.function == nullptr) { + // Termination request, notify the future. + job.future->storeResultUnlocked(nullptr); + theState()->removeWorkerUnlocked(worker->id()); + break; + } + ObjHolder argumentHolder; + KRef argument = AdoptStablePointer(job.argument, argumentHolder.slot()); + // Note that this is a bit hacky, as we must not auto-release resultRef, + // 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); + // Notify the future. + job.future->storeResultUnlocked(result); + } + + DeinitRuntime(state); + + delete worker; + + return nullptr; +} + +KInt startWorker() { + Worker* worker = theState()->addWorkerUnlocked(); + if (worker == nullptr) return -1; + pthread_t thread = 0; + pthread_create(&thread, nullptr, workerRoutine, worker); + return worker->id(); +} + +KInt schedule(KInt id, KInt transferMode, KRef producer, KNativePtr jobFunction) { + Job job; + // Note that this is a bit hacky, as we must not auto-release jobArgumentRef, + // so we don't use ObjHolder. + KRef jobArgumentRef = nullptr; + WorkerLaunchpad(producer, &jobArgumentRef); + KNativePtr jobArgument = transfer(jobArgumentRef, transferMode); + Future* future = theState()->addJobToWorkerUnlocked(id, jobFunction, jobArgument, false, transferMode); + if (future == nullptr) ThrowWorkerInvalidState(); + return future->id(); +} + +OBJ_GETTER(shallowCopy, KConstRef object) { + if (object == nullptr) RETURN_OBJ(nullptr); + + const TypeInfo* typeInfo = object->type_info(); + bool isArray = typeInfo->instanceSize_ < 0; + KRef result = isArray ? + AllocArrayInstance(typeInfo, object->array()->count_, OBJ_RESULT) : + AllocInstance(typeInfo, OBJ_RESULT); + // TODO: what to do when object references exist. + if (isArray) { + RuntimeAssert(object->array()->count_ == 0 || typeInfo != theArrayTypeInfo, "Object array copy unimplemented"); + memcpy(result->array() + 1, object->array() + 1, ArrayDataSizeBytes(object->array())); + } else { + RuntimeAssert(typeInfo->objOffsetsCount_ == 0, "Object reference copy unimplemented"); + memcpy(result + 1, object + 1, typeInfo->instanceSize_); + } + return result; +} + +KInt stateOfFuture(KInt id) { + return theState()->stateOfFutureUnlocked(id); +} + +OBJ_GETTER(consumeFuture, KInt id) { + RETURN_RESULT_OF(theState()->consumeFutureUnlocked, id); +} + +KInt requestTermination(KInt id, KBoolean processScheduledJobs) { + Future* future = theState()->addJobToWorkerUnlocked( + id, nullptr, nullptr, /* toFront = */ !processScheduledJobs, UNCHECKED); + if (future == nullptr) ThrowWorkerInvalidState(); + return future->id(); +} + +KBoolean waitForAnyFuture(KInt version, KInt millis) { + return theState()->waitForAnyFuture(version, millis); +} + +KInt versionToken() { + return theState()->versionToken(); +} + +#else + +KInt startWorker() { + ThrowWorkerUnsupported(); + return -1; +} + +OBJ_GETTER(shallowCopy, KConstRef object) { + ThrowWorkerUnsupported(); + RETURN_OBJ(nullptr); +} + +KInt stateOfFuture(KInt id) { + ThrowWorkerUnsupported(); + return 0; +} + +OBJ_GETTER(consumeFuture, KInt id) { + ThrowWorkerUnsupported(); + RETURN_OBJ(nullptr); +} + +KInt requestTermination(KInt id, KBoolean processScheduledJobs) { + ThrowWorkerUnsupported(); + return -1; +} + +KBoolean waitForAnyFuture(KInt versionToken, KInt millis) { + ThrowWorkerUnsupported(); + return false; +} + +KInt versionToken() { + ThrowWorkerUnsupported(); + return 0; +} + +#endif // WITH_WORKERS + +} // namespace + +extern "C" { + +KInt Kotlin_Worker_startInternal() { + return startWorker(); +} + +KInt Kotlin_Worker_requestTerminationWorkerInternal(KInt id, KBoolean processScheduledJobs) { + return requestTermination(id, processScheduledJobs); +} + +KInt Kotlin_Worker_scheduleInternal(KInt id, KInt transferMode, KRef producer, KNativePtr job) { + return schedule(id, transferMode, producer, job); +} + +OBJ_GETTER(Kotlin_Worker_shallowCopyInternal, KConstRef object) { + RETURN_RESULT_OF(shallowCopy, object); +} + +KInt Kotlin_Worker_stateOfFuture(KInt id) { + return stateOfFuture(id); +} + +OBJ_GETTER(Kotlin_Worker_consumeFuture, KInt id) { + RETURN_RESULT_OF(consumeFuture, id); +} + +KBoolean Kotlin_Worker_waitForAnyFuture(KInt versionToken, KInt millis) { + return waitForAnyFuture(versionToken, millis); +} + +KInt Kotlin_Worker_versionToken() { + return versionToken(); +} + + +} // extern "C" diff --git a/runtime/src/main/kotlin/konan/Annotations.kt b/runtime/src/main/kotlin/konan/Annotations.kt index 0c62455c14a..1bf097a986a 100644 --- a/runtime/src/main/kotlin/konan/Annotations.kt +++ b/runtime/src/main/kotlin/konan/Annotations.kt @@ -33,13 +33,16 @@ annotation class SymbolName(val name: String) //@Retention(AnnotationRetention.SOURCE) annotation class ExportTypeInfo(val name: String) +/** + * * If lambda shall be carefully lowered by the compiler. + */ +annotation class VolatileLambda /** * Preserve the function entry point during global optimizations */ public annotation class Used - /** * Need to be fixed because of reification support. */ diff --git a/runtime/src/main/kotlin/konan/worker/Worker.kt b/runtime/src/main/kotlin/konan/worker/Worker.kt new file mode 100644 index 00000000000..8e0a2f04691 --- /dev/null +++ b/runtime/src/main/kotlin/konan/worker/Worker.kt @@ -0,0 +1,243 @@ +/* + * Copyright 2010-2017 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 +import kotlinx.cinterop.* + +/** + * Workers: theory of operations. + * + * Worker represent asynchronous and concurrent computation, usually performed by other threads + * in the same process. Object passing between workers is performed using transfer operation, so that + * object graph belongs to one worker at the time, but can be disconnected and reconnected as needed. + * See 'Object Transfer Basics' below for more details on how objects shall be transferred. + * This approach ensures that no concurrent access happens to same object, while data may flow between + * 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 internal constructor(val id: FutureId) { + /** + * Blocks execution until the future is ready. + */ + fun consume(code: (T) -> Unit) { + when (state) { + FutureState.SCHEDULED, FutureState.COMPUTED -> { + val value = consumeFuture(id) as T + code(value) + } + FutureState.INVALID -> + throw IllegalStateException("Future is in an invalid state: $state") + 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 hashCode(): Int { + return id + } +} + +/** + * Class representing worker. + */ +// TODO: make me value class! +class Worker(val id: WorkerId) { + /** + * Requests termination of the worker. `processScheduledJobs` controls is we shall wait + * until all scheduled jobs processed, or terminate immediately. + */ + fun requestTermination(processScheduledJobs: Boolean = true) = + Future(requestTerminationInternal(id, processScheduledJobs)) + + /** + * Schedule a job for further execution in the worker. Schedule is a two-phase operation, + * first `producer` function is executed, and resulting object and whatever it refers to + * is analyzed for being an isolated object subgraph, if in checked mode. + * Afterwards, this disconnected object graph and `job` function pointer is being added to jobs queue + * of the selected worker. Note that `job` must not capture any state itself, so that whole state is + * explicitly stored in object produced by `producer`. Scheduled job is being executed by the worker, + * and result of such a execution is being disconnected from worker's object graph. Whoever will consume + * the future, can use result of worker's computations. + */ + fun schedule(mode: TransferMode, producer: () -> T1, + @VolatileLambda job: (T1) -> T2): Future = + /** + * This function is a magical operation, handled by lowering in the compiler, and replaced with call to + * scheduleImpl(worker, mode, producer, job) + * but first ensuring that `job` parameter doesn't capture any state. + */ + throw RuntimeException("Shall not be called directly") + + override fun equals(other: Any?): Boolean { + return (other is Worker) && (id == other.id) + } + + override fun hashCode(): Int { + return id + } +} + +/** + * Start new scheduling primitive, such as thread, to accept new tasks via `schedule` interface. + * Typically new worker may be needed for computations offload to another core, for IO it may be + * better to use non-blocking IO combined with more lightweight coroutines. + */ +fun startWorker() : Worker = Worker(startInternal()) + +/** + * Wait for availability of futures in the collection. Returns set with all futures which have + * value available for the consumption. + */ +fun Collection>.waitForMultipleFutures(millis: Int) : Set> { + val result = mutableSetOf>() + + 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.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. + */ +fun T.deepCopy(): T = TODO() + +// Implementation details. +@konan.internal.ExportForCompiler +internal fun scheduleImpl(worker: Worker, mode: TransferMode, producer: () -> Any?, + job: CPointer>) : Future = + Future(scheduleInternal(worker.id, mode.value, producer, job)) + +@SymbolName("Kotlin_Worker_startInternal") +external internal fun startInternal() : WorkerId + +@SymbolName("Kotlin_Worker_requestTerminationWorkerInternal") +external internal fun requestTerminationInternal(id: WorkerId, processScheduledJobs: Boolean): FutureId + +@SymbolName("Kotlin_Worker_scheduleInternal") +external internal fun scheduleInternal( + id: WorkerId, mode: Int, producer: () -> Any?, job: CPointer>) : 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") +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") + +@ExportForCppRuntime +internal fun ThrowWorkerInvalidState(): Unit = + throw IllegalStateException("Illegal transfer state") + +@ExportForCppRuntime +internal fun WorkerLaunchpad(function: () -> Any?) = function()