From 51bb86c4d727fd113d7d602d5d47d0ec6619598c Mon Sep 17 00:00:00 2001 From: Alexander Shabalin Date: Mon, 13 Apr 2020 15:45:03 +0300 Subject: [PATCH] Implement WorkerBoundReference (#4044) --- backend.native/tests/build.gradle | 10 + .../concurrent/worker_bound_reference0.kt | 455 ++++++++++++ .../concurrent/worker_bound_reference1.kt | 684 ++++++++++++++++++ runtime/src/main/cpp/Memory.cpp | 11 +- runtime/src/main/cpp/Types.h | 1 + runtime/src/main/cpp/WorkerBoundReference.cpp | 44 ++ runtime/src/main/cpp/WorkerBoundReference.h | 16 + .../native/concurrent/WorkerBoundReference.kt | 71 ++ 8 files changed, 1290 insertions(+), 2 deletions(-) create mode 100644 backend.native/tests/runtime/concurrent/worker_bound_reference0.kt create mode 100644 backend.native/tests/runtime/concurrent/worker_bound_reference1.kt create mode 100644 runtime/src/main/cpp/WorkerBoundReference.cpp create mode 100644 runtime/src/main/cpp/WorkerBoundReference.h create mode 100644 runtime/src/main/kotlin/kotlin/native/concurrent/WorkerBoundReference.kt diff --git a/backend.native/tests/build.gradle b/backend.native/tests/build.gradle index b1722d7e142..736e106241d 100644 --- a/backend.native/tests/build.gradle +++ b/backend.native/tests/build.gradle @@ -873,6 +873,16 @@ task empty_substring(type: KonanLocalTest) { source = "runtime/basic/empty_substring.kt" } +task worker_bound_reference0(type: KonanLocalTest) { + enabled = (project.testTarget != 'wasm32') // Workers need pthreads. + source = "runtime/concurrent/worker_bound_reference0.kt" +} + +task worker_bound_reference1(type: KonanLocalTest) { + enabled = (project.testTarget != 'wasm32') // Workers need pthreads. + source = "runtime/concurrent/worker_bound_reference1.kt" +} + task worker0(type: KonanLocalTest) { enabled = (project.testTarget != 'wasm32') // Workers need pthreads. goldValue = "Got Input processed\nOK\n" diff --git a/backend.native/tests/runtime/concurrent/worker_bound_reference0.kt b/backend.native/tests/runtime/concurrent/worker_bound_reference0.kt new file mode 100644 index 00000000000..a5cc3cc6908 --- /dev/null +++ b/backend.native/tests/runtime/concurrent/worker_bound_reference0.kt @@ -0,0 +1,455 @@ +/* + * Copyright 2010-2020 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.concurrent.worker_bound_reference0 + +import kotlin.test.* + +import kotlin.native.concurrent.* +import kotlin.native.internal.GC +import kotlin.native.ref.WeakReference + +class A(var a: Int) + +val global1: WorkerBoundReference = WorkerBoundReference(A(3)) + +@Test +fun testGlobal() { + assertEquals(3, global1.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, {}) { + global1 + } + + val value = future.result + assertEquals(3, value.value.a) + worker.requestTermination().result +} + +val global2: WorkerBoundReference = WorkerBoundReference(A(3)) + +@Test +fun testGlobalDenyAccessOnWorker() { + assertEquals(3, global2.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, {}) { + val local = global2 + assertFailsWith { + local.value + } + Unit + } + + future.result + worker.requestTermination().result +} + +val global3: WorkerBoundReference = WorkerBoundReference(A(3).freeze()) + +@Test +fun testGlobalAccessOnWorkerFrozenInitially() { + assertEquals(3, global3.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, {}) { + global3.value.a + } + + val value = future.result + assertEquals(3, value) + worker.requestTermination().result +} + +val global4: WorkerBoundReference = WorkerBoundReference(A(3)) + +@Test +fun testGlobalAccessOnWorkerFrozenBeforePassing() { + assertEquals(3, global4.value.a) + global4.value.freeze() + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, {}) { + global4.value.a + } + + val value = future.result + assertEquals(3, value) + worker.requestTermination().result +} + +val global5: WorkerBoundReference = WorkerBoundReference(A(3)) + +@Test +fun testGlobalAccessOnWorkerFrozenBeforeAccess() { + val semaphore: AtomicInt = AtomicInt(0) + + assertEquals(3, global5.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, { semaphore }) { semaphore -> + semaphore.increment() + while (semaphore.value < 2) { + } + + global5.value.a + } + + while (semaphore.value < 1) { + } + global5.value.freeze() + semaphore.increment() + + val value = future.result + assertEquals(3, value) + worker.requestTermination().result +} + +val global6: WorkerBoundReference = WorkerBoundReference(A(3)) + +@Test +fun testGlobalModification() { + val semaphore: AtomicInt = AtomicInt(0) + + assertEquals(3, global6.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, { semaphore }) { semaphore -> + semaphore.increment() + while (semaphore.value < 2) { + } + global6 + } + + while (semaphore.value < 1) { + } + global6.value.a = 4 + semaphore.increment() + + val value = future.result + assertEquals(4, value.value.a) + worker.requestTermination().result +} + +@Test +fun testLocal() { + val local = WorkerBoundReference(A(3)) + assertEquals(3, local.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, { local }) { local -> + local + } + + val value = future.result + assertEquals(3, value.value.a) + worker.requestTermination().result +} + +@Test +fun testLocalDenyAccessOnWorker() { + val local = WorkerBoundReference(A(3)) + assertEquals(3, local.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, { local }) { local -> + assertFailsWith { + local.value + } + Unit + } + + future.result + worker.requestTermination().result +} + +@Test +fun testLocalAccessOnWorkerFrozenInitially() { + val local = WorkerBoundReference(A(3).freeze()) + assertEquals(3, local.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, { local }) { local -> + local.value.a + } + + val value = future.result + assertEquals(3, value) + worker.requestTermination().result +} + +@Test +fun testLocalAccessOnWorkerFrozenBeforePassing() { + val local = WorkerBoundReference(A(3)) + assertEquals(3, local.value.a) + local.value.freeze() + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, { local }) { local -> + local.value.a + } + + val value = future.result + assertEquals(3, value) + worker.requestTermination().result +} + +@Test +fun testLocalAccessOnWorkerFrozenBeforeAccess() { + val semaphore: AtomicInt = AtomicInt(0) + + val local = WorkerBoundReference(A(3)) + assertEquals(3, local.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, { Pair(local, semaphore) }) { (local, semaphore) -> + semaphore.increment() + while (semaphore.value < 2) { + } + + local.value.a + } + + while (semaphore.value < 1) { + } + local.value.freeze() + semaphore.increment() + + val value = future.result + assertEquals(3, value) + worker.requestTermination().result +} + + +@Test +fun testLocalDenyAccessOnMainThread() { + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, {}) { + WorkerBoundReference(A(3)) + } + + val value = future.result + assertFailsWith { + value.value + } + + worker.requestTermination().result +} + +@Test +fun testLocalModification() { + val semaphore: AtomicInt = AtomicInt(0) + + val local = WorkerBoundReference(A(3)) + assertEquals(3, local.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, { Pair(local, semaphore) }) { (local, semaphore) -> + semaphore.increment() + while (semaphore.value < 2) { + } + local + } + + while (semaphore.value < 1) { + } + local.value.a = 4 + semaphore.increment() + + val value = future.result + assertEquals(4, value.value.a) + worker.requestTermination().result +} + +fun getOwnerAndWeaks(initial: Int): Triple?>, WeakReference>, WeakReference> { + val ref = WorkerBoundReference(A(initial)) + val refOwner: AtomicReference?> = AtomicReference(ref) + val refWeak = WeakReference(ref) + val refValueWeak = WeakReference(ref.value) + + return Triple(refOwner, refWeak, refValueWeak) +} + +@Test +fun testCollect() { + val (refOwner, refWeak, refValueWeak) = getOwnerAndWeaks(3) + + refOwner.value = null + GC.collect() + + // Last reference to WorkerBoundReference is gone, so it and it's referent are destroyed. + assertNull(refWeak.value) + assertNull(refValueWeak.value) +} + +fun collectInWorker(worker: Worker, semaphore: AtomicInt): Pair, Future> { + val (refOwner, _, refValueWeak) = getOwnerAndWeaks(3) + + val future = worker.execute(TransferMode.SAFE, { Pair(refOwner, semaphore) }) { (refOwner, semaphore) -> + semaphore.increment() + while (semaphore.value < 2) { + } + + refOwner.value = null + GC.collect() + } + + while (semaphore.value < 1) { + } + // At this point worker is spinning on semaphore. refOwner still contains reference to + // WorkerBoundReference, so referent is kept alive. + GC.collect() + assertNotNull(refValueWeak.value) + + return Pair(refValueWeak, future) +} + +@Test +fun testCollectInWorker() { + val semaphore: AtomicInt = AtomicInt(0) + + val worker = Worker.start() + + val (refValueWeak, future) = collectInWorker(worker, semaphore) + semaphore.increment() + future.result + + // At this point WorkerBoundReference no longer has a reference, so it's referent is destroyed. + // WorkerBoundReference, so referent is kept alive. + GC.collect() + assertNull(refValueWeak.value) + + worker.requestTermination().result +} + +fun doNotCollectInWorker(worker: Worker, semaphore: AtomicInt): Future> { + val ref = WorkerBoundReference(A(3)) + + return worker.execute(TransferMode.SAFE, { Pair(ref, semaphore) }) { (ref, semaphore) -> + semaphore.increment() + while (semaphore.value < 2) { + } + + GC.collect() + ref + } +} + +@Test +fun testDoNotCollectInWorker() { + val semaphore: AtomicInt = AtomicInt(0) + + val worker = Worker.start() + + val future = doNotCollectInWorker(worker, semaphore) + while (semaphore.value < 1) { + } + GC.collect() + semaphore.increment() + + val value = future.result + assertEquals(3, value.value.a) + worker.requestTermination().result +} + +class B1 { + lateinit var b2: WorkerBoundReference +} + +data class B2(val b1: WorkerBoundReference) + +fun createCyclicGarbage(): Triple?>, WeakReference, WeakReference> { + val ref1 = WorkerBoundReference(B1()) + val ref1Owner: AtomicReference?> = AtomicReference(ref1) + val ref1Weak = WeakReference(ref1.value) + + val ref2 = WorkerBoundReference(B2(ref1)) + val ref2Weak = WeakReference(ref2.value) + + ref1.value.b2 = ref2 + + return Triple(ref1Owner, ref1Weak, ref2Weak) +} + +@Test +fun doesNotCollectCyclicGarbage() { + val (ref1Owner, ref1Weak, ref2Weak) = createCyclicGarbage() + + ref1Owner.value = null + GC.collect() + + // If these asserts fail, that means WorkerBoundReference managed to clean up cyclic garbage all by itself. + assertNotNull(ref1Weak.value) + assertNotNull(ref2Weak.value) +} + +fun createCrossThreadCyclicGarbage( + worker: Worker +): Triple?>, WeakReference, WeakReference> { + val ref1 = WorkerBoundReference(B1()) + val ref1Owner: AtomicReference?> = AtomicReference(ref1) + val ref1Weak = WeakReference(ref1.value) + + val future = worker.execute(TransferMode.SAFE, { ref1 }) { ref1 -> + val ref2 = WorkerBoundReference(B2(ref1)) + Pair(ref2, WeakReference(ref2.value)) + } + val (ref2, ref2Weak) = future.result + + ref1.value.b2 = ref2 + + return Triple(ref1Owner, ref1Weak, ref2Weak) +} + +@Test +fun doesNotCollectCrossThreadCyclicGarbage() { + val worker = Worker.start() + + val (ref1Owner, ref1Weak, ref2Weak) = createCrossThreadCyclicGarbage(worker) + + ref1Owner.value = null + GC.collect() + worker.execute(TransferMode.SAFE, {}) { GC.collect() }.result + + // If these asserts fail, that means WorkerBoundReference managed to clean up cyclic garbage all by itself. + assertNotNull(ref1Weak.value) + assertNotNull(ref2Weak.value) + + worker.requestTermination().result +} + +@Test +fun concurrentAccess() { + val workerCount = 10 + val workerUnlocker = AtomicInt(0) + + val ref = WorkerBoundReference(A(3)) + assertEquals(3, ref.value.a) + + val workers = Array(workerCount) { + Worker.start() + } + val futures = Array(workers.size) { + workers[it].execute(TransferMode.SAFE, { Pair(ref, workerUnlocker) }) { (ref, workerUnlocker) -> + while (workerUnlocker.value < 1) { + } + + assertFailsWith { + ref.value + } + Unit + } + } + workerUnlocker.increment() + + for (future in futures) { + future.result + } + + for (worker in workers) { + worker.requestTermination().result + } +} diff --git a/backend.native/tests/runtime/concurrent/worker_bound_reference1.kt b/backend.native/tests/runtime/concurrent/worker_bound_reference1.kt new file mode 100644 index 00000000000..502cf06d4d1 --- /dev/null +++ b/backend.native/tests/runtime/concurrent/worker_bound_reference1.kt @@ -0,0 +1,684 @@ +/* + * Copyright 2010-2020 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.concurrent.worker_bound_reference1 + +import kotlin.test.* + +import kotlin.native.concurrent.* +import kotlin.native.internal.GC +import kotlin.native.ref.WeakReference + +class A(var a: Int) + +val global1: DisposableWorkerBoundReference = DisposableWorkerBoundReference(A(3)) + +@Test +fun testGlobal() { + assertEquals(3, global1.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, {}) { + global1 + } + + val value = future.result + assertEquals(3, value.value.a) + worker.requestTermination().result +} + +val global2: DisposableWorkerBoundReference = DisposableWorkerBoundReference(A(3)) + +@Test +fun testGlobalDenyAccessOnWorker() { + assertEquals(3, global2.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, {}) { + val local = global2 + assertFailsWith { + local.value + } + Unit + } + + future.result + worker.requestTermination().result +} + +val global3: DisposableWorkerBoundReference = DisposableWorkerBoundReference(A(3).freeze()) + +@Test +fun testGlobalAccessOnWorkerFrozenInitially() { + assertEquals(3, global3.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, {}) { + global3.value.a + } + + val value = future.result + assertEquals(3, value) + worker.requestTermination().result +} + +val global4: DisposableWorkerBoundReference = DisposableWorkerBoundReference(A(3)) + +@Test +fun testGlobalAccessOnWorkerFrozenBeforePassing() { + assertEquals(3, global4.value.a) + global4.value.freeze() + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, {}) { + global4.value.a + } + + val value = future.result + assertEquals(3, value) + worker.requestTermination().result +} + +val global5: DisposableWorkerBoundReference = DisposableWorkerBoundReference(A(3)) + +@Test +fun testGlobalAccessOnWorkerFrozenBeforeAccess() { + val semaphore: AtomicInt = AtomicInt(0) + + assertEquals(3, global5.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, { semaphore }) { semaphore -> + semaphore.increment() + while (semaphore.value < 2) { + } + + global5.value.a + } + + while (semaphore.value < 1) { + } + global5.value.freeze() + semaphore.increment() + + val value = future.result + assertEquals(3, value) + worker.requestTermination().result +} + +val global6: DisposableWorkerBoundReference = DisposableWorkerBoundReference(A(3)) + +@Test +fun testGlobalModification() { + val semaphore: AtomicInt = AtomicInt(0) + + assertEquals(3, global6.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, { semaphore }) { semaphore -> + semaphore.increment() + while (semaphore.value < 2) { + } + global6 + } + + while (semaphore.value < 1) { + } + global6.value.a = 4 + semaphore.increment() + + val value = future.result + assertEquals(4, value.value.a) + worker.requestTermination().result +} + +val global7: DisposableWorkerBoundReference = DisposableWorkerBoundReference(A(3)) + +@Test +fun testGlobalDispose() { + assertEquals(3, global7.value.a) + + global7.dispose() + global7.dispose() +} + +val global8: DisposableWorkerBoundReference = DisposableWorkerBoundReference(A(3)) + +@Test +fun testGlobalAccessAfterDispose() { + assertEquals(3, global8.value.a) + + global8.dispose() + assertFailsWith { + global8.value.a + } +} + +@Test +fun testLocal() { + val local = DisposableWorkerBoundReference(A(3)) + assertEquals(3, local.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, { local }) { local -> + local + } + + val value = future.result + assertEquals(3, value.value.a) + worker.requestTermination().result +} + +@Test +fun testLocalDenyAccessOnWorker() { + val local = DisposableWorkerBoundReference(A(3)) + assertEquals(3, local.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, { local }) { local -> + assertFailsWith { + local.value + } + Unit + } + + future.result + worker.requestTermination().result +} + +@Test +fun testLocalAccessOnWorkerFrozenInitially() { + val local = DisposableWorkerBoundReference(A(3).freeze()) + assertEquals(3, local.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, { local }) { local -> + local.value.a + } + + val value = future.result + assertEquals(3, value) + worker.requestTermination().result +} + +@Test +fun testLocalAccessOnWorkerFrozenBeforePassing() { + val local = DisposableWorkerBoundReference(A(3)) + assertEquals(3, local.value.a) + local.value.freeze() + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, { local }) { local -> + local.value.a + } + + val value = future.result + assertEquals(3, value) + worker.requestTermination().result +} + +@Test +fun testLocalAccessOnWorkerFrozenBeforeAccess() { + val semaphore: AtomicInt = AtomicInt(0) + + val local = DisposableWorkerBoundReference(A(3)) + assertEquals(3, local.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, { Pair(local, semaphore) }) { (local, semaphore) -> + semaphore.increment() + while (semaphore.value < 2) { + } + + local.value.a + } + + while (semaphore.value < 1) { + } + local.value.freeze() + semaphore.increment() + + val value = future.result + assertEquals(3, value) + worker.requestTermination().result +} + +@Test +fun testLocalDenyAccessOnMainThread() { + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, {}) { + DisposableWorkerBoundReference(A(3)) + } + + val value = future.result + assertFailsWith { + value.value + } + + worker.requestTermination().result +} + +@Test +fun testLocalModification() { + val semaphore: AtomicInt = AtomicInt(0) + + val local = DisposableWorkerBoundReference(A(3)) + assertEquals(3, local.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, { Pair(local, semaphore) }) { (local, semaphore) -> + semaphore.increment() + while (semaphore.value < 2) { + } + local + } + + while (semaphore.value < 1) { + } + local.value.a = 4 + semaphore.increment() + + val value = future.result + assertEquals(4, value.value.a) + worker.requestTermination().result +} + +@Test +fun testLocalDispose() { + val local = DisposableWorkerBoundReference(A(3)) + assertEquals(3, local.value.a) + + local.dispose() + local.dispose() +} + +@Test +fun testLocalAccessAfterDispose() { + val local = DisposableWorkerBoundReference(A(3)) + assertEquals(3, local.value.a) + + local.dispose() + assertFailsWith { + local.value.a + } +} + +fun getOwnerAndWeaks(initial: Int): Triple?>, WeakReference>, WeakReference> { + val ref = DisposableWorkerBoundReference(A(initial)) + val refOwner: AtomicReference?> = AtomicReference(ref) + val refWeak = WeakReference(ref) + val refValueWeak = WeakReference(ref.value) + + return Triple(refOwner, refWeak, refValueWeak) +} + +fun callDispose(ref: AtomicReference?>) { + ref.value!!.dispose() +} + +@Test +fun testCollect() { + val (refOwner, refWeak, refValueWeak) = getOwnerAndWeaks(3) + + refOwner.value = null + GC.collect() + + // Last reference to DisposableWorkerBoundReference is gone, so it and it's referent are destroyed. + assertNull(refWeak.value) + assertNull(refValueWeak.value) +} + +@Test +fun testDisposeAndCollect() { + val (refOwner, refWeak, refValueWeak) = getOwnerAndWeaks(3) + + callDispose(refOwner) + GC.collect() + + // refOwner still contains a reference to DisposableWorkerBoundReference. But it's referent is + // destroyed because of explicit dispose call. + assertNotNull(refWeak.value) + assertNull(refValueWeak.value) +} + +fun collectInWorker(worker: Worker, semaphore: AtomicInt): Pair, Future> { + val (refOwner, _, refValueWeak) = getOwnerAndWeaks(3) + + val future = worker.execute(TransferMode.SAFE, { Pair(refOwner, semaphore) }) { (refOwner, semaphore) -> + semaphore.increment() + while (semaphore.value < 2) { + } + + refOwner.value = null + GC.collect() + } + + while (semaphore.value < 1) { + } + // At this point worker is spinning on semaphore. refOwner still contains reference to + // DisposableWorkerBoundReference, so referent is kept alive. + GC.collect() + assertNotNull(refValueWeak.value) + + return Pair(refValueWeak, future) +} + +@Test +fun testCollectInWorker() { + val semaphore: AtomicInt = AtomicInt(0) + + val worker = Worker.start() + + val (refValueWeak, future) = collectInWorker(worker, semaphore) + semaphore.increment() + future.result + + // At this point DisposableWorkerBoundReference no longer has a reference, so it's referent is destroyed. + // DisposableWorkerBoundReference, so referent is kept alive. + GC.collect() + assertNull(refValueWeak.value) + + worker.requestTermination().result +} + +fun doNotCollectInWorker(worker: Worker, semaphore: AtomicInt): Future> { + val ref = DisposableWorkerBoundReference(A(3)) + + return worker.execute(TransferMode.SAFE, { Pair(ref, semaphore) }) { (ref, semaphore) -> + semaphore.increment() + while (semaphore.value < 2) { + } + + GC.collect() + ref + } +} + +@Test +fun testDoNotCollectInWorker() { + val semaphore: AtomicInt = AtomicInt(0) + + val worker = Worker.start() + + val future = doNotCollectInWorker(worker, semaphore) + while (semaphore.value < 1) { + } + GC.collect() + semaphore.increment() + + val value = future.result + assertEquals(3, value.value.a) + worker.requestTermination().result +} + +fun disposeInWorker(worker: Worker, semaphore: AtomicInt): Triple>, WeakReference, Future?>>> { + val (refOwner, refWeak, refValueWeak) = getOwnerAndWeaks(3) + + val future = worker.execute(TransferMode.SAFE, { Pair(refOwner, semaphore) }) { (refOwner, semaphore) -> + semaphore.increment() + while (semaphore.value < 2) { + } + + callDispose(refOwner) + GC.collect() + refOwner + } + + while (semaphore.value < 1) { + } + // At this point worker is spinning on semaphore. refOwner still contains reference to + // DisposableWorkerBoundReference, so referent is kept alive. + GC.collect() + assertNotNull(refValueWeak.value) + + return Triple(refWeak, refValueWeak, future) +} + +@Test +fun testDisposeInWorker() { + val semaphore: AtomicInt = AtomicInt(0) + + val worker = Worker.start() + + val (refWeak, refValueWeak, future) = disposeInWorker(worker, semaphore) + semaphore.increment() + val refOwner = future.result + + // At this point refOwner still has a reference, but it's explicitly disposed, + // so referent is destroyed. + GC.collect() + assertNotNull(refWeak.value) + assertNull(refValueWeak.value) + + worker.requestTermination().result +} + +@Test +fun testDisposeOnMainThreadAndAccessInWorker() { + val ref = DisposableWorkerBoundReference(A(3)) + assertEquals(3, ref.value.a) + + ref.dispose() + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, { ref }) { ref -> + var result = 0 + assertFailsWith { + result = ref.value.a + } + result + } + + val value = future.result + assertEquals(0, value) + worker.requestTermination().result +} + +@Test +fun testDisposeInWorkerAndAccessOnMainThread() { + val ref = DisposableWorkerBoundReference(A(3)) + assertEquals(3, ref.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, { ref }) { ref -> + ref.dispose() + } + + future.result + assertFailsWith { + ref.value.a + } + worker.requestTermination().result +} + +class B1 { + lateinit var b2: DisposableWorkerBoundReference +} + +data class B2(val b1: DisposableWorkerBoundReference) + +fun createCyclicGarbage(): Triple?>, WeakReference, WeakReference> { + val ref1 = DisposableWorkerBoundReference(B1()) + val ref1Owner: AtomicReference?> = AtomicReference(ref1) + val ref1Weak = WeakReference(ref1.value) + + val ref2 = DisposableWorkerBoundReference(B2(ref1)) + val ref2Weak = WeakReference(ref2.value) + + ref1.value.b2 = ref2 + + return Triple(ref1Owner, ref1Weak, ref2Weak) +} + +@Test +fun doesNotCollectCyclicGarbage() { + val (ref1Owner, ref1Weak, ref2Weak) = createCyclicGarbage() + + ref1Owner.value = null + GC.collect() + + // If these asserts fail, that means DisposableWorkerBoundReference managed to clean up cyclic garbage all by itself. + assertNotNull(ref1Weak.value) + assertNotNull(ref2Weak.value) +} + +@Test +fun collectCyclicGarbageWithExplicitDispose() { + val (ref1Owner, ref1Weak, ref2Weak) = createCyclicGarbage() + + callDispose(ref1Owner) + GC.collect() + + assertNull(ref1Weak.value) + assertNull(ref2Weak.value) +} + +fun createCrossThreadCyclicGarbage( + worker: Worker +): Triple?>, WeakReference, WeakReference> { + val ref1 = DisposableWorkerBoundReference(B1()) + val ref1Owner: AtomicReference?> = AtomicReference(ref1) + val ref1Weak = WeakReference(ref1.value) + + val future = worker.execute(TransferMode.SAFE, { ref1 }) { ref1 -> + val ref2 = DisposableWorkerBoundReference(B2(ref1)) + Pair(ref2, WeakReference(ref2.value)) + } + val (ref2, ref2Weak) = future.result + + ref1.value.b2 = ref2 + + return Triple(ref1Owner, ref1Weak, ref2Weak) +} + +@Test +fun doesNotCollectCrossThreadCyclicGarbage() { + val worker = Worker.start() + + val (ref1Owner, ref1Weak, ref2Weak) = createCrossThreadCyclicGarbage(worker) + + ref1Owner.value = null + GC.collect() + worker.execute(TransferMode.SAFE, {}) { GC.collect() }.result + + // If these asserts fail, that means DisposableWorkerBoundReference managed to clean up cyclic garbage all by itself. + assertNotNull(ref1Weak.value) + assertNotNull(ref2Weak.value) + + worker.requestTermination().result +} + +@Test +fun collectCrossThreadCyclicGarbageWithExplicitDispose() { + val worker = Worker.start() + + val (ref1Owner, ref1Weak, ref2Weak) = createCrossThreadCyclicGarbage(worker) + + callDispose(ref1Owner) + GC.collect() + worker.execute(TransferMode.SAFE, {}) { GC.collect() }.result + + assertNull(ref1Weak.value) + assertNull(ref2Weak.value) + + worker.requestTermination().result +} + +@Test +fun concurrentAccess() { + val workerCount = 10 + val workerUnlocker = AtomicInt(0) + + val ref = DisposableWorkerBoundReference(A(3)) + assertEquals(3, ref.value.a) + + val workers = Array(workerCount) { + Worker.start() + } + val futures = Array(workers.size) { + workers[it].execute(TransferMode.SAFE, { Pair(ref, workerUnlocker) }) { (ref, workerUnlocker) -> + while (workerUnlocker.value < 1) { + } + + assertFailsWith { + ref.value + } + Unit + } + } + workerUnlocker.increment() + + for (future in futures) { + future.result + } + + for (worker in workers) { + worker.requestTermination().result + } +} + +@Test +fun concurrentDispose() { + val workerCount = 10 + val workerUnlocker = AtomicInt(0) + + val ref = DisposableWorkerBoundReference(A(3)) + assertEquals(3, ref.value.a) + + val workers = Array(workerCount) { + Worker.start() + } + val futures = Array(workers.size) { + workers[it].execute(TransferMode.SAFE, { Pair(ref, workerUnlocker) }) { (ref, workerUnlocker) -> + while (workerUnlocker.value < 1) { + } + + ref.dispose() + } + } + workerUnlocker.increment() + + for (future in futures) { + future.result + } + + assertFailsWith { + ref.value.a + } + + for (worker in workers) { + worker.requestTermination().result + } +} + +@Test +fun concurrentDisposeAndAccess() { + val workerUnlocker = AtomicInt(0) + + val ref = DisposableWorkerBoundReference(A(3)) + assertEquals(3, ref.value.a) + + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, { Pair(ref, workerUnlocker) }) { (ref, workerUnlocker) -> + while (workerUnlocker.value < 1) { + } + + ref.dispose() + } + workerUnlocker.increment() + + var result = 0 + // This is a race, but it should either get value successfully or get IllegalStateException. + // Any other kind of failure is unacceptable. + try { + result = ref.value.a + } catch (e: IllegalStateException) { + result = 3 + } + assertEquals(3, result) + + future.result + worker.requestTermination().result +} diff --git a/runtime/src/main/cpp/Memory.cpp b/runtime/src/main/cpp/Memory.cpp index d5d1afd4802..1319c1315a8 100644 --- a/runtime/src/main/cpp/Memory.cpp +++ b/runtime/src/main/cpp/Memory.cpp @@ -35,6 +35,7 @@ #include "Natives.h" #include "Porting.h" #include "Runtime.h" +#include "WorkerBoundReference.h" // If garbage collection algorithm for cyclic garbage to be used. // We are using the Bacon's algorithm for GC, see @@ -928,11 +929,17 @@ void freeAggregatingFrozenContainer(ContainerHeader* container) { MEMORY_LOG("Freeing subcontainers done\n"); } -void runDeallocationHooks(ContainerHeader* container) { +// This is called from 2 places where it's unconditionally called, +// so better be inlined. +ALWAYS_INLINE void runDeallocationHooks(ContainerHeader* container) { ObjHeader* obj = reinterpret_cast(container + 1); for (int index = 0; index < container->objectCount(); index++) { + auto* type_info = obj->type_info(); + if (type_info == theWorkerBoundReferenceTypeInfo) { + DisposeWorkerBoundReference(obj); + } #if USE_CYCLIC_GC - if ((obj->type_info()->flags_ & TF_LEAK_DETECTOR_CANDIDATE) != 0) { + if ((type_info->flags_ & TF_LEAK_DETECTOR_CANDIDATE) != 0) { cyclicRemoveAtomicRoot(obj); } #endif // USE_CYCLIC_GC diff --git a/runtime/src/main/cpp/Types.h b/runtime/src/main/cpp/Types.h index b010cef5a6b..6f64ce6ac42 100644 --- a/runtime/src/main/cpp/Types.h +++ b/runtime/src/main/cpp/Types.h @@ -102,6 +102,7 @@ extern const TypeInfo* theShortArrayTypeInfo; extern const TypeInfo* theStringTypeInfo; extern const TypeInfo* theThrowableTypeInfo; extern const TypeInfo* theUnitTypeInfo; +extern const TypeInfo* theWorkerBoundReferenceTypeInfo; KBoolean IsInstance(const ObjHeader* obj, const TypeInfo* type_info) RUNTIME_PURE; void CheckCast(const ObjHeader* obj, const TypeInfo* type_info); diff --git a/runtime/src/main/cpp/WorkerBoundReference.cpp b/runtime/src/main/cpp/WorkerBoundReference.cpp new file mode 100644 index 00000000000..376ea225cf1 --- /dev/null +++ b/runtime/src/main/cpp/WorkerBoundReference.cpp @@ -0,0 +1,44 @@ +/* + * Copyright 2010-2020 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. + */ + +#include "WorkerBoundReference.h" + +#include "Alloc.h" +#include "Memory.h" +#include "MemorySharedRefs.hpp" + +namespace { + +struct WorkerBoundReference { + ObjHeader header; + KRefSharedHolder* holder; +}; + +WorkerBoundReference* asWorkerBoundReference(KRef thiz) { + return reinterpret_cast(thiz); +} + +} // namespace + +RUNTIME_NOTHROW void DisposeWorkerBoundReference(KRef thiz) { + // DisposeSharedRef is only called when all references to thiz are gone. + auto* holder = asWorkerBoundReference(thiz)->holder; + holder->dispose(); + konanDestructInstance(holder); +} + +extern "C" { + +KNativePtr Kotlin_WorkerBoundReference_create(KRef value) { + auto* holder = konanConstructInstance(); + holder->init(value); + return holder; +} + +OBJ_GETTER(Kotlin_WorkerBoundReference_deref, KNativePtr holder) { + RETURN_OBJ(reinterpret_cast(holder)->ref()); +} + +} diff --git a/runtime/src/main/cpp/WorkerBoundReference.h b/runtime/src/main/cpp/WorkerBoundReference.h new file mode 100644 index 00000000000..1c950dec3f9 --- /dev/null +++ b/runtime/src/main/cpp/WorkerBoundReference.h @@ -0,0 +1,16 @@ +/* + * Copyright 2010-2020 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. + */ + +#ifndef RUNTIME_SHAREDREF_H +#define RUNTIME_SHAREDREF_H + +#include "Common.h" +#include "Types.h" + +// Not inlining this call as it affects deallocation performance for +// all types. +RUNTIME_NOTHROW void DisposeWorkerBoundReference(KRef thiz) NO_INLINE; + +#endif // RUNTIME_SHAREDREF_H diff --git a/runtime/src/main/kotlin/kotlin/native/concurrent/WorkerBoundReference.kt b/runtime/src/main/kotlin/kotlin/native/concurrent/WorkerBoundReference.kt new file mode 100644 index 00000000000..5fbf3cb3d6c --- /dev/null +++ b/runtime/src/main/kotlin/kotlin/native/concurrent/WorkerBoundReference.kt @@ -0,0 +1,71 @@ +/* + * Copyright 2010-2020 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 kotlin.native.concurrent + +import kotlin.native.internal.* + +@SymbolName("Kotlin_WorkerBoundReference_create") +external private fun createWorkerBoundReference(value: Any): NativePtr + +@SymbolName("Kotlin_WorkerBoundReference_deref") +external private fun derefWorkerBoundReference(ref: NativePtr): Any + +/** + * A frozen shared reference to a Kotlin object + * + * Can be safely passed between workers, but [value] can only be accessed on the worker [WorkerBoundReference] was created on, + * unless the referred object is frozen too + * + * Note: Garbage collector currently cannot free any reference cycles with [WorkerBoundReference] or [DisposableWorkerBoundReference] in them. + * Consider using [DisposableWorkerBoundReference] to manually resolve cycles with explicit calls to [DisposableWorkerBoundReference.dispose] + */ +@Frozen +@NoReorderFields +@ExportTypeInfo("theWorkerBoundReferenceTypeInfo") +public class WorkerBoundReference(value: T) { + + private val ptr = createWorkerBoundReference(value) + + /** + * The referenced value. + * @throws IncorrectDereferenceException if referred object is not frozen and current worker is different from the one created [this] + */ + val value: T + get() = @Suppress("UNCHECKED_CAST") (derefWorkerBoundReference(ptr) as T) +} + +/** + * A frozen shared reference to a Kotlin object + * + * Can be safely passed between workers, but [value] can only be accessed on the worker [DisposableWorkerBoundReference] was created on, + * unless the referred object is frozen too + * Garbage collector currently cannot free any reference cycles with [WorkerBoundReference] or [DisposableWorkerBoundReference] in them. + * Call [dispose] manually to resolve cycles + * + * Note: This class has more expensive [value] getter than [WorkerBoundReference]. If you don't have reference + * cycles with [WorkerBoundReference] or [DisposableWorkerBoundReference], consider using [WorkerBoundReference] + */ +@Frozen +public class DisposableWorkerBoundReference(value: T) { + + private val ref: AtomicReference?> = AtomicReference(WorkerBoundReference(value)) + + /** + * Free the reference. Any call to [DisposableWorkerBoundReference.value] after that will + * fail with [IllegalStateException] + */ + fun dispose() { + ref.value = null + } + + /** + * The referenced value. + * @throws IncorrectDereferenceException if referred object is not frozen and current worker is different from the one created [this] + * @throws IllegalStateException if [DisposableWorkerBoundReference.dispose] was called on this reference. + */ + val value: T + get() = ref.value?.value ?: throw IllegalStateException("illegal attempt to dereference disposed $this") +}