[K/N] Fork concurrent-mark-and-sweep GC sources

kotlin-native/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/GC.kt

@@ -9,5 +9,5 @@ enum class GC(val shortcut: String? = null) {
     NOOP,
     STOP_THE_WORLD_MARK_AND_SWEEP("stwms"),
     PARALLEL_MARK_CONCURRENT_SWEEP("pmcs"),
-    // TODO: Bring back CONCURRENT_MARK_AND_SWEEP when we get concurrent mark
+    CONCURRENT_MARK_AND_SWEEP("cms"),
 }

kotlin-native/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/KonanConfig.kt

@@ -364,12 +364,14 @@ class KonanConfig(val project: Project, val configuration: CompilerConfiguration
                 GC.STOP_THE_WORLD_MARK_AND_SWEEP -> add("same_thread_ms_gc_custom.bc")
                 GC.NOOP -> add("noop_gc_custom.bc")
                 GC.PARALLEL_MARK_CONCURRENT_SWEEP -> add("pmcs_gc_custom.bc")
+                GC.CONCURRENT_MARK_AND_SWEEP -> add("concurrent_ms_gc_custom.bc")
             }
         } else {
             when (gc) {
                 GC.STOP_THE_WORLD_MARK_AND_SWEEP -> add("same_thread_ms_gc.bc")
                 GC.NOOP -> add("noop_gc.bc")
                 GC.PARALLEL_MARK_CONCURRENT_SWEEP -> add("pmcs_gc.bc")
+                GC.CONCURRENT_MARK_AND_SWEEP -> add("concurrent_ms_gc.bc")
             }
         }
         if (shouldCoverLibraries || shouldCoverSources) add("profileRuntime.bc")

kotlin-native/runtime/build.gradle.kts

@@ -325,6 +325,26 @@ bitcode {
             compilerArgs.add("-DCUSTOM_ALLOCATOR")
         }
 
