Improve WorkerBoundReference exceptions and add valueOrNull (#4115)

This commit is contained in:
Alexander Shabalin
2020-04-23 18:31:43 +03:00
committed by GitHub
parent 1b6dac76d5
commit 5b75a009db
9 changed files with 241 additions and 756 deletions
-5
View File
@@ -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"
@@ -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<A> = 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<IncorrectDereferenceException> {
local.value
}
assertEquals(null, local.valueOrNull)
Unit
}
@@ -53,6 +57,7 @@ val global3: WorkerBoundReference<A> = 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<IncorrectDereferenceException> {
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<IncorrectDereferenceException> {
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<WorkerBoundReference<C2>?>
fun dispose() {
c2.value = null
}
}
data class C2(val c1: AtomicReference<WorkerBoundReference<C1>>)
fun createCyclicGarbageWithAtomics(): Triple<AtomicReference<WorkerBoundReference<C1>?>, WeakReference<C1>, WeakReference<C2>> {
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<WorkerBoundReference<C1>?>) {
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<WorkerBoundReference> 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<AtomicReference<WorkerBoundReference<C1>?>, WeakReference<C1>, WeakReference<C2>> {
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<WorkerBoundReference> 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<IncorrectDereferenceException> {
value.value
}
assertTrue(messagePattern matches exception.message!!)
worker.requestTermination().result
}
@@ -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<A> = 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<A> = 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<IncorrectDereferenceException> {
local.value
}
Unit
}
future.result
worker.requestTermination().result
}
val global3: DisposableWorkerBoundReference<A> = 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<A> = 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<A> = 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<A> = 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<A> = DisposableWorkerBoundReference(A(3))
@Test
fun testGlobalDispose() {
assertEquals(3, global7.value.a)
global7.dispose()
global7.dispose()
}
val global8: DisposableWorkerBoundReference<A> = DisposableWorkerBoundReference(A(3))
@Test
fun testGlobalAccessAfterDispose() {
assertEquals(3, global8.value.a)
global8.dispose()
assertFailsWith<IllegalStateException> {
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<IncorrectDereferenceException> {
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<IncorrectDereferenceException> {
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<IllegalStateException> {
local.value.a
}
}
fun getOwnerAndWeaks(initial: Int): Triple<AtomicReference<DisposableWorkerBoundReference<A>?>, WeakReference<DisposableWorkerBoundReference<A>>, WeakReference<A>> {
val ref = DisposableWorkerBoundReference(A(initial))
val refOwner: AtomicReference<DisposableWorkerBoundReference<A>?> = AtomicReference(ref)
val refWeak = WeakReference(ref)
val refValueWeak = WeakReference(ref.value)
return Triple(refOwner, refWeak, refValueWeak)
}
fun <T : Any> callDispose(ref: AtomicReference<DisposableWorkerBoundReference<T>?>) {
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<WeakReference<A>, Future<Unit>> {
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<DisposableWorkerBoundReference<A>> {
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<DisposableWorkerBoundReference<A>>, WeakReference<A>, Future<AtomicReference<DisposableWorkerBoundReference<A>?>>> {
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<IllegalStateException> {
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<IllegalStateException> {
ref.value.a
}
worker.requestTermination().result
}
class B1 {
lateinit var b2: DisposableWorkerBoundReference<B2>
}
data class B2(val b1: DisposableWorkerBoundReference<B1>)
fun createCyclicGarbage(): Triple<AtomicReference<DisposableWorkerBoundReference<B1>?>, WeakReference<B1>, WeakReference<B2>> {
val ref1 = DisposableWorkerBoundReference(B1())
val ref1Owner: AtomicReference<DisposableWorkerBoundReference<B1>?> = 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<AtomicReference<DisposableWorkerBoundReference<B1>?>, WeakReference<B1>, WeakReference<B2>> {
val ref1 = DisposableWorkerBoundReference(B1())
val ref1Owner: AtomicReference<DisposableWorkerBoundReference<B1>?> = 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<IncorrectDereferenceException> {
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<IllegalStateException> {
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
}
+47 -19
View File
@@ -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"
} // extern "C"
+12 -1
View File
@@ -6,6 +6,8 @@
#ifndef RUNTIME_MEMORYSHAREDREFS_HPP
#define RUNTIME_MEMORYSHAREDREFS_HPP
#include <type_traits>
#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<KRefSharedHolder>::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<BackRefFromAssociatedObject>::value,
"BackRefFromAssociatedObject destructor is not guaranteed to be called.");
#endif // RUNTIME_MEMORYSHAREDREFS_HPP
@@ -38,7 +38,11 @@ KNativePtr Kotlin_WorkerBoundReference_create(KRef value) {
}
OBJ_GETTER(Kotlin_WorkerBoundReference_deref, KNativePtr holder) {
RETURN_OBJ(reinterpret_cast<KRefSharedHolder*>(holder)->ref());
RETURN_OBJ(reinterpret_cast<KRefSharedHolder*>(holder)->refOrNull());
}
OBJ_GETTER(Kotlin_WorkerBoundReference_describe, KNativePtr holder) {
RETURN_RESULT_OF0(reinterpret_cast<KRefSharedHolder*>(holder)->describe);
}
}
@@ -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<Any>(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 ?: "<object>"
val unsignedIdentity = identity.toLong() and 0xffffffffL
val identityStr = unsignedIdentity.toString(16)
return "$className@$identityStr"
}
@SymbolName("Kotlin_AtomicReference_checkIfFrozen")
external internal fun checkIfFrozen(ref: Any?)
@@ -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<WorkerBoundReference?>] which can be explicitly
* nulled out.
*/
@Frozen
@NoReorderFields
@@ -28,44 +32,21 @@ external private fun derefWorkerBoundReference(ref: NativePtr): Any
public class WorkerBoundReference<out T : Any>(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<out T : Any>(value: T) {
private val ref: AtomicReference<WorkerBoundReference<T>?> = 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?)
}
@@ -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<Any>(typeInfo)
return debugDescription(kClass, address.toLong().toInt())
}
internal fun debugDescription(kClass: KClass<*>, identity: Int): String {
val className = kClass.qualifiedName ?: kClass.simpleName ?: "<object>"
val unsignedIdentity = identity.toLong() and 0xffffffffL
val identityStr = unsignedIdentity.toString(16)
return "$className@$identityStr"
}