From 5b75a009dbee8da38d80782d636a27a2a616eb66 Mon Sep 17 00:00:00 2001 From: Alexander Shabalin Date: Thu, 23 Apr 2020 18:31:43 +0300 Subject: [PATCH] Improve WorkerBoundReference exceptions and add valueOrNull (#4115) --- backend.native/tests/build.gradle | 5 - .../concurrent/worker_bound_reference0.kt | 135 ++++ .../concurrent/worker_bound_reference1.kt | 684 ------------------ runtime/src/main/cpp/MemorySharedRefs.cpp | 66 +- runtime/src/main/cpp/MemorySharedRefs.hpp | 13 +- runtime/src/main/cpp/WorkerBoundReference.cpp | 6 +- .../kotlin/native/concurrent/Internal.kt | 12 +- .../native/concurrent/WorkerBoundReference.kt | 55 +- .../kotlin/kotlin/native/internal/Utils.kt | 21 + 9 files changed, 241 insertions(+), 756 deletions(-) delete mode 100644 backend.native/tests/runtime/concurrent/worker_bound_reference1.kt create mode 100644 runtime/src/main/kotlin/kotlin/native/internal/Utils.kt diff --git a/backend.native/tests/build.gradle b/backend.native/tests/build.gradle index 6358e65e775..74279662c58 100644 --- a/backend.native/tests/build.gradle +++ b/backend.native/tests/build.gradle @@ -885,11 +885,6 @@ task worker_bound_reference0(type: KonanLocalTest) { 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 index a5cc3cc6908..25be9c232bf 100644 --- a/backend.native/tests/runtime/concurrent/worker_bound_reference0.kt +++ b/backend.native/tests/runtime/concurrent/worker_bound_reference0.kt @@ -10,6 +10,7 @@ import kotlin.test.* import kotlin.native.concurrent.* import kotlin.native.internal.GC import kotlin.native.ref.WeakReference +import kotlin.text.Regex class A(var a: Int) @@ -18,6 +19,7 @@ val global1: WorkerBoundReference = WorkerBoundReference(A(3)) @Test fun testGlobal() { assertEquals(3, global1.value.a) + assertEquals(3, global1.valueOrNull?.a) val worker = Worker.start() val future = worker.execute(TransferMode.SAFE, {}) { @@ -26,6 +28,7 @@ fun testGlobal() { val value = future.result assertEquals(3, value.value.a) + assertEquals(3, value.valueOrNull?.a) worker.requestTermination().result } @@ -41,6 +44,7 @@ fun testGlobalDenyAccessOnWorker() { assertFailsWith { local.value } + assertEquals(null, local.valueOrNull) Unit } @@ -53,6 +57,7 @@ val global3: WorkerBoundReference = WorkerBoundReference(A(3).freeze()) @Test fun testGlobalAccessOnWorkerFrozenInitially() { assertEquals(3, global3.value.a) + assertEquals(3, global3.valueOrNull?.a) val worker = Worker.start() val future = worker.execute(TransferMode.SAFE, {}) { @@ -131,6 +136,7 @@ fun testGlobalModification() { val value = future.result assertEquals(4, value.value.a) + assertEquals(4, value.valueOrNull?.a) worker.requestTermination().result } @@ -138,6 +144,7 @@ fun testGlobalModification() { fun testLocal() { val local = WorkerBoundReference(A(3)) assertEquals(3, local.value.a) + assertEquals(3, local.valueOrNull?.a) val worker = Worker.start() val future = worker.execute(TransferMode.SAFE, { local }) { local -> @@ -146,6 +153,7 @@ fun testLocal() { val value = future.result assertEquals(3, value.value.a) + assertEquals(3, value.valueOrNull?.a) worker.requestTermination().result } @@ -159,6 +167,7 @@ fun testLocalDenyAccessOnWorker() { assertFailsWith { local.value } + assertEquals(null, local.valueOrNull) Unit } @@ -170,6 +179,7 @@ fun testLocalDenyAccessOnWorker() { fun testLocalAccessOnWorkerFrozenInitially() { val local = WorkerBoundReference(A(3).freeze()) assertEquals(3, local.value.a) + assertEquals(3, local.valueOrNull?.a) val worker = Worker.start() val future = worker.execute(TransferMode.SAFE, { local }) { local -> @@ -235,6 +245,7 @@ fun testLocalDenyAccessOnMainThread() { assertFailsWith { value.value } + assertEquals(null, value.valueOrNull) worker.requestTermination().result } @@ -261,6 +272,7 @@ fun testLocalModification() { val value = future.result assertEquals(4, value.value.a) + assertEquals(4, value.valueOrNull?.a) worker.requestTermination().result } @@ -421,6 +433,110 @@ fun doesNotCollectCrossThreadCyclicGarbage() { worker.requestTermination().result } +class C1 { + lateinit var c2: AtomicReference?> + + fun dispose() { + c2.value = null + } +} + +data class C2(val c1: AtomicReference>) + +fun createCyclicGarbageWithAtomics(): Triple?>, WeakReference, WeakReference> { + val ref1 = WorkerBoundReference(C1()) + val ref1Weak = WeakReference(ref1.value) + + val ref2 = WorkerBoundReference(C2(AtomicReference(ref1))) + val ref2Weak = WeakReference(ref2.value) + + ref1.value.c2 = AtomicReference(ref2) + + return Triple(AtomicReference(ref1), ref1Weak, ref2Weak) +} + +fun dispose(refOwner: AtomicReference?>) { + refOwner.value!!.value.dispose() + refOwner.value = null +} + +@Test +fun doesNotCollectCyclicGarbageWithAtomics() { + val (ref1Owner, ref1Weak, ref2Weak) = createCyclicGarbageWithAtomics() + + ref1Owner.value = null + GC.collect() + + // If these asserts fail, that means AtomicReference managed to clean up cyclic garbage all by itself. + assertNotNull(ref1Weak.value) + assertNotNull(ref2Weak.value) +} + +@Test +fun collectCyclicGarbageWithAtomics() { + val (ref1Owner, ref1Weak, ref2Weak) = createCyclicGarbageWithAtomics() + + dispose(ref1Owner) + GC.collect() + + assertNull(ref1Weak.value) + assertNull(ref2Weak.value) +} + +fun createCrossThreadCyclicGarbageWithAtomics( + worker: Worker +): Triple?>, WeakReference, WeakReference> { + val ref1 = WorkerBoundReference(C1()) + val ref1Weak = WeakReference(ref1.value) + + val future = worker.execute(TransferMode.SAFE, { ref1 }) { ref1 -> + val ref2 = WorkerBoundReference(C2(AtomicReference(ref1))) + Pair(ref2, WeakReference(ref2.value)) + } + val (ref2, ref2Weak) = future.result + + ref1.value.c2 = AtomicReference(ref2) + + return Triple(AtomicReference(ref1), ref1Weak, ref2Weak) +} + +@Test +fun doesNotCollectCrossThreadCyclicGarbageWithAtomics() { + val worker = Worker.start() + + val (ref1Owner, ref1Weak, ref2Weak) = createCrossThreadCyclicGarbageWithAtomics(worker) + + ref1Owner.value = null + GC.collect() + worker.execute(TransferMode.SAFE, {}) { GC.collect() }.result + + // If these asserts fail, that means AtomicReference managed to clean up cyclic garbage all by itself. + assertNotNull(ref1Weak.value) + assertNotNull(ref2Weak.value) + + worker.requestTermination().result +} + +@Test +fun collectCrossThreadCyclicGarbageWithAtomics() { + val worker = Worker.start() + + val (ref1Owner, ref1Weak, ref2Weak) = createCrossThreadCyclicGarbageWithAtomics(worker) + + dispose(ref1Owner) + // This marks C2 as gone on the main thread + GC.collect() + // This cleans up all the references from the worker thread and destroys C2, but C1 is still alive. + worker.execute(TransferMode.SAFE, {}) { GC.collect() }.result + // And this finally destroys C1 + GC.collect() + + assertNull(ref1Weak.value) + assertNull(ref2Weak.value) + + worker.requestTermination().result +} + @Test fun concurrentAccess() { val workerCount = 10 @@ -453,3 +569,22 @@ fun concurrentAccess() { worker.requestTermination().result } } + +@Test +fun testExceptionMessage() { + val worker = Worker.start() + val future = worker.execute(TransferMode.SAFE, {}) { + WorkerBoundReference(A(3)) + } + val value = future.result + + val ownerName = worker.name + val messagePattern = Regex("illegal attempt to access non-shared runtime\\.concurrent\\.worker_bound_reference0\\.A@[a-f0-9]+ bound to `$ownerName` from `${Worker.current.name}`") + + val exception = assertFailsWith { + value.value + } + assertTrue(messagePattern matches exception.message!!) + + 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 deleted file mode 100644 index 502cf06d4d1..00000000000 --- a/backend.native/tests/runtime/concurrent/worker_bound_reference1.kt +++ /dev/null @@ -1,684 +0,0 @@ -/* - * 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/MemorySharedRefs.cpp b/runtime/src/main/cpp/MemorySharedRefs.cpp index d08a633a098..c60bb103b12 100644 --- a/runtime/src/main/cpp/MemorySharedRefs.cpp +++ b/runtime/src/main/cpp/MemorySharedRefs.cpp @@ -7,6 +7,33 @@ #include "MemoryPrivate.hpp" #include "MemorySharedRefs.hpp" #include "Runtime.h" +#include "Types.h" + +extern "C" { +// Returns a string describing object at `address` of type `typeInfo`. +OBJ_GETTER(DescribeObjectForDebugging, KConstNativePtr typeInfo, KConstNativePtr address); +} // extern "C" + +namespace { + +inline bool isForeignRefAccessible(ObjHeader* object, ForeignRefContext context) { + if (!Kotlin_hasRuntime()) { + // So the object is either unowned or shared. + // In the former case initialized runtime is required to throw exceptions + // in the latter case -- to provide proper execution context for caller. + Kotlin_initRuntimeIfNeeded(); + } + + return IsForeignRefAccessible(object, context); +} + +RUNTIME_NORETURN inline void throwIllegalSharingException(ObjHeader* object) { + // TODO: add some info about the context. + // Note: retrieving 'type_info()' is supposed to be correct even for unowned object. + ThrowIllegalObjectSharingException(object->type_info(), object); +} + +} // namespace void KRefSharedHolder::initLocal(ObjHeader* obj) { RuntimeAssert(obj != nullptr, "must not be null"); @@ -21,24 +48,18 @@ void KRefSharedHolder::init(ObjHeader* obj) { } ObjHeader* KRefSharedHolder::ref() const { - ensureRefAccessible(); - AdoptReferenceFromSharedVariable(obj_); - return obj_; + if (auto* result = refOrNull()) + return result; + + throwIllegalSharingException(obj_); } -static inline void ensureForeignRefAccessible(ObjHeader* object, ForeignRefContext context) { - if (!Kotlin_hasRuntime()) { - // So the object is either unowned or shared. - // In the former case initialized runtime is required to throw the exception below, - // in the latter case -- to provide proper execution context for caller. - Kotlin_initRuntimeIfNeeded(); - } - - if (!IsForeignRefAccessible(object, context)) { - // TODO: add some info about the context. - // Note: retrieving 'type_info()' is supposed to be correct even for unowned object. - ThrowIllegalObjectSharingException(object->type_info(), object); +ObjHeader* KRefSharedHolder::refOrNull() const { + if (!isRefAccessible()) { + return nullptr; } + AdoptReferenceFromSharedVariable(obj_); + return obj_; } void KRefSharedHolder::dispose() const { @@ -50,8 +71,13 @@ void KRefSharedHolder::dispose() const { DeinitForeignRef(obj_, context_); } -void KRefSharedHolder::ensureRefAccessible() const { - ensureForeignRefAccessible(obj_, context_); +OBJ_GETTER0(KRefSharedHolder::describe) const { + // Note: retrieving 'type_info()' is supposed to be correct even for unowned object. + RETURN_RESULT_OF(DescribeObjectForDebugging, obj_->type_info(), obj_); +} + +bool KRefSharedHolder::isRefAccessible() const { + return isForeignRefAccessible(obj_, context_); } void BackRefFromAssociatedObject::initAndAddRef(ObjHeader* obj) { @@ -106,7 +132,9 @@ ObjHeader* BackRefFromAssociatedObject::ref() const { } void BackRefFromAssociatedObject::ensureRefAccessible() const { - ensureForeignRefAccessible(obj_, context_); + if (!isForeignRefAccessible(obj_, context_)) { + throwIllegalSharingException(obj_); + } } extern "C" { @@ -125,4 +153,4 @@ RUNTIME_NOTHROW void KRefSharedHolder_dispose(const KRefSharedHolder* holder) { ObjHeader* KRefSharedHolder_ref(const KRefSharedHolder* holder) { return holder->ref(); } -} // extern "C" \ No newline at end of file +} // extern "C" diff --git a/runtime/src/main/cpp/MemorySharedRefs.hpp b/runtime/src/main/cpp/MemorySharedRefs.hpp index bd291263e61..a32b78e98c9 100644 --- a/runtime/src/main/cpp/MemorySharedRefs.hpp +++ b/runtime/src/main/cpp/MemorySharedRefs.hpp @@ -6,6 +6,8 @@ #ifndef RUNTIME_MEMORYSHAREDREFS_HPP #define RUNTIME_MEMORYSHAREDREFS_HPP +#include + #include "Memory.h" class KRefSharedHolder { @@ -15,16 +17,22 @@ class KRefSharedHolder { void init(ObjHeader* obj); ObjHeader* ref() const; + ObjHeader* refOrNull() const; void dispose() const; + OBJ_GETTER0(describe) const; + private: ObjHeader* obj_; ForeignRefContext context_; - void ensureRefAccessible() const; + bool isRefAccessible() const; }; +static_assert(std::is_trivially_destructible::value, + "KRefSharedHolder destructor is not guaranteed to be called."); + class BackRefFromAssociatedObject { public: void initAndAddRef(ObjHeader* obj); @@ -49,4 +57,7 @@ class BackRefFromAssociatedObject { void ensureRefAccessible() const; }; +static_assert(std::is_trivially_destructible::value, + "BackRefFromAssociatedObject destructor is not guaranteed to be called."); + #endif // RUNTIME_MEMORYSHAREDREFS_HPP diff --git a/runtime/src/main/cpp/WorkerBoundReference.cpp b/runtime/src/main/cpp/WorkerBoundReference.cpp index 376ea225cf1..bc9b3a24259 100644 --- a/runtime/src/main/cpp/WorkerBoundReference.cpp +++ b/runtime/src/main/cpp/WorkerBoundReference.cpp @@ -38,7 +38,11 @@ KNativePtr Kotlin_WorkerBoundReference_create(KRef value) { } OBJ_GETTER(Kotlin_WorkerBoundReference_deref, KNativePtr holder) { - RETURN_OBJ(reinterpret_cast(holder)->ref()); + RETURN_OBJ(reinterpret_cast(holder)->refOrNull()); +} + +OBJ_GETTER(Kotlin_WorkerBoundReference_describe, KNativePtr holder) { + RETURN_RESULT_OF0(reinterpret_cast(holder)->describe); } } diff --git a/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt b/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt index fb29e3c5233..eb13f3309d6 100644 --- a/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt +++ b/runtime/src/main/kotlin/kotlin/native/concurrent/Internal.kt @@ -5,7 +5,9 @@ package kotlin.native.concurrent +import kotlin.native.internal.DescribeObjectForDebugging import kotlin.native.internal.ExportForCppRuntime +import kotlin.native.internal.debugDescription import kotlin.native.identityHashCode import kotlin.reflect.KClass import kotlinx.cinterop.* @@ -92,17 +94,9 @@ internal fun ThrowInvalidMutabilityException(where: Any): Nothing { @ExportForCppRuntime internal fun ThrowIllegalObjectSharingException(typeInfo: NativePtr, address: NativePtr) { - val kClass = kotlin.native.internal.KClassImpl(typeInfo) - val description = debugDescription(kClass, address.toLong().toInt()) + val description = DescribeObjectForDebugging(typeInfo, address) throw IncorrectDereferenceException("illegal attempt to access non-shared $description from other thread") } -private fun debugDescription(kClass: KClass<*>, identity: Int): String { - val className = kClass.qualifiedName ?: kClass.simpleName ?: "" - val unsignedIdentity = identity.toLong() and 0xffffffffL - val identityStr = unsignedIdentity.toString(16) - return "$className@$identityStr" -} - @SymbolName("Kotlin_AtomicReference_checkIfFrozen") external internal fun checkIfFrozen(ref: Any?) diff --git a/runtime/src/main/kotlin/kotlin/native/concurrent/WorkerBoundReference.kt b/runtime/src/main/kotlin/kotlin/native/concurrent/WorkerBoundReference.kt index 5fbf3cb3d6c..c3a953dc0f5 100644 --- a/runtime/src/main/kotlin/kotlin/native/concurrent/WorkerBoundReference.kt +++ b/runtime/src/main/kotlin/kotlin/native/concurrent/WorkerBoundReference.kt @@ -11,16 +11,20 @@ import kotlin.native.internal.* external private fun createWorkerBoundReference(value: Any): NativePtr @SymbolName("Kotlin_WorkerBoundReference_deref") -external private fun derefWorkerBoundReference(ref: NativePtr): Any +external private fun derefWorkerBoundReference(ref: NativePtr): Any? + +@SymbolName("Kotlin_WorkerBoundReference_describe") +external private fun describeWorkerBoundReference(ref: NativePtr): String /** - * A frozen shared reference to a Kotlin object + * 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 + * 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] + * Note: Garbage collector currently cannot free any reference cycles with [WorkerBoundReference] in them. + * To resolve such cycles consider using [AtomicReference] which can be explicitly + * nulled out. */ @Frozen @NoReorderFields @@ -28,44 +32,21 @@ external private fun derefWorkerBoundReference(ref: NativePtr): Any public class WorkerBoundReference(value: T) { private val ptr = createWorkerBoundReference(value) + private val ownerName = Worker.current.name + + private val valueDescription + get() = describeWorkerBoundReference(ptr) /** * The referenced value. - * @throws IncorrectDereferenceException if referred object is not frozen and current worker is different from the one created [this] + * @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)) + get() = valueOrNull ?: throw IncorrectDereferenceException("illegal attempt to access non-shared $valueDescription bound to `$ownerName` from `${Worker.current.name}`") /** - * Free the reference. Any call to [DisposableWorkerBoundReference.value] after that will - * fail with [IllegalStateException] + * The referenced value or null if referred object is not frozen and current worker is different from the one created [this]. */ - 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") + val valueOrNull: T? + get() = @Suppress("UNCHECKED_CAST") (derefWorkerBoundReference(ptr) as T?) } diff --git a/runtime/src/main/kotlin/kotlin/native/internal/Utils.kt b/runtime/src/main/kotlin/kotlin/native/internal/Utils.kt new file mode 100644 index 00000000000..d1d4435bc53 --- /dev/null +++ b/runtime/src/main/kotlin/kotlin/native/internal/Utils.kt @@ -0,0 +1,21 @@ +/* + * 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 kotlin.native.internal + +import kotlin.reflect.KClass + +@ExportForCppRuntime +internal fun DescribeObjectForDebugging(typeInfo: NativePtr, address: NativePtr): String { + val kClass = kotlin.native.internal.KClassImpl(typeInfo) + return debugDescription(kClass, address.toLong().toInt()) +} + +internal fun debugDescription(kClass: KClass<*>, identity: Int): String { + val className = kClass.qualifiedName ?: kClass.simpleName ?: "" + val unsignedIdentity = identity.toLong() and 0xffffffffL + val identityStr = unsignedIdentity.toString(16) + return "$className@$identityStr" +}