+        module("concurrent_ms_gc") {
+            srcRoot.set(layout.projectDirectory.dir("src/gc/cms"))
+            headersDirs.from(files("src/alloc/common/cpp", "src/gcScheduler/common/cpp", "src/gc/common/cpp", "src/mm/cpp", "src/main/cpp", "src/alloc/legacy/cpp"))
+            sourceSets {
+                main {}
+                test {}
+            }
+        }
+
+        module("concurrent_ms_gc_custom") {
+            srcRoot.set(layout.projectDirectory.dir("src/gc/cms"))
+            headersDirs.from(files("src/alloc/common/cpp", "src/gcScheduler/common/cpp", "src/gc/common/cpp", "src/mm/cpp", "src/main/cpp", "src/alloc/custom/cpp"))
+            sourceSets {
+                main {}
+                test {}
+            }
+
+            compilerArgs.add("-DCUSTOM_ALLOCATOR")
+        }
+
         module("common_gcScheduler") {
             srcRoot.set(layout.projectDirectory.dir("src/gcScheduler/common"))
             headersDirs.from(files("src/alloc/common/cpp", "src/gc/common/cpp", "src/mm/cpp", "src/main/cpp"))
@@ -400,6 +420,19 @@ bitcode {
             testSupportModules.addAll("main", "common_alloc", "common_gc", "common_gcScheduler", "manual_gcScheduler", "custom_alloc", "objc")
         }
 
+        testsGroup("experimentalMM_cms_mimalloc_runtime_tests") {
+            testedModules.addAll("main", "mm", "common_alloc", "common_gc", "common_gcScheduler", "manual_gcScheduler", "concurrent_ms_gc", "mimalloc", "mimalloc_alloc", "legacy_alloc", "objc")
+        }
+
+        testsGroup("experimentalMM_cms_std_alloc_runtime_tests") {
+            testedModules.addAll("main", "mm", "common_alloc", "common_gc", "common_gcScheduler", "manual_gcScheduler", "concurrent_ms_gc", "std_alloc", "legacy_alloc", "objc")
+        }
+
+        testsGroup("experimentalMM_cms_custom_alloc_runtime_tests") {
+            testedModules.addAll("mm", "concurrent_ms_gc_custom")
+            testSupportModules.addAll("main", "common_alloc", "common_gc", "common_gcScheduler", "manual_gcScheduler", "custom_alloc", "objc")
+        }
+
         testsGroup("experimentalMM_noop_mimalloc_runtime_tests") {
             testedModules.addAll("main", "mm", "common_alloc", "common_gc", "common_gcScheduler", "manual_gcScheduler", "noop_gc", "mimalloc", "mimalloc_alloc", "legacy_alloc", "objc")
         }

kotlin-native/runtime/src/gc/cms/cpp/Barriers.cpp

@@ -0,0 +1,108 @@
+/*
+ * Copyright 2010-2023 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.
+ */
+
+#include "Barriers.hpp"
+
+#include <algorithm>
+#include <atomic>
+
+#include "GCImpl.hpp"
+#include "SafePoint.hpp"
+#include "ThreadData.hpp"
+#include "ThreadRegistry.hpp"
+
+using namespace kotlin;
+
+namespace {
+
+std::atomic<ObjHeader* (*)(ObjHeader*)> weakRefBarrier = nullptr;
+std::atomic<int64_t> weakProcessingEpoch = 0;
+
+ObjHeader* weakRefBarrierImpl(ObjHeader* weakReferee) noexcept {
+    if (!weakReferee) return nullptr;
+    // When weak ref barriers are enabled, marked state cannot change and the
+    // object cannot be deleted.
+    if (!gc::isMarked(weakReferee)) {
+        return nullptr;
+    }
+    return weakReferee;
+}
+
+NO_INLINE ObjHeader* weakRefReadSlowPath(std::atomic<ObjHeader*>& weakReferee) noexcept {
+    // reread an action to avoid register pollution outside the function
+    auto barrier = weakRefBarrier.load(std::memory_order_seq_cst);
+    auto* weak = weakReferee.load(std::memory_order_relaxed);
+    return barrier ? barrier(weak) : weak;
+}
+
+} // namespace
+
+void gc::BarriersThreadData::onThreadRegistration() noexcept {
+    if (weakRefBarrier.load(std::memory_order_acquire) != nullptr) {
+        startMarkingNewObjects(GCHandle::getByEpoch(weakProcessingEpoch.load(std::memory_order_relaxed)));
+    }
+}
+
+ALWAYS_INLINE void gc::BarriersThreadData::onSafePoint() noexcept {}
+
+void gc::BarriersThreadData::startMarkingNewObjects(gc::GCHandle gcHandle) noexcept {
+    RuntimeAssert(weakRefBarrier.load(std::memory_order_relaxed) != nullptr, "New allocations marking may only be requested by weak ref barriers");
+    markHandle_ = gcHandle.mark();
+}
+
+void gc::BarriersThreadData::stopMarkingNewObjects() noexcept {
+    RuntimeAssert(weakRefBarrier.load(std::memory_order_relaxed) == nullptr, "New allocations marking could only been requested by weak ref barriers");
+    markHandle_ = std::nullopt;
+}
+
+bool gc::BarriersThreadData::shouldMarkNewObjects() const noexcept {
+    return markHandle_.has_value();
+}
+
+ALWAYS_INLINE void gc::BarriersThreadData::onAllocation(ObjHeader* allocated) {
+    if (compiler::concurrentWeakSweep()) {
+        bool shouldMark = shouldMarkNewObjects();
+        bool barriersEnabled = weakRefBarrier.load(std::memory_order_relaxed) != nullptr;
+        RuntimeAssert(shouldMark == barriersEnabled, "New allocations marking must happen with and only with weak ref barriers");
+        if (shouldMark) {
+            auto& objectData = alloc::objectDataForObject(allocated);
+            objectData.markUncontended();
+            markHandle_->addObject();
+        }
+    }
+}
+
+void gc::EnableWeakRefBarriers(int64_t epoch) noexcept {
+    auto mutators = mm::ThreadRegistry::Instance().LockForIter();
+    weakProcessingEpoch.store(epoch, std::memory_order_relaxed);
+    weakRefBarrier.store(weakRefBarrierImpl, std::memory_order_seq_cst);
+    for (auto& mutator: mutators) {
+        mutator.gc().impl().gc().barriers().startMarkingNewObjects(GCHandle::getByEpoch(epoch));
+    }
+}
+
+void gc::DisableWeakRefBarriers() noexcept {
+    auto mutators = mm::ThreadRegistry::Instance().LockForIter();
+    weakRefBarrier.store(nullptr, std::memory_order_seq_cst);
+    for (auto& mutator: mutators) {
+        mutator.gc().impl().gc().barriers().stopMarkingNewObjects();
+    }
+}
+
+OBJ_GETTER(gc::WeakRefRead, std::atomic<ObjHeader*>& weakReferee) noexcept {
+    if (!compiler::concurrentWeakSweep()) {
+        RETURN_OBJ(weakReferee.load(std::memory_order_relaxed));
+    }
+
+    // Copying the scheme from SafePoint.cpp: branch + indirect call.
+    auto barrier = weakRefBarrier.load(std::memory_order_relaxed);
+    ObjHeader* result;
+    if (__builtin_expect(barrier != nullptr, false)) {
+        result = weakRefReadSlowPath(weakReferee);
+    } else {
+        result = weakReferee.load(std::memory_order_relaxed);
+    }
+    RETURN_OBJ(result);
+}

kotlin-native/runtime/src/gc/cms/cpp/Barriers.hpp

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2010-2023 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.
+ */
+
+#pragma once
+
+#include <atomic>
+#include <optional>
+
+#include "Memory.h"
+#include "Utils.hpp"
+#include "GCStatistics.hpp"
+
+namespace kotlin::gc {
+
+class BarriersThreadData : private Pinned {
+public:
+    void onThreadRegistration() noexcept;
+    void onSafePoint() noexcept;
+    
+    void startMarkingNewObjects(GCHandle gcHandle) noexcept;
+    void stopMarkingNewObjects() noexcept;
+    bool shouldMarkNewObjects() const noexcept;
+
+    void onAllocation(ObjHeader* allocated);
+private:
+    std::optional<GCHandle::GCMarkScope> markHandle_{};
+};
+
+// Must be called during STW.
+void EnableWeakRefBarriers(int64_t epoch) noexcept;
+void DisableWeakRefBarriers() noexcept;
+
+OBJ_GETTER(WeakRefRead, std::atomic<ObjHeader*>& weakReferee) noexcept;
+
+} // namespace kotlin::gc

kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.cpp

@@ -0,0 +1,233 @@
+/*
+ * Copyright 2010-2021 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.
+ */
+
+#include "ConcurrentMarkAndSweep.hpp"
+
+#include <optional>
+
+#include "AllocatorImpl.hpp"
+#include "CallsChecker.hpp"
+#include "CompilerConstants.hpp"
+#include "Logging.hpp"
+#include "MarkAndSweepUtils.hpp"
+#include "Memory.h"
+#include "ThreadData.hpp"
+#include "ThreadSuspension.hpp"
+#include "GCState.hpp"
+#include "GCStatistics.hpp"
+
+using namespace kotlin;
+
+namespace {
+
+[[clang::no_destroy]] std::mutex gcMutex;
+
+template<typename Body>
+ScopedThread createGCThread(const char* name, Body&& body) {
+    return ScopedThread(ScopedThread::attributes().name(name), [name, body] {
+        RuntimeLogDebug({kTagGC}, "%s %d starts execution", name, konan::currentThreadId());
+        body();
+        RuntimeLogDebug({kTagGC}, "%s %d finishes execution", name, konan::currentThreadId());
+    });
+}
+
+#ifndef CUSTOM_ALLOCATOR
+// TODO move to common
+[[maybe_unused]] inline void checkMarkCorrectness(alloc::ObjectFactoryImpl::Iterable& heap) {
+    if (compiler::runtimeAssertsMode() == compiler::RuntimeAssertsMode::kIgnore) return;
+    for (auto objRef: heap) {
+        auto obj = objRef.GetObjHeader();
+        auto& objData = objRef.ObjectData();
+        if (objData.marked()) {
+            traverseReferredObjects(obj, [obj](ObjHeader* field) {
+                if (field->heap()) {
+                    auto& fieldObjData = alloc::ObjectFactoryImpl::NodeRef::From(field).ObjectData();
+                    RuntimeAssert(fieldObjData.marked(), "Field %p of an alive obj %p must be alive", field, obj);
+                }
+            });
+        }
+    }
+}
+#endif
+
+} // namespace
+
+void gc::ConcurrentMarkAndSweep::ThreadData::OnSuspendForGC() noexcept {
+    CallsCheckerIgnoreGuard guard;
+
+    gc_.markDispatcher_.runOnMutator(commonThreadData());
+}
+
+bool gc::ConcurrentMarkAndSweep::ThreadData::tryLockRootSet() {
+    bool expected = false;
+    bool locked = rootSetLocked_.compare_exchange_strong(expected, true, std::memory_order_acq_rel);
+    if (locked) {
+        RuntimeLogDebug({kTagGC}, "Thread %d have exclusively acquired thread %d's root set", konan::currentThreadId(), threadData_.threadId());
+    }
+    return locked;
+}
+
+void gc::ConcurrentMarkAndSweep::ThreadData::publish() {
+    threadData_.Publish();
+    published_.store(true, std::memory_order_release);
+}
+
+bool gc::ConcurrentMarkAndSweep::ThreadData::published() const {
+    return published_.load(std::memory_order_acquire);
+}
+
+void gc::ConcurrentMarkAndSweep::ThreadData::clearMarkFlags() {
+    published_.store(false, std::memory_order_relaxed);
+    rootSetLocked_.store(false, std::memory_order_release);
+}
+
+mm::ThreadData& gc::ConcurrentMarkAndSweep::ThreadData::commonThreadData() const {
+    return threadData_;
+}
+
+gc::ConcurrentMarkAndSweep::ConcurrentMarkAndSweep(
+        alloc::Allocator& allocator, gcScheduler::GCScheduler& gcScheduler, bool mutatorsCooperate, std::size_t auxGCThreads) noexcept :
+    allocator_(allocator),
+    gcScheduler_(gcScheduler),
+    finalizerProcessor_([this](int64_t epoch) {
+        GCHandle::getByEpoch(epoch).finalizersDone();
+        state_.finalized(epoch);
+    }),
+    markDispatcher_(mutatorsCooperate),
+    mainThread_(createGCThread("Main GC thread", [this] { mainGCThreadBody(); })) {
+    for (std::size_t i = 0; i < auxGCThreads; ++i) {
+        auxThreads_.emplace_back(createGCThread("Auxiliary GC thread", [this] { auxiliaryGCThreadBody(); }));
+    }
+    RuntimeLogInfo({kTagGC}, "Parallel Mark & Concurrent Sweep GC initialized");
+}
+
+gc::ConcurrentMarkAndSweep::~ConcurrentMarkAndSweep() {
+    state_.shutdown();
+}
+
+void gc::ConcurrentMarkAndSweep::StartFinalizerThreadIfNeeded() noexcept {
+    NativeOrUnregisteredThreadGuard guard(true);
+    finalizerProcessor_.StartFinalizerThreadIfNone();
+    finalizerProcessor_.WaitFinalizerThreadInitialized();
+}
+
+void gc::ConcurrentMarkAndSweep::StopFinalizerThreadIfRunning() noexcept {
+    NativeOrUnregisteredThreadGuard guard(true);
+    finalizerProcessor_.StopFinalizerThread();
+}
+
+bool gc::ConcurrentMarkAndSweep::FinalizersThreadIsRunning() noexcept {
+    return finalizerProcessor_.IsRunning();
+}
+
+void gc::ConcurrentMarkAndSweep::mainGCThreadBody() {
+    RuntimeLogWarning({kTagGC}, "Initializing Concurrent Mark and Sweep GC. Concurrent mark is not implemented yet.");
+    while (true) {
+        auto epoch = state_.waitScheduled();
+        if (epoch.has_value()) {
+            PerformFullGC(*epoch);
+        } else {
+            break;
+        }
+    }
+    markDispatcher_.requestShutdown();
+}
+
+void gc::ConcurrentMarkAndSweep::auxiliaryGCThreadBody() {
+    RuntimeAssert(!compiler::gcMarkSingleThreaded(), "Should not reach here during single threaded mark");
+    while (!markDispatcher_.shutdownRequested()) {
+        markDispatcher_.runAuxiliary();
+    }
+}
+
+void gc::ConcurrentMarkAndSweep::PerformFullGC(int64_t epoch) noexcept {
+    std::unique_lock mainGCLock(gcMutex);
+    auto gcHandle = GCHandle::create(epoch);
+
+    markDispatcher_.beginMarkingEpoch(gcHandle);
+    GCLogDebug(epoch, "Main GC requested marking in mutators");
+
+    stopTheWorld(gcHandle);
+
+    auto& scheduler = gcScheduler_;
+    scheduler.onGCStart();
+
+    state_.start(epoch);
+
+    markDispatcher_.runMainInSTW();
+
+    markDispatcher_.endMarkingEpoch();
+
+    if (compiler::concurrentWeakSweep()) {
+        // Expected to happen inside STW.
+        gc::EnableWeakRefBarriers(epoch);
+        resumeTheWorld(gcHandle);
+    }
+
+    gc::processWeaks<DefaultProcessWeaksTraits>(gcHandle, mm::SpecialRefRegistry::instance());
+
+    if (compiler::concurrentWeakSweep()) {
+        stopTheWorld(gcHandle);
+        gc::DisableWeakRefBarriers();
+    }
+
+    // TODO outline as mark_.isolateMarkedHeapAndFinishMark()
+    // By this point all the alive heap must be marked.
+    // All the mutations (incl. allocations) after this method will be subject for the next GC.
+    // This should really be done by each individual thread while waiting
+    int threadCount = 0;
+    for (auto& thread : kotlin::mm::ThreadRegistry::Instance().LockForIter()) {
+        thread.allocator().prepareForGC();
+        ++threadCount;
+    }
+    allocator_.prepareForGC();
+
+#ifndef CUSTOM_ALLOCATOR
+    // Taking the locks before the pause is completed. So that any destroying thread
+    // would not publish into the global state at an unexpected time.
+    std::optional objectFactoryIterable = allocator_.impl().objectFactory().LockForIter();
+    std::optional extraObjectFactoryIterable = allocator_.impl().extraObjectDataFactory().LockForIter();
+
+    checkMarkCorrectness(*objectFactoryIterable);
+#endif
+
+    resumeTheWorld(gcHandle);
+
+#ifndef CUSTOM_ALLOCATOR
+    alloc::SweepExtraObjects<alloc::DefaultSweepTraits<alloc::ObjectFactoryImpl>>(gcHandle, *extraObjectFactoryIterable);
+    extraObjectFactoryIterable = std::nullopt;
+    auto finalizerQueue = alloc::Sweep<alloc::DefaultSweepTraits<alloc::ObjectFactoryImpl>>(gcHandle, *objectFactoryIterable);
+    objectFactoryIterable = std::nullopt;
+    alloc::compactObjectPoolInMainThread();
+#else
+    // also sweeps extraObjects
+    auto finalizerQueue = allocator_.impl().heap().Sweep(gcHandle);
+    for (auto& thread : kotlin::mm::ThreadRegistry::Instance().LockForIter()) {
+        finalizerQueue.TransferAllFrom(thread.allocator().impl().alloc().ExtractFinalizerQueue());
+    }
+    finalizerQueue.TransferAllFrom(allocator_.impl().heap().ExtractFinalizerQueue());
+#endif
+    scheduler.onGCFinish(epoch, gcHandle.getKeptSizeBytes() + threadCount * allocator_.estimateOverheadPerThread());
+    state_.finish(epoch);
+    gcHandle.finalizersScheduled(finalizerQueue.size());
+    gcHandle.finished();
+
+    // This may start a new thread. On some pthreads implementations, this may block waiting for concurrent thread
+    // destructors running. So, it must ensured that no locks are held by this point.
+    // TODO: Consider having an always on sleeping finalizer thread.
+    finalizerProcessor_.ScheduleTasks(std::move(finalizerQueue), epoch);
+}
+
+void gc::ConcurrentMarkAndSweep::reconfigure(std::size_t maxParallelism, bool mutatorsCooperate, std::size_t auxGCThreads) noexcept {
+    if (compiler::gcMarkSingleThreaded()) {
+        RuntimeCheck(auxGCThreads == 0, "Auxiliary GC threads must not be created with gcMarkSingleThread");
+        return;
+    }
+    std::unique_lock mainGCLock(gcMutex);
+    markDispatcher_.reset(maxParallelism, mutatorsCooperate, [this] { auxThreads_.clear(); });
+    for (std::size_t i = 0; i < auxGCThreads; ++i) {
+        auxThreads_.emplace_back(createGCThread("Auxiliary GC thread", [this] { auxiliaryGCThreadBody(); }));
+    }
+}

kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.hpp

@@ -0,0 +1,94 @@
+/*
+ * Copyright 2010-2021 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.
+ */
+
+#pragma once
+
+#include <atomic>
+#include <cstddef>
+#include <memory>
+#include <vector>
+
+#include "AllocatorImpl.hpp"
+#include "Barriers.hpp"
+#include "FinalizerProcessor.hpp"
+#include "GCScheduler.hpp"
+#include "GCState.hpp"
+#include "GCStatistics.hpp"
+#include "IntrusiveList.hpp"
+#include "MarkAndSweepUtils.hpp"
+#include "ObjectData.hpp"
+#include "ParallelMark.hpp"
+#include "ScopedThread.hpp"
+#include "ThreadData.hpp"
+#include "Types.h"
+#include "Utils.hpp"
+
+namespace kotlin {
+namespace gc {
+
+// TODO concurrent mark + concurrent sweep. The GC runs in a separate thread, finalizers run in another thread of their own.
+// TODO: Make marking run concurrently with Kotlin threads.
+class ConcurrentMarkAndSweep : private Pinned {
+public:
+    class ThreadData : private Pinned {
+    public:
+        explicit ThreadData(ConcurrentMarkAndSweep& gc, mm::ThreadData& threadData) noexcept : gc_(gc), threadData_(threadData) {}
+        ~ThreadData() = default;
+
+        void OnSuspendForGC() noexcept;
+
+        void safePoint() noexcept { barriers_.onSafePoint(); }
+
+        void onThreadRegistration() noexcept { barriers_.onThreadRegistration(); }
+
+        BarriersThreadData& barriers() noexcept { return barriers_; }
+
+        bool tryLockRootSet();
+        void publish();
+        bool published() const;
+        void clearMarkFlags();
+
+        mm::ThreadData& commonThreadData() const;
+
+    private:
+        friend ConcurrentMarkAndSweep;
+        ConcurrentMarkAndSweep& gc_;
+        mm::ThreadData& threadData_;
+        BarriersThreadData barriers_;
+
+        std::atomic<bool> rootSetLocked_ = false;
+        std::atomic<bool> published_ = false;
+    };
+
+    ConcurrentMarkAndSweep(
+            alloc::Allocator& allocator, gcScheduler::GCScheduler& scheduler, bool mutatorsCooperate, std::size_t auxGCThreads) noexcept;
+    ~ConcurrentMarkAndSweep();
+
+    void StartFinalizerThreadIfNeeded() noexcept;
+    void StopFinalizerThreadIfRunning() noexcept;
+    bool FinalizersThreadIsRunning() noexcept;
+
+    void reconfigure(std::size_t maxParallelism, bool mutatorsCooperate, size_t auxGCThreads) noexcept;
+
+    GCStateHolder& state() noexcept { return state_; }
+
+private:
+    void mainGCThreadBody();
+    void auxiliaryGCThreadBody();
+    void PerformFullGC(int64_t epoch) noexcept;
+
+    alloc::Allocator& allocator_;
+    gcScheduler::GCScheduler& gcScheduler_;
+
+    GCStateHolder state_;
+    FinalizerProcessor<alloc::FinalizerQueue, alloc::FinalizerQueueTraits> finalizerProcessor_;
+
+    mark::ParallelMark markDispatcher_;
+    ScopedThread mainThread_;
+    std::vector<ScopedThread> auxThreads_;
+};
+
+} // namespace gc
+} // namespace kotlin

kotlin-native/runtime/src/gc/cms/cpp/GCImpl.cpp

@@ -0,0 +1,112 @@
+/*
+ * Copyright 2010-2021 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.
+ */
+
+#include "GCImpl.hpp"
+
+#include <memory>
+
+#include "ConcurrentMarkAndSweep.hpp"
+#include "GC.hpp"
+#include "GCStatistics.hpp"
+#include "MarkAndSweepUtils.hpp"
+#include "ObjectOps.hpp"
+
+using namespace kotlin;
+
+gc::GC::ThreadData::ThreadData(GC& gc, mm::ThreadData& threadData) noexcept : impl_(std::make_unique<Impl>(gc, threadData)) {}
+
+gc::GC::ThreadData::~ThreadData() = default;
+
+ALWAYS_INLINE void gc::GC::ThreadData::onAllocation(ObjHeader* object) noexcept {
+    impl().gc().barriers().onAllocation(object);
+}
+
+void gc::GC::ThreadData::OnSuspendForGC() noexcept {
+    impl_->gc().OnSuspendForGC();
+}
+
+void gc::GC::ThreadData::safePoint() noexcept {
+    impl_->gc().safePoint();
+}
+
+void gc::GC::ThreadData::onThreadRegistration() noexcept {
+    impl_->gc().onThreadRegistration();
+}
+
+gc::GC::GC(alloc::Allocator& allocator, gcScheduler::GCScheduler& gcScheduler) noexcept :
+    impl_(std::make_unique<Impl>(allocator, gcScheduler)) {}
+
+gc::GC::~GC() = default;
+
+void gc::GC::ClearForTests() noexcept {
+    impl_->gc().StopFinalizerThreadIfRunning();
+    GCHandle::ClearForTests();
+}
+
+void gc::GC::StartFinalizerThreadIfNeeded() noexcept {
+    impl_->gc().StartFinalizerThreadIfNeeded();
+}
+
+void gc::GC::StopFinalizerThreadIfRunning() noexcept {
+    impl_->gc().StopFinalizerThreadIfRunning();
+}
+
+bool gc::GC::FinalizersThreadIsRunning() noexcept {
+    return impl_->gc().FinalizersThreadIsRunning();
+}
+
+// static
+ALWAYS_INLINE void gc::GC::processObjectInMark(void* state, ObjHeader* object) noexcept {
+    gc::internal::processObjectInMark<gc::mark::ParallelMark::MarkTraits>(state, object);
+}
+
+// static
+ALWAYS_INLINE void gc::GC::processArrayInMark(void* state, ArrayHeader* array) noexcept {
+    gc::internal::processArrayInMark<gc::mark::ParallelMark::MarkTraits>(state, array);
+}
+
+// static
+ALWAYS_INLINE void gc::GC::processFieldInMark(void* state, ObjHeader* field) noexcept {
+    gc::internal::processFieldInMark<gc::mark::ParallelMark::MarkTraits>(state, field);
+}
+
+int64_t gc::GC::Schedule() noexcept {
+    return impl_->gc().state().schedule();
+}
+
+void gc::GC::WaitFinished(int64_t epoch) noexcept {
+    impl_->gc().state().waitEpochFinished(epoch);
+}
+
+void gc::GC::WaitFinalizers(int64_t epoch) noexcept {
+    impl_->gc().state().waitEpochFinalized(epoch);
+}
+
+bool gc::isMarked(ObjHeader* object) noexcept {
+    return alloc::objectDataForObject(object).marked();
+}
+
+ALWAYS_INLINE OBJ_GETTER(gc::tryRef, std::atomic<ObjHeader*>& object) noexcept {
+    RETURN_RESULT_OF(gc::WeakRefRead, object);
+}
+
+ALWAYS_INLINE bool gc::tryResetMark(GC::ObjectData& objectData) noexcept {
+    return objectData.tryResetMark();
+}
+
+// static
+ALWAYS_INLINE uint64_t type_layout::descriptor<gc::GC::ObjectData>::type::size() noexcept {
+    return sizeof(gc::GC::ObjectData);
+}
+
+// static
+ALWAYS_INLINE size_t type_layout::descriptor<gc::GC::ObjectData>::type::alignment() noexcept {
+    return alignof(gc::GC::ObjectData);
+}
+
+// static
+ALWAYS_INLINE gc::GC::ObjectData* type_layout::descriptor<gc::GC::ObjectData>::type::construct(uint8_t* ptr) noexcept {
+    return new (ptr) gc::GC::ObjectData();
+}

kotlin-native/runtime/src/gc/cms/cpp/GCImpl.hpp

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2010-2021 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.
+ */
+
+#pragma once
+
+#include "GC.hpp"
+
+#include "ConcurrentMarkAndSweep.hpp"
+
+namespace kotlin {
+namespace gc {
+
+class GC::Impl : private Pinned {
+public:
+    Impl(alloc::Allocator& allocator, gcScheduler::GCScheduler& gcScheduler) noexcept :
+        gc_(allocator, gcScheduler, compiler::gcMutatorsCooperate(), compiler::auxGCThreads()) {}
+
+    ConcurrentMarkAndSweep& gc() noexcept { return gc_; }
+
+private:
+    ConcurrentMarkAndSweep gc_;
+};
+
+class GC::ThreadData::Impl : private Pinned {
+public:
+    Impl(GC& gc, mm::ThreadData& threadData) noexcept : gc_(gc.impl_->gc(), threadData) {}
+
+    ConcurrentMarkAndSweep::ThreadData& gc() noexcept { return gc_; }
+
+private:
+    ConcurrentMarkAndSweep::ThreadData gc_;
+};
+
+} // namespace gc
+} // namespace kotlin

kotlin-native/runtime/src/gc/cms/cpp/ObjectData.hpp

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2010-2023 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.
+ */
+
+#pragma once
+
+#include <atomic>
+#include <type_traits>
+
+#include "Allocator.hpp"
+#include "GC.hpp"
+#include "IntrusiveList.hpp"
+#include "KAssert.h"
+
+namespace kotlin::gc {
+
+class GC::ObjectData {
+    static constexpr intptr_t kNoQueueMark = 1;
+public:
+    bool tryMark() noexcept { return trySetNext(reinterpret_cast<ObjectData*>(kNoQueueMark)); }
+
+    void markUncontended() noexcept {
+        RuntimeAssert(!marked(), "Must not be marked previously");
+        auto nextVal = reinterpret_cast<ObjectData*>(kNoQueueMark);
+        setNext(nextVal);
+        RuntimeAssert(next() == nextVal, "Non-atomic marking must not be contended");
+    }
+
+    bool marked() const noexcept { return next() != nullptr; }
+
+    bool tryResetMark() noexcept {
+        if (next() == nullptr) return false;
+        next_.store(nullptr, std::memory_order_relaxed);
+        return true;
+    }
+
+private:
+    friend struct DefaultIntrusiveForwardListTraits<ObjectData>;
+
+    ObjectData* next() const noexcept { return next_.load(std::memory_order_relaxed); }
+    void setNext(ObjectData* next) noexcept {
+        RuntimeAssert(next, "next cannot be nullptr");
+        next_.store(next, std::memory_order_relaxed);
+    }
+    bool trySetNext(ObjectData* next) noexcept {
+        RuntimeAssert(next, "next cannot be nullptr");
+        ObjectData* expected = nullptr;
+        return next_.compare_exchange_strong(expected, next, std::memory_order_relaxed);
+    }
+
+    std::atomic<ObjectData*> next_ = nullptr;
+};
+static_assert(std::is_trivially_destructible_v<GC::ObjectData>);
+
+} // namespace kotlin::gc

kotlin-native/runtime/src/gc/cms/cpp/ParallelMark.cpp

@@ -0,0 +1,235 @@
+/*
+ * Copyright 2010-2023 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.
+ */
+
+#include "ParallelMark.hpp"
+
+#include "MarkAndSweepUtils.hpp"
+#include "GCStatistics.hpp"
+#include "Utils.hpp"
+
+// required to access gc thread data
+#include "GCImpl.hpp"
+
+using namespace kotlin;
+
+namespace {
+
+template<typename Cond>
+void spinWait(Cond&& until) {
+    while (!until()) {
+        std::this_thread::yield();
+    }
+}
+
+} // namespace
+
+bool gc::mark::MarkPacer::is(gc::mark::MarkPacer::Phase phase) const {
+    std::unique_lock lock(mutex_);
+    return phase_ == phase;
+}
+
+void gc::mark::MarkPacer::begin(gc::mark::MarkPacer::Phase phase) {
+    {
+        std::unique_lock lock(mutex_);
+        phase_ = phase;
+    }
+    cond_.notify_all();
+}
+
+void gc::mark::MarkPacer::wait(gc::mark::MarkPacer::Phase phase) {
+    std::unique_lock lock(mutex_);
+    cond_.wait(lock, [=]() { return phase_ >= phase; });
+}
+
+void gc::mark::MarkPacer::beginEpoch(uint64_t epoch) {
+    epoch_ = epoch;
+    begin(Phase::kReady);
+    GCLogDebug(epoch_.load(), "Mark is ready to recruit workers in a new epoch.");
+}
+
+void gc::mark::MarkPacer::waitNewEpochReadyOrShutdown() const {
+    std::unique_lock lock(mutex_);
+    cond_.wait(lock, [this]() { return phase_ >= Phase::kReady; });
+}
+
+void gc::mark::MarkPacer::waitEpochFinished(uint64_t currentEpoch) const {
+    std::unique_lock lock(mutex_);
+    cond_.wait(lock, [this, currentEpoch]() {
+        return phase_ == Phase::kIdle || phase_ == Phase::kShutdown || epoch_.load(std::memory_order_relaxed) > currentEpoch;
+    });
+}
+
+bool gc::mark::MarkPacer::acceptingNewWorkers() const {
+    std::unique_lock lock(mutex_);
+    return Phase::kReady <= phase_ && phase_ <= Phase::kParallelMark;
+}
+
+
+gc::mark::ParallelMark::ParallelMark(bool mutatorsCooperate) {
+    std::size_t maxParallelism = std::thread::hardware_concurrency();
+    if (maxParallelism == 0) {
+        maxParallelism = std::numeric_limits<std::size_t>::max();
+    }
+    setParallelismLevel(maxParallelism, mutatorsCooperate);
+}
+
+void gc::mark::ParallelMark::beginMarkingEpoch(gc::GCHandle gcHandle) {
+    gcHandle_ = gcHandle;
+
+    lockedMutatorsList_ = mm::ThreadRegistry::Instance().LockForIter();
+
+    parallelProcessor_.construct();
+
+    if (!compiler::gcMarkSingleThreaded()) {
+        std::unique_lock guard(workerCreationMutex_);
+        pacer_.beginEpoch(gcHandle.getEpoch());
+        // main worker is always accounted, so others would not be able to exhaust all the parallelism before main is instantiated
+        activeWorkersCount_ = 1;
+    }
+}
+
+void gc::mark::ParallelMark::endMarkingEpoch() {
+    if (!compiler::gcMarkSingleThreaded()) {
+        // We must now wait for every worker to finish the Mark procedure:
+        // wake up from possible waiting, publish statistics, etc.
+        // Only then it's safe to destroy the parallelProcessor and proceed to other GC tasks such as sweep.
+        spinWait([=]() { return activeWorkersCount_.load(std::memory_order_relaxed) == 0; });
+
+        std::unique_lock guard(workerCreationMutex_);
+        RuntimeAssert(activeWorkersCount_ == 0, "All the workers must already finish");
+        pacer_.begin(MarkPacer::Phase::kIdle);
+    }
+    parallelProcessor_.destroy();
+    resetMutatorFlags();
+    lockedMutatorsList_ = std::nullopt;
+}
+
+void gc::mark::ParallelMark::runMainInSTW() {
+    if (compiler::gcMarkSingleThreaded()) {
+        ParallelProcessor::Worker worker(*parallelProcessor_);
+        gc::collectRootSet<MarkTraits>(gcHandle(), worker, [] (mm::ThreadData&) { return true; });
+        gc::Mark<MarkTraits>(gcHandle(), worker);
+    } else {
+        RuntimeAssert(activeWorkersCount_ > 0, "Main worker must always be accounted");
+        ParallelProcessor::Worker mainWorker(*parallelProcessor_);
+        GCLogDebug(gcHandle().getEpoch(), "Creating main (#0) mark worker");
+
+        pacer_.begin(MarkPacer::Phase::kRootSet);
+        completeMutatorsRootSet(mainWorker);
+        spinWait([this] {
+            return allMutators([](mm::ThreadData& mut) { return mut.gc().impl().gc().published(); });
+        });
+        // global root set must be collected after all the mutator's global data have been published
+        collectRootSetGlobals<MarkTraits>(gcHandle(), mainWorker);
+
+        pacer_.begin(MarkPacer::Phase::kParallelMark);
+        parallelMark(mainWorker);
+    }
+}
+
+void gc::mark::ParallelMark::runOnMutator(mm::ThreadData& mutatorThread) {
+    if (compiler::gcMarkSingleThreaded() || !mutatorsCooperate_) return;
+
+    auto parallelWorker = createWorker();
+    if (parallelWorker) {
+        GCLogDebug(gcHandle().getEpoch(), "Mutator thread cooperates in marking");
+
+        tryCollectRootSet(mutatorThread, *parallelWorker);
+
+        completeRootSetAndMark(*parallelWorker);
+    }
+}
+
+void gc::mark::ParallelMark::runAuxiliary() {
+    RuntimeAssert(!compiler::gcMarkSingleThreaded(), "Should not reach here during single threaded mark");
+
+    pacer_.waitNewEpochReadyOrShutdown();
+    if (pacer_.is(MarkPacer::Phase::kShutdown)) return;
+
+    auto curEpoch = gcHandle().getEpoch();
+    auto parallelWorker = createWorker();
+    if (parallelWorker) {
+        completeRootSetAndMark(*parallelWorker);
+    }
+
+    pacer_.waitEpochFinished(curEpoch);
+}
+
+void gc::mark::ParallelMark::requestShutdown() {
+    pacer_.begin(MarkPacer::Phase::kShutdown);
+}
+
+bool gc::mark::ParallelMark::shutdownRequested() const {
+    return pacer_.is(MarkPacer::Phase::kShutdown);
+}
+
+
+gc::GCHandle& gc::mark::ParallelMark::gcHandle() {
+    RuntimeAssert(gcHandle_.isValid(), "GCHandle must be initialized");
+    return gcHandle_;
+}
+
+void gc::mark::ParallelMark::setParallelismLevel(size_t maxParallelism, bool mutatorsCooperate) {
+    RuntimeCheck(maxParallelism > 0, "Parallelism level can't be 0");
+    maxParallelism_ = maxParallelism;
+    mutatorsCooperate_ = mutatorsCooperate;
+    RuntimeLogInfo({kTagGC},
+                   "Set up parallel mark with maxParallelism = %zu and %s" "cooperative mutators",
+                   maxParallelism_, (mutatorsCooperate_ ? "" : "non-"));
+}
+
+void gc::mark::ParallelMark::completeRootSetAndMark(ParallelProcessor::Worker& parallelWorker) {
+    pacer_.wait(MarkPacer::Phase::kRootSet);
+    completeMutatorsRootSet(parallelWorker);
+    pacer_.wait(MarkPacer::Phase::kParallelMark);
+    parallelMark(parallelWorker);
+}
+
+void gc::mark::ParallelMark::completeMutatorsRootSet(MarkTraits::MarkQueue& markQueue) {
+    // workers compete for mutators to collect their root set
+    for (auto& thread: *lockedMutatorsList_) {
+        tryCollectRootSet(thread, markQueue);
+    }
+}
+
+void gc::mark::ParallelMark::tryCollectRootSet(mm::ThreadData& thread, MarkTraits::MarkQueue& markQueue) {
+    auto& gcData = thread.gc().impl().gc();
+    if (!gcData.tryLockRootSet()) return;
+
+    GCLogDebug(gcHandle().getEpoch(), "Root set collection on thread %d for thread %d",
+               konan::currentThreadId(), thread.threadId());
+    gcData.publish();
+    collectRootSetForThread<MarkTraits>(gcHandle(), markQueue, thread);
+}
+
+void gc::mark::ParallelMark::parallelMark(ParallelProcessor::Worker& worker) {
+    GCLogDebug(gcHandle().getEpoch(), "Mark loop has begun");
+    Mark<MarkTraits>(gcHandle(), worker);
+
+    std::unique_lock guard(workerCreationMutex_);
+    activeWorkersCount_.fetch_sub(1, std::memory_order_relaxed);
+}
+
+std::optional<gc::mark::ParallelMark::ParallelProcessor::Worker> gc::mark::ParallelMark::createWorker() {
+    std::unique_lock guard(workerCreationMutex_);
+    if (!pacer_.acceptingNewWorkers() ||
+        activeWorkersCount_.load(std::memory_order_relaxed) >= maxParallelism_ ||
+        activeWorkersCount_.load(std::memory_order_relaxed) == 0) return std::nullopt;
+
+    auto num = activeWorkersCount_.fetch_add(1, std::memory_order_relaxed);
+    GCLogDebug(gcHandle().getEpoch(), "Creating mark worker #%zu", num);
+    return std::make_optional<ParallelProcessor::Worker>(*parallelProcessor_);
+}
+
+void gc::mark::ParallelMark::resetMutatorFlags() {
+    for (auto& mut: *lockedMutatorsList_) {
+        auto& gcData = mut.gc().impl().gc();
+        if (!compiler::gcMarkSingleThreaded()) {
+            // single threaded mark do not use this flag
+            RuntimeAssert(gcData.published(), "Must have been published during mark");
+        }
+        gcData.clearMarkFlags();
+    }
+}

kotlin-native/runtime/src/gc/cms/cpp/ParallelMark.hpp

@@ -0,0 +1,182 @@
+/*
+ * Copyright 2010-2023 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.
+ */
+
+#pragma once
+
+#include <condition_variable>
+#include <mutex>
+
+#include "GCStatistics.hpp"
+#include "ManuallyScoped.hpp"
+#include "ObjectData.hpp"
+#include "ParallelProcessor.hpp"
+#include "ThreadRegistry.hpp"
+#include "Utils.hpp"
+
+namespace kotlin::gc::mark {
+
+class MarkPacer : private Pinned {
+public:
+    enum class Phase {
+        /** Mark is not in progress. */
+        kIdle,
+        /**
+         * MarkDispatcher is ready to recruit new workers.
+         *
+         * In case of cooperative mark mutator threads are welcome to mark their own root sets.
+         * Each thread is free to start as soon as it reaches a safe point.
+         * No need to wait for others.
+         */
+        kReady,
+        /**
+         * All mutator threads must be in a safe state at this point:
+         * 1) Suspended on a safe point;
+         * 2) In the native code;
+         * 3) Registered as cooperative markers during previous phase.
+         *
+         * Now all the GC workers are summoned to participate in a root set collection.
+         */
+        kRootSet,
+        /**
+         * Root set is collected. No more workers can be instantiated, time to begin parallel mark.
+         * Parallel mark can't stop before all the created workers begin the marking.
+         */
+        kParallelMark,
+        /** A shutdown was requested. There is nothing more to wait for. */
+        kShutdown,
+    };
+
+    bool is(Phase phase) const;
+    void begin(Phase phase);
+    void wait(Phase phase);
+
+    void beginEpoch(uint64_t epoch);
+    void waitNewEpochReadyOrShutdown() const;
+    void waitEpochFinished(uint64_t epoch) const;
+
+    bool acceptingNewWorkers() const;
+
+private:
+    std::atomic<uint64_t> epoch_ = 0;
+    Phase phase_ = Phase::kIdle;
+    mutable std::mutex mutex_;
+    mutable std::condition_variable cond_;
+};
+
+/**
+ * Parallel mark dispatcher.
+ * Mark can be performed on one or more threads.
+ * Each threads wanting to participate have to execute an appropriate run- routine when ready to mark.
+ * There must be exactly one executor of a `runMainInSTW()`.
+ *
+ * Mark workers are able to balance work between each other through sharing/stealing.
+ */
+class ParallelMark : private Pinned {
+    using MarkStackImpl = intrusive_forward_list<GC::ObjectData>;
+    // work balancing parameters were chosen pretty arbitrary
+    using ParallelProcessor = ParallelProcessor<MarkStackImpl, 512, 4096>;
+public:
+    class MarkTraits {
+    public:
+        using MarkQueue = ParallelProcessor::Worker;
+
+        static void clear(MarkQueue& queue) noexcept {
+            RuntimeAssert(queue.localEmpty(), "Mark queue must be empty");
+        }
+
+        static ALWAYS_INLINE ObjHeader* tryDequeue(MarkQueue& queue) noexcept {
+            auto* obj = compiler::gcMarkSingleThreaded() ? queue.tryPopLocal() : queue.tryPop();
+            if (obj) {
+                return alloc::objectForObjectData(*obj);
+            }
+            return nullptr;
+        }
+
+        static ALWAYS_INLINE bool tryEnqueue(MarkQueue& queue, ObjHeader* object) noexcept {
+            auto& objectData = alloc::objectDataForObject(object);
+            return compiler::gcMarkSingleThreaded() ? queue.tryPushLocal(objectData) : queue.tryPush(objectData);
+        }
+
+        static ALWAYS_INLINE bool tryMark(ObjHeader* object) noexcept {
+            auto& objectData = alloc::objectDataForObject(object);
+            return objectData.tryMark();
+        }
+
+        static ALWAYS_INLINE void processInMark(MarkQueue& markQueue, ObjHeader* object) noexcept {
+            auto process = object->type_info()->processObjectInMark;
+            RuntimeAssert(process != nullptr, "Got null processObjectInMark for object %p", object);
+            process(static_cast<void*>(&markQueue), object);
+        }
+    };
+
+    ParallelMark(bool mutatorsCooperate);
+
+    void beginMarkingEpoch(gc::GCHandle gcHandle);
+    void endMarkingEpoch();
+
+    /** To be run by a single "main" GC thread during STW. */
+    void runMainInSTW();
+
+    /**
+     * To be run by mutator threads that would like to participate in mark.
+     * Will wait for STW detection by a "main" routine.
+     */
+    void runOnMutator(mm::ThreadData& mutatorThread);
+
+    /**
+     * To be run by auxiliary GC threads.
+     * Will wait for STW detection by a "main" routine.
+     */
+    void runAuxiliary();
+
+    void requestShutdown();
+    bool shutdownRequested() const;
+
+    template<typename Pred>
+    void reset(std::size_t maxParallelism, bool mutatorsCooperate, Pred waitForWorkersToFinish) {
+        pacer_.begin(MarkPacer::Phase::kShutdown);
+        waitForWorkersToFinish();
+        pacer_.begin(MarkPacer::Phase::kIdle);
+        setParallelismLevel(maxParallelism, mutatorsCooperate);
+    }
+
+private:
+    GCHandle& gcHandle();
+
+    void setParallelismLevel(size_t maxParallelism, bool mutatorsCooperate);
+
+    template <typename Pred>
+    bool allMutators(Pred predicate) noexcept {
+        for (auto& thread : *lockedMutatorsList_) {
+            if (!predicate(thread)) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    void completeRootSetAndMark(ParallelProcessor::Worker& parallelWorker);
+    void completeMutatorsRootSet(MarkTraits::MarkQueue& markQueue);
+    void tryCollectRootSet(mm::ThreadData& thread, ParallelProcessor::Worker& markQueue);
+    void parallelMark(ParallelProcessor::Worker& worker);
+
+    std::optional<ParallelProcessor::Worker> createWorker();
+
+    void resetMutatorFlags();
+
+    std::size_t maxParallelism_ = 1;
+    bool mutatorsCooperate_ = false;
+
+    GCHandle gcHandle_ = GCHandle::invalid();
+    MarkPacer pacer_;
+    std::optional<mm::ThreadRegistry::Iterable> lockedMutatorsList_;
+    ManuallyScoped<ParallelProcessor> parallelProcessor_{};
+
+    std::mutex workerCreationMutex_;
+    std::atomic<std::size_t> activeWorkersCount_ = 0;
+};
+
+} // namespace kotlin::gc::mark
+