From f818d7639fa37a51bb52dae2f402c76255c585c1 Mon Sep 17 00:00:00 2001 From: Nikolay Igotti Date: Fri, 31 Jan 2020 16:03:19 +0300 Subject: [PATCH] Shared cyclic garbage collector (#3742) --- backend.native/tests/build.gradle | 6 + .../tests/runtime/memory/cycle_collector.kt | 190 +++++++ .../tests/runtime/memory/leak_detector.kt | 24 +- runtime/src/main/cpp/Atomic.h | 2 +- runtime/src/main/cpp/CyclicCollector.cpp | 489 ++++++++++++++++++ runtime/src/main/cpp/CyclicCollector.h | 17 + runtime/src/main/cpp/Memory.cpp | 102 +++- runtime/src/main/cpp/Memory.h | 5 + runtime/src/main/cpp/Worker.cpp | 15 +- .../kotlin/native/concurrent/Atomics.kt | 7 +- .../main/kotlin/kotlin/native/internal/GC.kt | 20 + 11 files changed, 857 insertions(+), 20 deletions(-) create mode 100644 backend.native/tests/runtime/memory/cycle_collector.kt create mode 100644 runtime/src/main/cpp/CyclicCollector.cpp create mode 100644 runtime/src/main/cpp/CyclicCollector.h diff --git a/backend.native/tests/build.gradle b/backend.native/tests/build.gradle index eac8bf8a89f..221faebb4a4 100644 --- a/backend.native/tests/build.gradle +++ b/backend.native/tests/build.gradle @@ -2735,6 +2735,12 @@ standaloneTest("leak_detector") { source = "runtime/memory/leak_detector.kt" } +standaloneTest("cycle_collector") { + disabled = project.globalTestArgs.contains('-opt') || (project.testTarget == 'wasm32') // Needs debug build. + flags = ['-g'] + source = "runtime/memory/cycle_collector.kt" +} + standaloneTest("mpp1") { source = "codegen/mpp/mpp1.kt" flags = ['-tr', '-Xmulti-platform'] diff --git a/backend.native/tests/runtime/memory/cycle_collector.kt b/backend.native/tests/runtime/memory/cycle_collector.kt new file mode 100644 index 00000000000..88ec23dcce7 --- /dev/null +++ b/backend.native/tests/runtime/memory/cycle_collector.kt @@ -0,0 +1,190 @@ +import kotlin.native.concurrent.* +import kotlin.native.internal.GC +import kotlin.test.* + +fun test1() { + val a = AtomicReference(null) + val b = AtomicReference(null) + a.value = b + b.value = a +} + +class Holder(var other: Any?) + +fun test2() { + val array = arrayOf(AtomicReference(null), AtomicReference(null)) + val obj1 = Holder(array).freeze() + array[0].value = obj1 +} + +fun test3() { + val a1 = FreezableAtomicReference(null) + val head = Holder(null) + var current = head + repeat(30) { + val next = Holder(null) + current.other = next + current = next + } + a1.value = head + current.other = a1 + current.freeze() +} + + +fun makeIt(): Holder { + val atomic = AtomicReference(null) + val holder = Holder(atomic).freeze() + atomic.value = holder + return holder +} + + +fun test4() { + val holder = makeIt() + // To clean rc count coming from rememberNewContainer(). + kotlin.native.internal.GC.collect() + // Request cyclic collection. + kotlin.native.internal.GC.collectCyclic() + // Ensure we processed delayed release. + repeat(10) { + // Wait a bit and process queue. + Worker.current.park(10) + Worker.current.processQueue() + kotlin.native.internal.GC.collect() + } + val value = @Suppress("UNCHECKED_CAST") (holder.other as? AtomicReference?) + assertTrue(value != null) + assertTrue(value.value == holder) +} + +fun createRef(): AtomicReference { + val atomic1 = AtomicReference(null) + val atomic2 = AtomicReference(null) + atomic1.value = atomic2 + atomic2.value = atomic1 + return atomic1 +} + +class Holder2(var value: AtomicReference) { + fun switch() { + value = value.value as AtomicReference + } +} + +fun createHolder2() = Holder2(createRef()) + +fun test5() { + val holder = createHolder2() + kotlin.native.internal.GC.collect() + kotlin.native.internal.GC.collectCyclic() + Worker.current.park(100 * 1000) + holder.switch() + kotlin.native.internal.GC.collect() + Worker.current.park(100 * 1000) + withWorker { + executeAfter(0L, { + kotlin.native.internal.GC.collect() + }.freeze()) + } + Worker.current.park(1000) + assertTrue(holder.value.value != null) +} + +fun test6() { + val atomic = AtomicReference(null) + atomic.value = Pair(atomic, Holder(atomic)).freeze() +} + +fun createRoot(): AtomicReference { + val ref1 = AtomicReference(null) + val ref2 = AtomicReference(null) + + ref1.value = Holder(ref2).freeze() + ref2.value = Any().freeze() + + return ref1 +} + +fun test7() { + val ref1 = createRoot() + kotlin.native.internal.GC.collect() + + kotlin.native.internal.GC.collectCyclic() + Worker.current.park(500 * 1000L) + + withWorker { + executeAfter(0L, {}.freeze()) + Worker.current.park(500 * 1000L) + + val node = ref1.value as Holder + val ref2 = node.other as AtomicReference + assertTrue(ref2.value != null) + } +} + +fun array(size: Int) = Array(size, { null }) + +fun test8() { + val ref = AtomicReference(null) + val obj1 = array(2) + val obj2 = array(1) + val obj3 = array(2) + + obj1[0] = obj2 + obj1[1] = obj3 + + obj2[0] = obj3 + + obj3[0] = obj2 + obj3[1] = ref + + ref.value = obj1.freeze() +} + +fun createNode1(): Holder { + val ref = AtomicReference(null) + val node2 = Holder(ref) + val node1 = Holder(node2) + ref.value = node1.freeze() + + return node1 +} + +fun getNode2(): Holder { + val node1 = createNode1() + GC.collect() + + return node1.other as Holder +} + +fun test9() { + withWorker { + val node2 = getNode2() + executeAfter(10 * 1000L, { GC.collectCyclic() }.freeze()) + + GC.collect() + + Worker.current.park(50 * 1000L) + + execute(TransferMode.SAFE, {}, {}).result + + val ref = node2.other as AtomicReference + assertTrue(ref.value != null) + } +} + +fun main() { + kotlin.native.internal.GC.cyclicCollectorEnabled = true + test1() + test2() + test3() + test4() + repeat(10) { + test5() + } + test6() + test7() + test8() + test9() +} \ No newline at end of file diff --git a/backend.native/tests/runtime/memory/leak_detector.kt b/backend.native/tests/runtime/memory/leak_detector.kt index 641c2d7e930..b0d2cd8939d 100644 --- a/backend.native/tests/runtime/memory/leak_detector.kt +++ b/backend.native/tests/runtime/memory/leak_detector.kt @@ -17,10 +17,15 @@ fun dumpLeaks() { fun test1() { val a = AtomicReference(null) - a.value = a + val b = AtomicReference(null) + a.value = b + b.value = a val cycles = GC.detectCycles()!! assertEquals(1, cycles.size) - assertTrue(arrayOf(a).contentEquals(GC.findCycle(cycles[0])!!)) + val cycle = GC.findCycle(cycles[0])!! + assertEquals(2, cycle.size) + assertTrue(cycle.contains(a)) + assertTrue(cycle.contains(b)) a.value = null } @@ -47,6 +52,7 @@ fun test3() { } a1.value = head current.other = a1 + current.freeze() val cycles = GC.detectCycles()!! assertEquals(1, cycles.size) val cycle = GC.findCycle(cycles[0])!! @@ -54,8 +60,18 @@ fun test3() { a1.value = null } + +fun test4() { + val atomic = AtomicReference(null) + atomic.value = Pair(atomic, Holder(atomic)).freeze() +} + fun main() { - test1() + // We must disable cyclic collector here, to avoid interfering with cycle detector. + kotlin.native.internal.GC.cyclicCollectorEnabled = false + /*test1() test2() - test3() + test3() */ + test4() + kotlin.native.internal.GC.cyclicCollectorEnabled = true } \ No newline at end of file diff --git a/runtime/src/main/cpp/Atomic.h b/runtime/src/main/cpp/Atomic.h index 86cb802eca7..40510582901 100644 --- a/runtime/src/main/cpp/Atomic.h +++ b/runtime/src/main/cpp/Atomic.h @@ -41,7 +41,7 @@ ALWAYS_INLINE inline bool compareAndSet(volatile T* where, T expectedValue, T ne #pragma clang diagnostic push -#if KONAN_ANDROID && (KONAN_ARM32 || KONAN_X86) +#if (KONAN_ANDROID || KONAN_IOS || KONAN_WATCHOS) && (KONAN_ARM32 || KONAN_X86) // On 32-bit Android clang generates library calls for "large" atomic operations // and warns about "significant performance penalty". See more details here: // https://github.com/llvm/llvm-project/blob/ce56e1a1cc5714f4af5675dd963cfebed766d9e1/clang/lib/CodeGen/CGAtomic.cpp#L775 diff --git a/runtime/src/main/cpp/CyclicCollector.cpp b/runtime/src/main/cpp/CyclicCollector.cpp new file mode 100644 index 00000000000..e62d5a3a938 --- /dev/null +++ b/runtime/src/main/cpp/CyclicCollector.cpp @@ -0,0 +1,489 @@ +/* + * Copyright 2010-2020 JetBrains s.r.o. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#ifndef KONAN_NO_THREADS +#define WITH_WORKERS 1 +#endif + +#include "Alloc.h" +#include "Atomic.h" +#include "KAssert.h" +#include "Memory.h" +#include "Natives.h" +#include "Porting.h" +#include "Types.h" + +#if WITH_WORKERS +#include +#include +#endif + +#if WITH_WORKERS + +// Define to 1 to print collector traces. +#define TRACE_COLLECTOR 0 + +#if TRACE_COLLECTOR +#define COLLECTOR_LOG(...) konan::consolePrintf(__VA_ARGS__); +#else +#define COLLECTOR_LOG(...) +#endif + +/** + * Theory of operations: + * + * Kotlin/Native runtime has concurrent cyclic garbage collection for the shared mutable objects, + * such as `AtomicReference` and `FreezableAtomicReference` instances (further known as the atomic rootset). + * We perform such analysis by iterating over the transitive closure of the atomic rootset, and computing + * aggregated inner reference counter for rootset elements over this transitive closure. + * Collector runs in its own thread and is started by an explicit request or after certain time interval since last + * collection passes, thus its operation does not affect UI responsiveness in most cases. + * Atomic rootset is built by maintaining the set of all atomic and freezable atomic references objects. + * Elements whose transitive closure inner reference count matches the actual reference count are ones + * belonging to the garbage cycles and thus can be discarded. + * We ignore elements reachable from objects having external references (i.e. inner rc != real rc). + * If during computations of the aggregated RC there were modifications in the reference counts of + * elements of the atomic rootset: + * - if it is being increased, then someone already got an external reference to this element, thus we may not + * end up matching the inner reference count anyway + * - if it is being decreased and object become garbage, it will be collected next time + * If transitive closure of the atomic rootset mutates, it could only happen via changing the atomics references, + * as all elements of this closure are frozen. + * To handle such mutations we keep collector flag, which is cleared before analysis and set on every + * atomic reference value update. If flag's value changes - collector restarts its analysis. + * There are not so much of complications in this algorithm due to the delayed reference counting as if there's a + * stack reference to the shared object - it's reflected in the reference counter (see rememberNewContainer()). + * We release objects found by the collector on a rendezvouz callback, but not on the main thread, + * to keep UI responsive, as taking GC lock can take time, sometimes. + */ +namespace { + +class Locker { + pthread_mutex_t* lock_; + + public: + Locker(pthread_mutex_t* alock): lock_(alock) { + pthread_mutex_lock(lock_); + } + + ~Locker() { + pthread_mutex_unlock(lock_); + } +}; + +template +inline void traverseObjectFields(ObjHeader* obj, func process) { + RuntimeAssert(obj != nullptr, "Must be non null"); + const TypeInfo* typeInfo = obj->type_info(); + if (typeInfo != theArrayTypeInfo) { + for (int index = 0; index < typeInfo->objOffsetsCount_; index++) { + ObjHeader** location = reinterpret_cast( + reinterpret_cast(obj) + typeInfo->objOffsets_[index]); + process(location); + } + } else { + ArrayHeader* array = obj->array(); + for (int index = 0; index < array->count_; index++) { + process(ArrayAddressOfElementAt(array, index)); + } + } +} + +inline bool isAtomicReference(ObjHeader* obj) { + return (obj->type_info()->flags_ & TF_LEAK_DETECTOR_CANDIDATE) != 0; +} + +#define CHECK_CALL(call, message) RuntimeCheck((call) == 0, message) + +class CyclicCollector { + pthread_mutex_t lock_; + pthread_mutex_t timestampLock_; + pthread_cond_t cond_; + pthread_t gcThread_; + + int currentAliveWorkers_; + int gcRunning_; + int mutatedAtomics_; + int pendingRelease_; + bool shallRunCollector_; + bool terminateCollector_; + int32_t currentTick_; + int32_t lastTick_; + int64_t lastTimestampUs_; + void* mainWorker_; + KStdUnorderedSet rootset_; + KStdUnorderedSet toRelease_; + + public: + CyclicCollector() { + CHECK_CALL(pthread_mutex_init(&lock_, nullptr), "Cannot init collector mutex") + CHECK_CALL(pthread_mutex_init(×tampLock_, nullptr), "Cannot init collector timestamp mutex") + CHECK_CALL(pthread_cond_init(&cond_, nullptr), "Cannot init collector condition") + CHECK_CALL(pthread_create(&gcThread_, nullptr, gcWorkerRoutine, this), "Cannot start collector thread") + } + + ~CyclicCollector() { + { + Locker locker(&lock_); + terminateCollector_ = true; + shallRunCollector_ = true; + CHECK_CALL(pthread_cond_signal(&cond_), "Cannot signal collector") + } + // TODO: improve waiting for collector termination. + while (atomicGet(&terminateCollector_)) {} + releasePendingUnlocked(nullptr); + pthread_cond_destroy(&cond_); + pthread_mutex_destroy(&lock_); + pthread_mutex_destroy(×tampLock_); + } + + static void* gcWorkerRoutine(void* argument) { + CyclicCollector* thiz = reinterpret_cast(argument); + thiz->gcProcessor(); + return nullptr; + } + + void gcProcessor() { + { + Locker locker(&lock_); + KStdDeque toVisit; + KStdUnorderedSet visited; + KStdUnorderedMap sideRefCounts; + int restartCount = 0; + while (!terminateCollector_) { + CHECK_CALL(pthread_cond_wait(&cond_, &lock_), "Cannot wait collector condition") + if (!shallRunCollector_) continue; + atomicSet(&gcRunning_, 1); + restartCount = 0; + restart: + COLLECTOR_LOG("start cycle GC\n"); + if (restartCount > 10) { + COLLECTOR_LOG("wait for some time to avoid GC trashing\n"); + struct timeval tv; + struct timespec ts; + long long nsDelta = 1000LL * 1000LL * (restartCount - 10); + ts.tv_nsec = (tv.tv_usec * 1000LL + nsDelta) % 1000000000LL; + ts.tv_sec = (tv.tv_sec * 1000000000LL + tv.tv_usec * 1000LL + nsDelta) / 1000000000LL ; + pthread_cond_timedwait(&cond_, &lock_, &ts); + } + atomicSet(&mutatedAtomics_, 0); + visited.clear(); + toVisit.clear(); + sideRefCounts.clear(); + for (auto* root: rootset_) { + // We only care about frozen values here, as only they could become part of shared cycles. + if (!root->container()->frozen()) continue; + COLLECTOR_LOG("process root %p\n", root); + toVisit.push_back(root); + sideRefCounts[root] = 0; + } + while (toVisit.size() > 0) { + if (atomicGet(&mutatedAtomics_) != 0) { + COLLECTOR_LOG("restarted during rootset visit\n") + restartCount++; + goto restart; + } + auto* obj = toVisit.front(); + toVisit.pop_front(); + COLLECTOR_LOG("visit %s%p\n", isAtomicReference(obj) ? "atomic " : "", obj); + auto* objContainer = obj->container(); + if (objContainer == nullptr) continue; // Permanent object. + RuntimeCheck(objContainer->shareable(), "Must be shareable"); + if (visited.count(obj) == 0) { + visited.insert(obj); + traverseObjectFields(obj, [&toVisit, obj, &sideRefCounts](ObjHeader** location) { + ObjHeader* ref = *location; + if (ref != nullptr) { + COLLECTOR_LOG("object field %p in %p\n", ref, obj) + int increment; + // We shall not account for edges inside the same frozen container, unless it originates + // from an atomic reference. + if (isAtomicReference(obj) || (obj->container() != ref->container())) { + COLLECTOR_LOG("counting %p -> %p\n", obj, ref) + increment = 1; + } else { + COLLECTOR_LOG("not counting %p -> %p\n", obj, ref) + increment = 0; + } + sideRefCounts[ref] += increment; + toVisit.push_back(ref); + } + }); + } + } + // Now find all elements with external references, and mark objects reachable from them as non suitable + // for collection by setting their side reference count to -1. + toVisit.clear(); + for (auto it: sideRefCounts) { + auto* obj = it.first; + auto* objContainer = obj->container(); + if (objContainer == nullptr) continue; // Permanent object. + int refCount; + // If object is in aggregated container - sum up RC for all elements. + if (objContainer->objectCount() != 1) { + RuntimeAssert(objContainer->frozen(), "Must be frozen aggregate"); + ContainerHeader** subContainer = reinterpret_cast(objContainer + 1); + refCount = 0; + for (int i = 0; i < objContainer->objectCount(); ++i) { + auto* componentObj = reinterpret_cast((*subContainer) + 1); + refCount += sideRefCounts[componentObj]; + subContainer++; + } + } else { + refCount = it.second; + } + RuntimeAssert(refCount <= objContainer->refCount(), "Must properly count inner refs"); + if (refCount != objContainer->refCount()) { + COLLECTOR_LOG("for %p mismatched RC: %d vs %d, adding as possible root\n", obj, refCount, objContainer->refCount()) + toVisit.push_back(it.first); + } + } + visited.clear(); + while (toVisit.size() > 0) { + auto* obj = toVisit.front(); + toVisit.pop_front(); + auto* objContainer = obj->container(); + if (objContainer == nullptr) continue; // Permanent object. + RuntimeCheck(objContainer->shareable(), "Must be shareable"); + sideRefCounts[obj] = -1; + visited.insert(obj); + if (atomicGet(&mutatedAtomics_) != 0) { + COLLECTOR_LOG("restarted during reachable visit\n") + restartCount++; + goto restart; + } + traverseObjectFields(obj, [&toVisit, &visited](ObjHeader** location) { + ObjHeader* ref = *location; + if (ref != nullptr && (visited.count(ref) == 0)) { + toVisit.push_back(ref); + } + }); + } + // Now release all atomic roots with matching reference counters, as only their destruction is controlled. + for (auto it: sideRefCounts) { + auto* obj = it.first; + // Only do that for atomic rootset elements. For them we also do not have sum up references from + // other elements of an aggregate, as atomic references are always in single object containers. + if (!isAtomicReference(obj)) { + continue; + } + if (atomicGet(&mutatedAtomics_) != 0) { + COLLECTOR_LOG("restarted during matching check\n") + restartCount++; + goto restart; + } + auto* objContainer = obj->container(); + if (!objContainer->frozen()) continue; + RuntimeAssert(objContainer->objectCount() == 1, "Must be single object"); + COLLECTOR_LOG("for %p inner %d actual %d\n", obj, it.second, objContainer->refCount()); + // All references are inner. We compare the number of counted + // inner references with the number of non-stack references and per-thread ownership value + // (see rememberNewContainer()). + if (it.second == objContainer->refCount()) { + COLLECTOR_LOG("adding %p to release candidates\n", it.first); + toRelease_.insert(it.first); + } + } + if (toRelease_.size() > 0) + atomicSet(&pendingRelease_, 1); + atomicSet(&gcRunning_, 0); + shallRunCollector_ = false; + COLLECTOR_LOG("end cycle GC\n"); + } + } + atomicSet(&terminateCollector_, false); + } + + void addWorker(void* worker) { + suggestLockRelease(); + Locker lock(&lock_); + currentAliveWorkers_++; + if (mainWorker_ == nullptr) mainWorker_ = worker; + } + + void removeWorker(void* worker) { + suggestLockRelease(); + Locker lock(&lock_); + // When exiting the worker - we shall collect the cyclic garbage here. + shallRunCollector_ = true; + CHECK_CALL(pthread_cond_signal(&cond_), "Cannot signal collector") + currentAliveWorkers_--; + } + + void addRoot(ObjHeader* obj) { + COLLECTOR_LOG("add root %p\n", obj); + // TODO: we can only add root when collector is not processing, which looks like a limitation, + // instead we can add elements to the side buffer or have a separate lock for that. + suggestLockRelease(); + Locker lock(&lock_); + rootset_.insert(obj); + } + + void removeRoot(ObjHeader* obj) { + COLLECTOR_LOG("remove root %p\n", obj); + // Note that we can only remove root when the collector is not processing. + suggestLockRelease(); + Locker lock(&lock_); + toRelease_.erase(obj); + rootset_.erase(obj); + } + + void mutateRoot(ObjHeader* newValue) { + // TODO: consider optimization, when clearing value (setting to null) in atomic reference shall not lead + // to invalidation of the collector analysis state. + atomicSet(&mutatedAtomics_, 1); + } + + void suggestLockRelease() { + atomicSet(&mutatedAtomics_, 1); + } + + bool checkIfShallCollect() { + auto tick = atomicAdd(¤tTick_, 1); + auto delta = tick - atomicGet(&lastTick_); + if (delta > 10 || delta < 0) { + auto currentTimestampUs = konan::getTimeMicros(); + if (currentTimestampUs - atomicGet(&lastTimestampUs_) > 10000) { + // Do we care if this lock is not here? + Locker locker(×tampLock_); + lastTick_ = currentTick_; + lastTimestampUs_ = currentTimestampUs; + return true; + } + } + return false; + } + + void releasePendingUnlocked(void* worker) { + // We are not doing that on the UI thread, as taking lock is slow, unless + // it happens on deinit of the collector or if there are no other workers. + if ((atomicGet(&pendingRelease_) != 0) && ((worker != mainWorker_) || (currentAliveWorkers_ == 1))) { + suggestLockRelease(); + Locker locker(&lock_); + COLLECTOR_LOG("clearing %d release candidates on %p\n", toRelease_.size(), worker); + for (auto* it: toRelease_) { + COLLECTOR_LOG("clear references in %p\n", it) + traverseObjectFields(it, [](ObjHeader** location) { + ZeroHeapRef(location); + }); + } + toRelease_.clear(); + atomicSet(&pendingRelease_, 0); + } + } + + void collectorCallaback(void* worker) { + if (atomicGet(&gcRunning_) != 0) return; + releasePendingUnlocked(worker); + if (checkIfShallCollect()) { + Locker locker(&lock_); + shallRunCollector_ = true; + CHECK_CALL(pthread_cond_signal(&cond_), "Cannot signal collector") + } + } + + void scheduleGarbageCollect() { + if (atomicGet(&gcRunning_) != 0) return; + Locker lock(&lock_); + shallRunCollector_ = true; + CHECK_CALL(pthread_cond_signal(&cond_), "Cannot signal collector") + } + + void localGC() { + // We just need to take GC lock here, to avoid release of object we walk on. + // TODO: consider optimization without taking the lock and just notifying collector via an atomic. + suggestLockRelease(); + Locker locker(&lock_); + } + +}; + +CyclicCollector* cyclicCollector = nullptr; + +} // namespace + +#endif // WITH_WORKERS + +void cyclicInit() { +#if WITH_WORKERS + RuntimeAssert(cyclicCollector == nullptr, "Must be not yet inited"); + cyclicCollector = konanConstructInstance(); +#endif +} + +void cyclicDeinit() { +#if WITH_WORKERS + RuntimeAssert(cyclicCollector != nullptr, "Must be inited"); + auto* local = cyclicCollector; + cyclicCollector = nullptr; + konanDestructInstance(local); +#endif // WITH_WORKERS +} + +void cyclicAddWorker(void* worker) { +#if WITH_WORKERS + if (cyclicCollector) + cyclicCollector->addWorker(worker); +#endif // WITH_WORKERS +} + +void cyclicRemoveWorker(void* worker) { +#if WITH_WORKERS + if (cyclicCollector) + cyclicCollector->removeWorker(worker); +#endif // WITH_WORKERS +} + +void cyclicCollectorCallback(void* worker) { +#if WITH_WORKERS + if (cyclicCollector) + cyclicCollector->collectorCallaback(worker); +#endif // WITH_WORKERS +} + +void cyclicScheduleGarbageCollect() { +#if WITH_WORKERS + if (cyclicCollector) + cyclicCollector->scheduleGarbageCollect(); +#endif // WITH_WORKERS +} + +void cyclicAddAtomicRoot(ObjHeader* obj) { +#if WITH_WORKERS + if (cyclicCollector) + cyclicCollector->addRoot(obj); +#endif // WITH_WORKERS +} + +void cyclicRemoveAtomicRoot(ObjHeader* obj) { +#if WITH_WORKERS + if (cyclicCollector) + cyclicCollector->removeRoot(obj); +#endif // WITH_WORKERS +} + +void cyclicMutateAtomicRoot(ObjHeader* newValue) { +#if WITH_WORKERS + if (cyclicCollector) + cyclicCollector->mutateRoot(newValue); +#endif // WITH_WORKERS +} + +void cyclicLocalGC() { +#if WITH_WORKERS + if (cyclicCollector) + cyclicCollector->localGC(); +#endif // WITH_WORKERS +} \ No newline at end of file diff --git a/runtime/src/main/cpp/CyclicCollector.h b/runtime/src/main/cpp/CyclicCollector.h new file mode 100644 index 00000000000..e958fbf0abf --- /dev/null +++ b/runtime/src/main/cpp/CyclicCollector.h @@ -0,0 +1,17 @@ +#ifndef RUNTIME_CYCLIC_COLLECTOR_H +#define RUNTIME_CYCLIC_COLLECTOR_H + +struct ObjHeader; + +void cyclicInit(); +void cyclicDeinit(); +void cyclicAddWorker(void* worker); +void cyclicRemoveWorker(void* worker); +void cyclicAddAtomicRoot(ObjHeader* obj); +void cyclicRemoveAtomicRoot(ObjHeader* obj); +void cyclicMutateAtomicRoot(ObjHeader* newValue); +void cyclicCollectorCallback(void* worker); +void cyclicLocalGC(); +void cyclicScheduleGarbageCollect(); + +#endif // RUNTIME_CYCLIC_COLLECTOR_H \ No newline at end of file diff --git a/runtime/src/main/cpp/Memory.cpp b/runtime/src/main/cpp/Memory.cpp index 55339ec0339..b9e436f0fcf 100644 --- a/runtime/src/main/cpp/Memory.cpp +++ b/runtime/src/main/cpp/Memory.cpp @@ -1,5 +1,5 @@ /* - * Copyright 2010-2018 JetBrains s.r.o. + * Copyright 2010-2020 JetBrains s.r.o. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -19,9 +19,15 @@ #include // for offsetof +// Allow concurrent global cycle collector. +#define USE_CYCLIC_GC 1 + #include "Alloc.h" #include "KAssert.h" #include "Atomic.h" +#if USE_CYCLIC_GC +#include "CyclicCollector.h" +#endif // USE_CYCLIC_GC #include "Exceptions.h" #include "KString.h" #include "Memory.h" @@ -114,6 +120,8 @@ KBoolean g_checkLeaks = KonanNeedDebugInfo; KRef g_leakCheckerGlobalList = nullptr; KInt g_leakCheckerGlobalLock = 0; +bool g_hasCyclicCollector = true; + // TODO: can we pass this variable as an explicit argument? THREAD_LOCAL_VARIABLE MemoryState* memoryState = nullptr; THREAD_LOCAL_VARIABLE FrameOverlay* currentFrame = nullptr; @@ -301,8 +309,6 @@ inline bool isShareable(ContainerHeader* container) { return container == nullptr || container->shareable(); } -void garbageCollect(); - } // namespace class ForeignRefManager { @@ -541,6 +547,7 @@ namespace { void freeContainer(ContainerHeader* header) NO_INLINE; #if USE_GC void garbageCollect(MemoryState* state, bool force) NO_INLINE; +void cyclicGarbageCollect() NO_INLINE; void rememberNewContainer(ContainerHeader* container); #endif // USE_GC @@ -927,6 +934,11 @@ void freeAggregatingFrozenContainer(ContainerHeader* container) { void runDeallocationHooks(ContainerHeader* container) { ObjHeader* obj = reinterpret_cast(container + 1); for (int index = 0; index < container->objectCount(); index++) { +#if USE_CYCLIC_GC + if ((obj->type_info()->flags_ & TF_LEAK_DETECTOR_CANDIDATE) != 0) { + cyclicRemoveAtomicRoot(obj); + } +#endif // USE_CYCLIC_GC if (obj->has_meta_object()) { if (KonanNeedDebugInfo && (obj->type_info()->flags_ & TF_LEAK_DETECTOR_CANDIDATE) != 0 && g_checkLeaks) { // Remove the object from the double-linked list of potentially cyclic objects. @@ -1600,6 +1612,12 @@ void garbageCollect(MemoryState* state, bool force) { state->gcEpoque++; incrementStack(state); +#if USE_CYCLIC_GC + // Block if the concurrent cycle collector is running. + // We must do that to ensure collector sees state where actual RC properly upper estimated. + if (g_hasCyclicCollector) + cyclicLocalGC(); +#endif // USE_CYCLIC_GC processDecrements(state); size_t beforeDecrements = state->toRelease->size(); decrementStack(state); @@ -1743,12 +1761,24 @@ MemoryState* initMemory() { #endif memoryState->tlsMap = konanConstructInstance(); memoryState->foreignRefManager = ForeignRefManager::create(); - atomicAdd(&aliveMemoryStatesCount, 1); + bool firstMemoryState = atomicAdd(&aliveMemoryStatesCount, 1) == 1; + if (firstMemoryState) { +#if USE_CYCLIC_GC + cyclicInit(); +#endif // USE_CYCLIC_GC + } return memoryState; } void deinitMemory(MemoryState* memoryState) { #if USE_GC + bool lastMemoryState = atomicAdd(&aliveMemoryStatesCount, -1) == 0; + if (lastMemoryState) { + garbageCollect(memoryState, true); +#if USE_CYCLIC_GC + cyclicDeinit(); +#endif // USE_CYCLIC_GC + } // Actual GC only implemented in strict memory model at the moment. do { GC_LOG("Calling garbageCollect from DeinitMemory()\n") @@ -1766,8 +1796,6 @@ void deinitMemory(MemoryState* memoryState) { #endif // USE_GC - bool lastMemoryState = atomicAdd(&aliveMemoryStatesCount, -1) == 0; - #if TRACE_MEMORY if (IsStrictMemoryModel && lastMemoryState && allocCount > 0) { MEMORY_LOG("*** Memory leaks, leaked %d containers ***\n", allocCount); @@ -1937,6 +1965,11 @@ OBJ_GETTER(allocInstance, const TypeInfo* type_info) { old->meta_object()->LeakDetector.previous_ = obj; unlock(&g_leakCheckerGlobalLock); } +#if USE_CYCLIC_GC + if ((obj->type_info()->flags_ & TF_LEAK_DETECTOR_CANDIDATE) != 0) { + cyclicAddAtomicRoot(obj); + } +#endif // USE_CYCLIC_GC #if USE_GC if (Strict) { rememberNewContainer(container.header()); @@ -2082,9 +2115,14 @@ OBJ_GETTER(swapHeapRefLocked, if (shallRemember) *cookie = realCookie; } if (oldValue == expectedValue) { +#if USE_CYCLIC_GC + if (g_hasCyclicCollector) + cyclicMutateAtomicRoot(newValue); +#endif // USE_CYCLIC_GC SetHeapRef(location, newValue); } UpdateReturnRef(OBJ_RESULT, oldValue); + if (IsStrictMemoryModel && shallRemember && oldValue != nullptr && oldValue != expectedValue) { // Only remember container if it is not known to this thread (i.e. != expectedValue). rememberNewContainer(oldValue->container()); @@ -2097,10 +2135,13 @@ OBJ_GETTER(swapHeapRefLocked, return oldValue; } - void setHeapRefLocked(ObjHeader** location, ObjHeader* newValue, int32_t* spinlock, int32_t* cookie) { lock(spinlock); ObjHeader* oldValue = *location; +#if USE_CYCLIC_GC + if (g_hasCyclicCollector) + cyclicMutateAtomicRoot(newValue); +#endif // USE_CYCLIC_GC // We do not use UpdateRef() here to avoid having ReleaseRef() on old value under the lock. SetHeapRef(location, newValue); *cookie = computeCookie(); @@ -2610,6 +2651,7 @@ OBJ_GETTER0(detectCyclicReferences) { while (!toVisit.empty() && !seenToRoot) { KRef current = toVisit.front(); toVisit.pop_front(); + if (cyclic.count(current) != 0) continue; if (current == root) seenToRoot = true; // TODO: racy against concurrent mutators. if (seen.count(current) == 0) { @@ -2845,7 +2887,6 @@ ArrayHeader* ArenaContainer::PlaceArray(const TypeInfo* type_info, uint32_t coun return result; } - // API of the memory manager. extern "C" { @@ -3017,6 +3058,14 @@ void Kotlin_native_internal_GC_collect(KRef) { #endif } +void Kotlin_native_internal_GC_collectCyclic(KRef) { +#if USE_CYCLIC_GC + cyclicScheduleGarbageCollect(); +#else + ThrowIllegalArgumentException(); +#endif +} + void Kotlin_native_internal_GC_suspend(KRef) { #if USE_GC suspendGC(); @@ -3185,4 +3234,41 @@ KRef* LookupTLS(void** key, int index) { return start + index; } + +void GC_RegisterWorker(void* worker) { +#if USE_CYCLIC_GC + cyclicAddWorker(worker); +#endif // USE_CYCLIC_GC +} + +void GC_UnregisterWorker(void* worker) { +#if USE_CYCLIC_GC + cyclicRemoveWorker(worker); +#endif // USE_CYCLIC_GC +} + +void GC_CollectorCallback(void* worker) { +#if USE_CYCLIC_GC + if (g_hasCyclicCollector) + cyclicCollectorCallback(worker); +#endif // USE_CYCLIC_GC +} + +KBoolean Kotlin_native_internal_GC_getCyclicCollector() { +#if USE_CYCLIC_GC + return g_hasCyclicCollector; +#else + return false; +#endif // USE_CYCLIC_GC +} + +void Kotlin_native_internal_GC_setCyclicCollector(KBoolean value) { +#if USE_CYCLIC_GC + g_hasCyclicCollector = value; +#else + if (value) + ThrowIllegalArgumentException(); +#endif // USE_CYCLIC_GC +} + } // extern "C" diff --git a/runtime/src/main/cpp/Memory.h b/runtime/src/main/cpp/Memory.h index 3ec4039fe91..64c41c00321 100644 --- a/runtime/src/main/cpp/Memory.h +++ b/runtime/src/main/cpp/Memory.h @@ -557,6 +557,11 @@ void ClearTLSRecord(MemoryState* memory, void** key) RUNTIME_NOTHROW; // Lookup element in TLS object storage. ObjHeader** LookupTLS(void** key, int index) RUNTIME_NOTHROW; +// APIs for the async GC. +void GC_RegisterWorker(void* worker) RUNTIME_NOTHROW; +void GC_UnregisterWorker(void* worker) RUNTIME_NOTHROW; +void GC_CollectorCallback(void* worker) RUNTIME_NOTHROW; + #ifdef __cplusplus } #endif diff --git a/runtime/src/main/cpp/Worker.cpp b/runtime/src/main/cpp/Worker.cpp index 8b58357af88..eaf1fdd2b4d 100644 --- a/runtime/src/main/cpp/Worker.cpp +++ b/runtime/src/main/cpp/Worker.cpp @@ -259,11 +259,14 @@ class State { } Worker* addWorkerUnlocked(bool errorReporting, KRef customName) { - Locker locker(&lock_); - Worker* worker = konanConstructInstance(nextWorkerId(), errorReporting, - customName); - if (worker == nullptr) return nullptr; - workers_[worker->id()] = worker; + Worker* worker = nullptr; + { + Locker locker(&lock_); + worker = konanConstructInstance(nextWorkerId(), errorReporting, customName); + if (worker == nullptr) return nullptr; + workers_[worker->id()] = worker; + } + GC_RegisterWorker(worker); return worker; } @@ -283,6 +286,7 @@ class State { workers_.erase(it); } } + GC_UnregisterWorker(worker); konanDestructInstance(worker); } @@ -823,6 +827,7 @@ bool Worker::park(KLong timeoutMicroseconds, bool process) { } JobKind Worker::processQueueElement(bool blocking) { + GC_CollectorCallback(this); ObjHolder argumentHolder; ObjHolder resultHolder; if (terminated_) return JOB_TERMINATE; diff --git a/runtime/src/main/kotlin/kotlin/native/concurrent/Atomics.kt b/runtime/src/main/kotlin/kotlin/native/concurrent/Atomics.kt index 4df66e34ad1..a5f370657e9 100644 --- a/runtime/src/main/kotlin/kotlin/native/concurrent/Atomics.kt +++ b/runtime/src/main/kotlin/kotlin/native/concurrent/Atomics.kt @@ -222,7 +222,9 @@ private fun debugString(value: Any?): String { @Frozen @LeakDetectorCandidate @NoReorderFields -public class AtomicReference(private var value_: T) { +public class AtomicReference { + private var value_: T + // A spinlock to fix potential ARC race. private var lock: Int = 0 @@ -233,8 +235,9 @@ public class AtomicReference(private var value_: T) { * Creates a new atomic reference pointing to given [ref]. * @throws InvalidMutabilityException if reference is not frozen. */ - init { + constructor(value: T) { checkIfFrozen(value) + value_ = value } /** diff --git a/runtime/src/main/kotlin/kotlin/native/internal/GC.kt b/runtime/src/main/kotlin/kotlin/native/internal/GC.kt index f90a6e9cb92..5aeef5c5ca8 100644 --- a/runtime/src/main/kotlin/kotlin/native/internal/GC.kt +++ b/runtime/src/main/kotlin/kotlin/native/internal/GC.kt @@ -29,6 +29,12 @@ object GC { @SymbolName("Kotlin_native_internal_GC_collect") external fun collect() + /** + * Request global cyclic collector, operation is async and just triggers the collection. + */ + @SymbolName("Kotlin_native_internal_GC_collectCyclic") + external fun collectCyclic() + /** * Suspend garbage collection. Release candidates are still collected, but * GC algorithm is not executed. @@ -78,6 +84,14 @@ object GC { get() = getTuneThreshold() set(value) = setTuneThreshold(value) + + /** + * If cyclic collector for atomic references to be deployed. + */ + var cyclicCollectorEnabled: Boolean + get() = getCyclicCollectorEnabled() + set(value) = setCyclicCollectorEnabled(value) + /** * Detect cyclic references going via atomic references and return list of cycle-inducing objects * or `null` if the leak detector is not available. Use [Platform.isMemoryLeakCheckerActive] to check @@ -113,4 +127,10 @@ object GC { @SymbolName("Kotlin_native_internal_GC_setTuneThreshold") private external fun setTuneThreshold(value: Boolean) + + @SymbolName("Kotlin_native_internal_GC_getCyclicCollector") + private external fun getCyclicCollectorEnabled(): Boolean + + @SymbolName("Kotlin_native_internal_GC_setCyclicCollector") + private external fun setCyclicCollectorEnabled(value: Boolean) } \ No newline at end of file