azalea/kotlin-fork
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Sharing the mutable binary data (#2311)

Browse Source
This commit is contained in:
Nikolay Igotti
2018-11-19 13:16:39 +03:00
committed by GitHub
parent 17c0dae0e4
commit 3e10e15f97
8 changed files with 363 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Interop/Runtime/src/main/kotlin/kotlinx/cinterop/StableRef.kt
+8 -8
View File
@@ -26,9 +26,9 @@ typealias StableObjPtr = StableRef<*>
*
* Any [StableRef] should be manually [disposed][dispose]
*/
data class StableRef<out T : Any> @PublishedApi internal constructor(
private val stablePtr: COpaquePointer,
private val ref: T // Note: storing the reference itself to avoid unchecked casts.
@Suppress("NON_PUBLIC_PRIMARY_CONSTRUCTOR_OF_INLINE_CLASS")
public inline class StableRef<out T : Any> @PublishedApi internal constructor(
private val stablePtr: COpaquePointer
) {
companion object {
@@ -36,7 +36,7 @@ data class StableRef<out T : Any> @PublishedApi internal constructor(
/**
* Creates a handle for given object.
*/
fun <T : Any> create(any: T) = StableRef<T>(createStablePointer(any), any)
fun <T : Any> create(any: T) = StableRef<T>(createStablePointer(any))
/**
* Creates [StableRef] from given raw value.
@@ -46,7 +46,6 @@ data class StableRef<out T : Any> @PublishedApi internal constructor(
@Deprecated("Use CPointer<*>.asStableRef<T>() instead", ReplaceWith("ptr.asStableRef<T>()"))
fun fromValue(value: COpaquePointer) = value.asStableRef<Any>()
}
@Deprecated("Use .asCPointer() instead", ReplaceWith("this.asCPointer()"))
val value: COpaquePointer get() = this.asCPointer()
@@ -66,12 +65,13 @@ data class StableRef<out T : Any> @PublishedApi internal constructor(
/**
* Returns the object this handle was [created][StableRef.create] for.
*/
fun get() = this.ref
@Suppress("UNCHECKED_CAST")
fun get() = derefStablePointer(this.stablePtr) as T
}
/**
* Converts to [StableRef] this opaque pointer produced by [StableRef.asCPointer].
*/
inline fun <reified T : Any> CPointer<*>.asStableRef() =
StableRef<T>(this, derefStablePointer(this) as T)
inline fun <reified T : Any> CPointer<*>.asStableRef(): StableRef<T> = StableRef<T>(this)
.also { it.get() as T }
platformLibs/src/platform/osx/Contacts.def
+9
View File
@@ -0,0 +1,9 @@
depends = CFNetwork CoreFoundation Foundation Security darwin posix
language = Objective-C
package = platform.Contacts
headers = Contacts/Contacts.h
headerFilter = Contacts/**
compilerOpts = -framework Contacts
linkerOpts = -framework Contacts
runtime/src/main/cpp/Memory.cpp
+22 -13
View File
@@ -414,14 +414,14 @@ void KRefSharedHolder::initRefOwner() {
}
void KRefSharedHolder::verifyRefOwner() const {
// Note: checking for 'permanentOrFrozen()' and retrieving 'type_info()'
// Note: checking for 'shareable()' and retrieving 'type_info()'
// are supposed to be correct even for unowned object.
if (owner_ != memoryState) {
// Initialized runtime is required to throw the exception below
// or to provide proper execution context for shared objects:
if (memoryState == nullptr) Kotlin_initRuntimeIfNeeded();
if (!obj_->container()->permanentOrFrozen()) {
if (!obj_->container()->shareable()) {
// TODO: add some info about the owner.
ThrowIllegalObjectSharingException(obj_->type_info(), obj_);
}
@@ -648,7 +648,7 @@ void MarkGray(ContainerHeader* start) {
auto childContainer = ref->container();
RuntimeAssert(!isArena(childContainer), "A reference to local object is encountered");
if (!childContainer->permanentOrFrozen()) {
if (!childContainer->shareable()) {
childContainer->decRefCount<false>();
toVisit.push_front(childContainer);
}
@@ -675,7 +675,7 @@ void ScanBlack(ContainerHeader* start) {
traverseContainerReferredObjects(container, [&toVisit](ObjHeader* ref) {
auto childContainer = ref->container();
RuntimeAssert(!isArena(childContainer), "A reference to local object is encountered");
if (!childContainer->permanentOrFrozen()) {
if (!childContainer->shareable()) {
childContainer->incRefCount<false>();
if (useColor) {
if (childContainer->color() != CONTAINER_TAG_GC_BLACK)
@@ -744,7 +744,7 @@ void Scan(ContainerHeader* container) {
traverseContainerReferredObjects(container, [](ObjHeader* ref) {
auto childContainer = ref->container();
RuntimeAssert(!isArena(childContainer), "A reference to local object is encountered");
if (!childContainer->permanentOrFrozen()) {
if (!childContainer->shareable()) {
Scan(childContainer);
}
});
@@ -764,7 +764,7 @@ void CollectWhite(MemoryState* state, ContainerHeader* start) {
if (ref == nullptr) return;
auto childContainer = ref->container();
RuntimeAssert(!isArena(childContainer), "A reference to local object is encountered");
if (childContainer->permanentOrFrozen()) {
if (childContainer->shareable()) {
UpdateRef(location, nullptr);
} else {
toVisit.push_front(childContainer);
@@ -787,6 +787,7 @@ inline void AddRef(ContainerHeader* header) {
IncrementRC<false>(header);
break;
case CONTAINER_TAG_FROZEN:
case CONTAINER_TAG_ATOMIC:
IncrementRC<true>(header);
break;
default:
@@ -806,6 +807,7 @@ inline void Release(ContainerHeader* header, bool useCycleCollector) {
DecrementRC<false>(header, useCycleCollector);
break;
case CONTAINER_TAG_FROZEN:
case CONTAINER_TAG_ATOMIC:
DecrementRC<true>(header, useCycleCollector);
break;
default:
@@ -1537,7 +1539,7 @@ bool hasExternalRefs(ContainerHeader* container, ContainerHeaderSet* visited) {
bool result = container->refCount() != 0;
traverseContainerReferredObjects(container, [&result, visited](ObjHeader* ref) {
auto child = ref->container();
if (!child->permanentOrFrozen() && (visited->find(child) == visited->end())) {
if (!child->shareable() && (visited->find(child) == visited->end())) {
result |= hasExternalRefs(child, visited);
}
});
@@ -1560,7 +1562,7 @@ bool ClearSubgraphReferences(ObjHeader* root, bool checked) {
if (!checked) {
hasExternalRefs(container, &visited);
} else {
if (!container->permanentOrFrozen()) {
if (!container->shareable()) {
container->decRefCount<false>();
MarkGray<false>(container);
auto bad = hasExternalRefs(container, &visited);
@@ -1604,7 +1606,7 @@ void depthFirstTraversal(ContainerHeader* container, bool* hasCycles,
return;
}
ContainerHeader* objContainer = obj->container();
if (!objContainer->permanentOrFrozen()) {
if (!objContainer->shareable()) {
// Marked GREY, there's cycle.
if (objContainer->seen()) *hasCycles = true;
@@ -1647,7 +1649,7 @@ void freezeAcyclic(ContainerHeader* rootContainer) {
current->freeze();
traverseContainerReferredObjects(current, [current, &queue](ObjHeader* obj) {
ContainerHeader* objContainer = obj->container();
if (!objContainer->permanentOrFrozen()) {
if (!objContainer->shareable()) {
if (objContainer->marked())
queue.push_back(objContainer);
}
@@ -1666,7 +1668,7 @@ void freezeCyclic(ContainerHeader* rootContainer, const KStdVector<ContainerHead
reversedEdges.emplace(current, KStdVector<ContainerHeader*>(0));
traverseContainerReferredObjects(current, [current, &queue, &reversedEdges](ObjHeader* obj) {
ContainerHeader* objContainer = obj->container();
if (!objContainer->permanentOrFrozen()) {
if (!objContainer->shareable()) {
if (objContainer->marked())
queue.push_back(objContainer);
reversedEdges.emplace(objContainer, KStdVector<ContainerHeader*>(0)).first->second.push_back(current);
@@ -1698,7 +1700,7 @@ void freezeCyclic(ContainerHeader* rootContainer, const KStdVector<ContainerHead
for (auto* container : component) {
totalCount += container->refCount();
traverseContainerReferredObjects(container, [&internalRefsCount](ObjHeader* obj) {
if (!obj->container()->permanentOrFrozen())
if (!obj->container()->shareable())
++internalRefsCount;
});
}
@@ -1746,7 +1748,7 @@ void FreezeSubgraph(ObjHeader* root) {
// First check that passed object graph has no cycles.
// If there are cycles - run graph condensation on cyclic graphs using Kosoraju-Sharir.
ContainerHeader* rootContainer = root->container();
if (rootContainer->permanentOrFrozen()) return;
if (rootContainer->shareable()) return;
// Do DFS cycle detection.
bool hasCycles = false;
@@ -1872,4 +1874,11 @@ KBoolean Konan_ensureAcyclicAndSet(ObjHeader* where, KInt index, ObjHeader* what
return true;
}
void Kotlin_Any_share(ObjHeader* obj) {
auto container = obj->container();
if (container->shareable()) return;
RuntimeCheck(container->objectCount() == 1, "Must be a single object container");
container->makeShareable();
}
} // extern "C"
runtime/src/main/cpp/Memory.h
+15 -5
View File
@@ -27,14 +27,16 @@ typedef enum {
CONTAINER_TAG_NORMAL = 0,
// Container is frozen, could only refer to other frozen objects.
// Refcounter update is atomics.
CONTAINER_TAG_FROZEN = 1,
CONTAINER_TAG_FROZEN = 1 | 1, // shareable
// Stack container, no need to free, children cleanup still shall be there.
CONTAINER_TAG_STACK = 2,
// Those container tags shall not be refcounted.
// Permanent container, cannot refer to non-permanent containers, so no need to cleanup those.
CONTAINER_TAG_PERMANENT = 2,
// Stack container, no need to free, children cleanup still shall be there.
CONTAINER_TAG_STACK = 3,
CONTAINER_TAG_PERMANENT = 3 | 1, // shareable
// Atomic container, reference counter is atomically updated.
CONTAINER_TAG_ATOMIC = 5 | 1, // shareable
// Shift to get actual counter.
CONTAINER_TAG_SHIFT = 2,
CONTAINER_TAG_SHIFT = 3,
// Actual value to increment/decrement container by. Tag is in lower bits.
CONTAINER_TAG_INCREMENT = 1 << CONTAINER_TAG_SHIFT,
// Mask for container type.
@@ -83,10 +85,18 @@ struct ContainerHeader {
refCount_ = (refCount_ & ~CONTAINER_TAG_MASK) | CONTAINER_TAG_FROZEN;
}
inline void makeShareable() {
refCount_ = (refCount_ & ~CONTAINER_TAG_MASK) | CONTAINER_TAG_ATOMIC;
}
inline bool permanentOrFrozen() const {
return tag() == CONTAINER_TAG_PERMANENT || tag() == CONTAINER_TAG_FROZEN;
}
inline bool shareable() const {
return (tag() & 1) != 0; // CONTAINER_TAG_PERMANENT || CONTAINER_TAG_FROZEN || CONTAINER_TAG_ATOMIC
}
inline bool stack() const {
return (refCount_ & CONTAINER_TAG_MASK) == CONTAINER_TAG_STACK;
}
runtime/src/main/cpp/Natives.cpp
+4
View File
@@ -78,4 +78,8 @@ const void* Kotlin_Any_getTypeInfo(KConstRef obj) {
return obj->type_info();
}
void Kotlin_CPointer_CopyMemory(KNativePtr to, KNativePtr from, KInt count) {
memcpy(to, from, count);
}
} // extern "C"
runtime/src/main/kotlin/kotlin/native/concurrent/Continuation.kt
+105
View File
@@ -0,0 +1,105 @@
/*
* 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.concurrent
import kotlin.native.internal.*
import kotlinx.cinterop.*
public class Continuation0(block: () -> Unit,
private val invoker: CPointer<CFunction<(COpaquePointer?) -> Unit>>,
private val singleShot: Boolean = false): Function0<Unit> {
private val stable = StableRef.create(block)
init {
freeze()
}
public override operator fun invoke() {
invoker(stable.asCPointer())
if (singleShot) {
stable.dispose()
}
}
public fun dispose() {
assert(!singleShot)
stable.dispose()
}
}
public class Continuation1<T1>(
block: (p1: T1) -> Unit,
private val invoker: CPointer<CFunction<(COpaquePointer?) -> Unit>>,
private val singleShot: Boolean = false) : Function1<T1, Unit> {
private val stable = StableRef.create(block)
init {
freeze()
}
public override operator fun invoke(p1: T1) {
val args = StableRef.create(Pair(stable, p1))
try {
invoker(args.asCPointer())
} finally {
args.dispose()
}
if (singleShot) {
stable.dispose()
}
}
public fun dispose() {
assert(!singleShot)
stable.dispose()
}
}
public class Continuation2<T1, T2>(
block: (p1: T1, p2: T2) -> Unit,
private val invoker: CPointer<CFunction<(COpaquePointer?) -> Unit>>,
private val singleShot: Boolean = false) : Function2<T1, T2, Unit> {
private val stable = StableRef.create(block)
init {
freeze()
}
public override operator fun invoke(p1: T1, p2: T2) {
val args = StableRef.create(Triple(stable, p1, p2))
try {
invoker(args.asCPointer())
} finally {
args.dispose()
}
if (singleShot) {
stable.dispose()
}
}
fun dispose() {
assert(!singleShot)
stable.dispose()
}
}
public fun COpaquePointer.callContinuation0() {
val single = this.asStableRef<() -> Unit>()
single.get()()
}
public fun <T1> COpaquePointer.callContinuation1() {
val pair = this.asStableRef<Pair<StableRef<(T1) -> Unit>, T1>>().get()
pair.first.get()(pair.second)
}
public fun <T1, T2> COpaquePointer.callContinuation2() {
val triple = this.asStableRef<Triple<StableRef<(T1, T2) -> Unit>, T1, T2>>().get()
triple.first.get()(triple.second, triple.third)
}
runtime/src/main/kotlin/kotlin/native/concurrent/MutableData.kt
+126
View File
@@ -0,0 +1,126 @@
/*
* 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.concurrent
import kotlin.native.internal.*
import kotlinx.cinterop.*
@SymbolName("Kotlin_Any_share")
external private fun Any.share()
@SymbolName("Kotlin_CPointer_CopyMemory")
external private fun CopyMemory(to: COpaquePointer?, from: COpaquePointer?, count: Int)
/**
* Mutable concurrently accessible data buffer. Could be accessed from several workers simulteniously.
*/
@Frozen
public class MutableData constructor(capacity: Int = 16) {
init {
if (capacity <= 0) throw IllegalArgumentException()
// Instance of MutableData is shared.
share()
}
private var buffer = ByteArray(capacity).apply { share() }
private var size_ = 0
private val lock = Lock()
private fun resizeDataLocked(newSize: Int): Int {
assert(newSize >= size)
if (newSize > buffer.size) {
val actualSize = maxOf(buffer.size * 3 / 2 + 1, newSize)
val newBuffer = ByteArray(actualSize)
buffer.copyRangeTo(newBuffer, 0, size, 0)
newBuffer.share()
buffer = newBuffer
}
val position = size
size_ = newSize
return position
}
/**
* Current data size, may concurrently change later on.
*/
public val size: Int
get() = size_
/**
* Reset the data buffer, makings its size 0.
*/
public fun reset() = locked(lock) {
size_ = 0
}
/**
* Appends data to the buffer.
*/
public fun append(data: MutableData) = locked(lock) {
val toCopy = data.size
val where = resizeDataLocked(size + toCopy)
data.copyInto(buffer, 0, toCopy, where)
}
/**
* Appends byte array to the buffer.
*/
public fun append(data: ByteArray, fromIndex: Int = 0, toIndex: Int = data.size): Unit = locked(lock) {
if (fromIndex > toIndex)
throw IndexOutOfBoundsException("$fromIndex is bigger than $toIndex")
if (toIndex == toIndex) return
val where = resizeDataLocked(this.size + (toIndex - fromIndex))
data.copyRangeTo(buffer, fromIndex, toIndex, where)
}
/**
* Appends C data to the buffer, if `data` is null or `count` is non-positive - return.
*/
public fun append(data: COpaquePointer?, count: Int): Unit = locked(lock) {
if (data == null || count <= 0) return
val where = resizeDataLocked(this.size + count)
buffer.usePinned {
it -> CopyMemory(it.addressOf(where), data, count)
}
}
/**
* Copies range of mutable data to the byte array.
*/
public fun copyInto(output: ByteArray, destinationIndex: Int, startIndex: Int, endIndex: Int): Unit = locked(lock) {
buffer.copyRangeTo(output, startIndex, endIndex, destinationIndex)
}
/**
* Get a byte from the mutable data.
*
* @Throws IndexOutOfBoundsException if index is beyond range.
*/
public operator fun get(index: Int): Byte = locked(lock) {
// index < 0 is checked below by array access.
if (index >= size)
throw IndexOutOfBoundsException("$index is not below $size")
buffer[index]
}
/**
* Executes provided block under lock with raw pointer to the data stored in the buffer.
* Block is executed under the spinlock, and must be short.
*/
public fun <R> withPointerLocked(block: (COpaquePointer, dataSize: Int) -> R) = locked(lock) {
buffer.usePinned {
it -> block(it.addressOf(0), size)
}
}
/**
* Executes provided block under lock with the raw data buffer.
* Block is executed under the spinlock, and must be short.
*/
public fun <R> withBufferLocked(block: (array: ByteArray, dataSize: Int) -> R) = locked(lock) {
block(buffer, size)
}
}
samples/objc/src/objcMain/kotlin/Window.kt
+74 -11
View File
@@ -7,12 +7,15 @@ package sample.objc
import kotlinx.cinterop.*
import platform.AppKit.*
import platform.Contacts.CNContactStore
import platform.Contacts.CNEntityType
import platform.Foundation.*
import platform.darwin.NSObject
import platform.darwin.dispatch_async_f
import kotlin.native.concurrent.DetachedObjectGraph
import kotlin.native.concurrent.attach
import kotlin.native.concurrent.freeze
import platform.darwin.dispatch_get_main_queue
import platform.darwin.dispatch_sync_f
import platform.posix.memcpy
import kotlin.native.concurrent.*
inline fun <reified T> executeAsync(queue: NSOperationQueue, crossinline producerConsumer: () -> Pair<T, (T) -> Unit>) {
dispatch_async_f(queue.underlyingQueue, DetachedObjectGraph {
@@ -23,8 +26,51 @@ inline fun <reified T> executeAsync(queue: NSOperationQueue, crossinline produce
})
}
inline fun mainContinuation(singleShot: Boolean = true, noinline block: () -> Unit) = Continuation0(
block, staticCFunction { invokerArg ->
if (NSThread.isMainThread()) {
invokerArg!!.callContinuation0()
} else {
dispatch_sync_f(dispatch_get_main_queue(), invokerArg, staticCFunction { args ->
args!!.callContinuation0()
})
}
}, singleShot)
inline fun <T1> mainContinuation(singleShot: Boolean = true, noinline block: (T1) -> Unit) = Continuation1(
block, staticCFunction { invokerArg ->
if (NSThread.isMainThread()) {
invokerArg!!.callContinuation1<T1>()
} else {
dispatch_sync_f(dispatch_get_main_queue(), invokerArg, staticCFunction { args ->
args!!.callContinuation1<T1>()
})
}
}, singleShot)
inline fun <T1, T2> mainContinuation(singleShot: Boolean = true, noinline block: (T1, T2) -> Unit) = Continuation2(
block, staticCFunction { invokerArg ->
if (NSThread.isMainThread()) {
invokerArg!!.callContinuation2<T1, T2>()
} else {
dispatch_sync_f(dispatch_get_main_queue(), invokerArg, staticCFunction { args ->
args!!.callContinuation2<T1, T2>()
})
}
}, singleShot)
data class QueryResult(val json: Map<String, *>?, val error: String?)
private fun MutableData.asNSData() = this.withPointerLocked { it, size ->
val result = NSMutableData.create(length = size.convert())!!
memcpy(result.mutableBytes, it, size.convert())
result
}
private fun MutableData.asJSON(): Map<String, *>? =
NSJSONSerialization.JSONObjectWithData(this.asNSData(), 0, null) as? Map<String, *>
fun main() {
autoreleasepool {
runApp()
@@ -45,7 +91,7 @@ private fun runApp() {
class Controller : NSObject() {
private var index = 1
private var httpDelegate = HttpDelegate()
private val httpDelegate = HttpDelegate()
@ObjCAction
fun onClick() {
@@ -63,16 +109,28 @@ class Controller : NSObject() {
NSApplication.sharedApplication().stop(this)
}
@ObjCAction
fun onRequest() {
val addressBookRef = CNContactStore()
addressBookRef.requestAccessForEntityType(CNEntityType.CNEntityTypeContacts, mainContinuation {
granted, error ->
appDelegate.contentText.string = if (granted)
"Access granted!"
else
"Access denied: $error"
})
}
class HttpDelegate: NSObject(), NSURLSessionDataDelegateProtocol {
private val asyncQueue = NSOperationQueue()
internal val receivedData = NSMutableData()
private val receivedData = MutableData()
init {
freeze()
}
fun fetchUrl(url: String) {
receivedData.setLength(0)
receivedData.reset()
val session = NSURLSession.sessionWithConfiguration(
NSURLSessionConfiguration.defaultSessionConfiguration(),
this,
@@ -83,7 +141,7 @@ class Controller : NSObject() {
override fun URLSession(session: NSURLSession, dataTask: NSURLSessionDataTask, didReceiveData: NSData) {
initRuntimeIfNeeded()
receivedData.appendData(didReceiveData)
receivedData.append(didReceiveData.bytes, didReceiveData.length.convert())
}
override fun URLSession(session: NSURLSession, task: NSURLSessionTask, didCompleteWithError: NSError?) {
@@ -94,10 +152,7 @@ class Controller : NSObject() {
Pair(when {
response == null -> QueryResult(null, didCompleteWithError?.localizedDescription)
response.statusCode.toInt() != 200 -> QueryResult(null, "${response.statusCode.toInt()})")
else -> QueryResult(
NSJSONSerialization.JSONObjectWithData(receivedData, 0, null) as? Map<String, *>,
null
)
else -> QueryResult(receivedData.asJSON(), null)
}, { result: QueryResult ->
appDelegate.contentText.string = result.json?.toString() ?: "Error: ${result.error}"
appDelegate.canClick = true
@@ -157,6 +212,14 @@ class MyAppDelegate() : NSObject(), NSApplicationDelegateProtocol {
action = NSSelectorFromString("onQuit")
}
window.contentView!!.addSubview(buttonQuit)
val buttonRequest = NSButton(NSMakeRect(230.0, 10.0, 100.0, 40.0)).apply {
title = "Request"
target = controller
action = NSSelectorFromString("onRequest")
}
window.contentView!!.addSubview(buttonRequest)
contentText = NSText(NSMakeRect(10.0, 80.0, 600.0, 350.0)).apply {
string = "Press 'Click' to start fetching"
verticallyResizable = false