Shared cyclic garbage collector (#3742)

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']

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<Any?>(null)
+    val b = AtomicReference<Any?>(null)
+    a.value = b
+    b.value = a
+}
+
+class Holder(var other: Any?)
+
+fun test2() {
+    val array = arrayOf(AtomicReference<Any?>(null), AtomicReference<Any?>(null))
+    val obj1 = Holder(array).freeze()
+    array[0].value = obj1
+}
+
+fun test3() {
+    val a1 = FreezableAtomicReference<Any?>(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<Holder?>(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<Holder?>?)
+    assertTrue(value != null)
+    assertTrue(value.value == holder)
+}
+
+fun createRef(): AtomicReference<Any?> {
+    val atomic1 = AtomicReference<Any?>(null)
+    val atomic2 = AtomicReference<Any?>(null)
+    atomic1.value = atomic2
+    atomic2.value = atomic1
+    return atomic1
+}
+
+class Holder2(var value: AtomicReference<Any?>) {
+    fun switch() {
+        value = value.value as AtomicReference<Any?>
+    }
+}
+
+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<Any?>(null)
+    atomic.value = Pair(atomic, Holder(atomic)).freeze()
+}
+
+fun createRoot(): AtomicReference<Any?> {
+    val ref1 = AtomicReference<Any?>(null)
+    val ref2 = AtomicReference<Any?>(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<Any?>
+        assertTrue(ref2.value != null)
+    }
+}
+
+fun array(size: Int) = Array<Any?>(size, { null })
+
+fun test8() {
+    val ref = AtomicReference<Any?>(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<Any?>(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<Any?>
+        assertTrue(ref.value != null)
+    }
+}
+
+fun main() {
+    kotlin.native.internal.GC.cyclicCollectorEnabled = true
+    test1()
+    test2()
+    test3()
+    test4()
+    repeat(10) {
+        test5()
+    }
+    test6()
+    test7()
+    test8()
+    test9()
+}

backend.native/tests/runtime/memory/leak_detector.kt

@@ -17,10 +17,15 @@ fun dumpLeaks() {
 
 fun test1() {
     val a = AtomicReference<Any?>(null)
-    a.value = a
+    val b = AtomicReference<Any?>(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<Any?>(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
 }

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

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 <pthread.h>
+#include <sys/time.h>
+#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 <typename func>
+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<ObjHeader**>(
+          reinterpret_cast<uintptr_t>(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<ObjHeader*> rootset_;
+  KStdUnorderedSet<ObjHeader*> toRelease_;
+
+ public:
+  CyclicCollector() {
+    CHECK_CALL(pthread_mutex_init(&lock_, nullptr), "Cannot init collector mutex")
+    CHECK_CALL(pthread_mutex_init(&timestampLock_, 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(&timestampLock_);
+  }
+
+  static void* gcWorkerRoutine(void* argument) {
+    CyclicCollector* thiz = reinterpret_cast<CyclicCollector*>(argument);
+    thiz->gcProcessor();
+    return nullptr;
+  }
+
+  void gcProcessor() {
+     {
+       Locker locker(&lock_);
+       KStdDeque<ObjHeader*> toVisit;
+       KStdUnorderedSet<ObjHeader*> visited;
+       KStdUnorderedMap<ObjHeader*, int> 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<ContainerHeader**>(objContainer + 1);
+             refCount = 0;
+             for (int i = 0; i < objContainer->objectCount(); ++i) {
+                 auto* componentObj = reinterpret_cast<ObjHeader*>((*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(&currentTick_, 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(&timestampLock_);
+        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<CyclicCollector>();
+#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
+}

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

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 <cstddef> // 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<ObjHeader*>(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<KThreadLocalStorageMap>();
   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"

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

runtime/src/main/cpp/Worker.cpp

@@ -259,11 +259,14 @@ class State {
   }
 
   Worker* addWorkerUnlocked(bool errorReporting, KRef customName) {
-    Locker locker(&lock_);
-    Worker* worker = konanConstructInstance<Worker>(nextWorkerId(), errorReporting,
-        customName);
-    if (worker == nullptr) return nullptr;
-    workers_[worker->id()] = worker;
+    Worker* worker = nullptr;
+    {
+      Locker locker(&lock_);
+      worker = konanConstructInstance<Worker>(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;

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<T>(private var value_: T) {
+public class AtomicReference<T> {
+    private var value_: T
+
     // A spinlock to fix potential ARC race.
     private var lock: Int = 0
 
@@ -233,8 +235,9 @@ public class AtomicReference<T>(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
     }
 
     /**

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)
 }