Fix bug preventing return from worker value passed in it. (#2135)
This commit is contained in:
@@ -663,6 +663,12 @@ task worker10(type: RunKonanTest) {
|
|||||||
source = "runtime/workers/worker10.kt"
|
source = "runtime/workers/worker10.kt"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
task worker11(type: RunKonanTest) {
|
||||||
|
disabled = (project.testTarget == 'wasm32') // Workers need pthreads.
|
||||||
|
goldValue = "OK\n"
|
||||||
|
source = "runtime/workers/worker11.kt"
|
||||||
|
}
|
||||||
|
|
||||||
task freeze0(type: RunKonanTest) {
|
task freeze0(type: RunKonanTest) {
|
||||||
disabled = (project.testTarget == 'wasm32') // No workers on WASM.
|
disabled = (project.testTarget == 'wasm32') // No workers on WASM.
|
||||||
goldValue = "frozen bit is true\n" +
|
goldValue = "frozen bit is true\n" +
|
||||||
|
|||||||
@@ -0,0 +1,48 @@
|
|||||||
|
/*
|
||||||
|
* Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
|
||||||
|
* that can be found in the LICENSE file.
|
||||||
|
*/
|
||||||
|
|
||||||
|
package runtime.workers.worker11
|
||||||
|
|
||||||
|
import kotlin.test.*
|
||||||
|
|
||||||
|
import kotlin.native.concurrent.*
|
||||||
|
import kotlinx.cinterop.convert
|
||||||
|
|
||||||
|
data class Job(val index: Int, var input: Int, var counter: Int)
|
||||||
|
|
||||||
|
fun initJobs(count: Int) = Array<Job?>(count) { i -> Job(i, i * 2, i)}
|
||||||
|
|
||||||
|
@Test fun runTest() {
|
||||||
|
val COUNT = 100
|
||||||
|
val workers = Array(COUNT, { _ -> Worker.start() })
|
||||||
|
val jobs = initJobs(COUNT)
|
||||||
|
val futures = Array(workers.size, { workerIndex ->
|
||||||
|
workers[workerIndex].execute(TransferMode.SAFE, {
|
||||||
|
val job = jobs[workerIndex]
|
||||||
|
jobs[workerIndex] = null
|
||||||
|
job!!
|
||||||
|
}) { job ->
|
||||||
|
job.counter += job.input
|
||||||
|
job
|
||||||
|
}
|
||||||
|
})
|
||||||
|
val futureSet = futures.toSet()
|
||||||
|
var consumed = 0
|
||||||
|
while (consumed < futureSet.size) {
|
||||||
|
val ready = futureSet.waitForMultipleFutures(10000)
|
||||||
|
ready.forEach {
|
||||||
|
it.consume { job ->
|
||||||
|
assertEquals(job.index * 3, job.counter)
|
||||||
|
jobs[job.index] = job
|
||||||
|
}
|
||||||
|
consumed++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
assertEquals(consumed, COUNT)
|
||||||
|
workers.forEach {
|
||||||
|
it.requestTermination().result
|
||||||
|
}
|
||||||
|
println("OK")
|
||||||
|
}
|
||||||
@@ -496,6 +496,7 @@ class ObjHolder {
|
|||||||
ObjHeader* obj() { return obj_; }
|
ObjHeader* obj() { return obj_; }
|
||||||
const ObjHeader* obj() const { return obj_; }
|
const ObjHeader* obj() const { return obj_; }
|
||||||
ObjHeader** slot() { return &obj_; }
|
ObjHeader** slot() { return &obj_; }
|
||||||
|
void clear() { ::UpdateRef(&obj_, nullptr); }
|
||||||
|
|
||||||
private:
|
private:
|
||||||
ObjHeader* obj_;
|
ObjHeader* obj_;
|
||||||
|
|||||||
@@ -379,6 +379,7 @@ void* workerRoutine(void* argument) {
|
|||||||
KNativePtr result = nullptr;
|
KNativePtr result = nullptr;
|
||||||
try {
|
try {
|
||||||
job.function(argument, &resultRef);
|
job.function(argument, &resultRef);
|
||||||
|
argumentHolder.clear();
|
||||||
// Transfer the result.
|
// Transfer the result.
|
||||||
result = transfer(resultRef, job.transferMode);
|
result = transfer(resultRef, job.transferMode);
|
||||||
} catch (ObjHolder& e) {
|
} catch (ObjHolder& e) {
|
||||||
|
|||||||
Reference in New Issue
Block a user