[K/N] GC: parallel mark with work balancing ^KT-57771

Merge-request: KT-MR-11460
Merged-by: Alexey Glushko <aleksei.glushko@jetbrains.com>

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

@@ -41,6 +41,10 @@ object BinaryOptions : BinaryOptionRegistry() {
 
     val concurrentWeakSweep by booleanOption()
 
+    val gcMutatorsCooperate by booleanOption()
+
+    val auxGCThreads by uintOption()
+
     val linkRuntime by option<RuntimeLinkageStrategyBinaryOption>()
 
     val bundleId by stringOption()
@@ -95,6 +99,15 @@ open class BinaryOptionRegistry {
                 }
             }
 
+    protected fun uintOption(): PropertyDelegateProvider<Any?, ReadOnlyProperty<Any?, CompilerConfigurationKey<UInt>>> =
+            PropertyDelegateProvider { _, property ->
+                val option = BinaryOption(property.name, UIntValueParser)
+                register(option)
+                ReadOnlyProperty { _, _ ->
+                    option.compilerConfigurationKey
+                }
+            }
+
     protected fun stringOption(): PropertyDelegateProvider<Any?, ReadOnlyProperty<Any?, CompilerConfigurationKey<String>>> =
             PropertyDelegateProvider { _, property ->
                 val option = BinaryOption(property.name, StringValueParser)
@@ -121,6 +134,13 @@ private object BooleanValueParser : BinaryOption.ValueParser<Boolean> {
         get() = "true|false"
 }
 
+private object UIntValueParser : BinaryOption.ValueParser<UInt> {
+    override fun parse(value: String): UInt? = value.toUIntOrNull()
+
+    override val validValuesHint: String?
+        get() = "non-negative-number"
+}
+
 private object StringValueParser : BinaryOption.ValueParser<String> {
     override fun parse(value: String) = value
     override val validValuesHint: String?

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

@@ -143,12 +143,41 @@ class KonanConfig(val project: Project, val configuration: CompilerConfiguration
         } ?: arg
     }
 
-    val gcMarkSingleThreaded: Boolean
-        get() = configuration.get(BinaryOptions.gcMarkSingleThreaded) ?: false
+    private val defaultGcMarkSingleThreaded get() = target.family == Family.MINGW
+
+    val gcMarkSingleThreaded: Boolean by lazy {
+        configuration.get(BinaryOptions.gcMarkSingleThreaded) ?: defaultGcMarkSingleThreaded
+    }
 
     val concurrentWeakSweep: Boolean
         get() = configuration.get(BinaryOptions.concurrentWeakSweep) ?: false
 
+    val gcMutatorsCooperate: Boolean by lazy {
+        val mutatorsCooperate = configuration.get(BinaryOptions.gcMutatorsCooperate)
+        if (gcMarkSingleThreaded) {
+            if (mutatorsCooperate == true) {
+                configuration.report(CompilerMessageSeverity.STRONG_WARNING,
+                        "Mutators cooperation is not supported during single threaded mark")
+            }
+            false
+        } else {
+            mutatorsCooperate ?: true
+        }
+    }
+
+    val auxGCThreads: UInt by lazy {
+        val auxGCThreads = configuration.get(BinaryOptions.auxGCThreads)
+        if (gcMarkSingleThreaded) {
+            if (auxGCThreads != null && auxGCThreads != 0U) {
+                configuration.report(CompilerMessageSeverity.STRONG_WARNING,
+                        "Auxiliary GC workers are not supported during single threaded mark")
+            }
+            0U
+        } else {
+            auxGCThreads ?: 0U
+        }
+    }
+
     val irVerificationMode: IrVerificationMode
         get() = configuration.getNotNull(KonanConfigKeys.VERIFY_IR)
 
@@ -408,6 +437,8 @@ class KonanConfig(val project: Project, val configuration: CompilerConfiguration
             append("-runtime_asserts=${runtimeAssertsMode.name}")
         if (disableMmap != defaultDisableMmap)
             append("-disable_mmap=${disableMmap}")
+        if (gcMarkSingleThreaded != defaultGcMarkSingleThreaded)
+            append("-gc_mark_single_threaded=${gcMarkSingleThreaded}")
     }
 
     private val userCacheFlavorString = buildString {

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

@@ -2780,7 +2780,8 @@ internal class CodeGeneratorVisitor(
             return
 
         overrideRuntimeGlobal("Kotlin_destroyRuntimeMode", llvm.constInt32(context.config.destroyRuntimeMode.value))
-        overrideRuntimeGlobal("Kotlin_gcMarkSingleThreaded", llvm.constInt32(if (context.config.gcMarkSingleThreaded) 1 else 0))
+        overrideRuntimeGlobal("Kotlin_gcMutatorsCooperate", llvm.constInt32(if (context.config.gcMutatorsCooperate) 1 else 0))
+        overrideRuntimeGlobal("Kotlin_auxGCThreads", llvm.constInt32(context.config.auxGCThreads.toInt()))
         overrideRuntimeGlobal("Kotlin_workerExceptionHandling", llvm.constInt32(context.config.workerExceptionHandling.value))
         overrideRuntimeGlobal("Kotlin_suspendFunctionsFromAnyThreadFromObjC", llvm.constInt32(if (context.config.suspendFunctionsFromAnyThreadFromObjC) 1 else 0))
         val getSourceInfoFunctionName = when (context.config.sourceInfoType) {
@@ -3015,6 +3016,7 @@ internal fun NativeGenerationState.generateRuntimeConstantsModule() : LLVMModule
     setRuntimeConstGlobal("Kotlin_freezingEnabled", llvm.constInt32(if (config.freezing.enableFreezeAtRuntime) 1 else 0))
     setRuntimeConstGlobal("Kotlin_freezingChecksEnabled", llvm.constInt32(if (config.freezing.enableFreezeChecks) 1 else 0))
     setRuntimeConstGlobal("Kotlin_concurrentWeakSweep", llvm.constInt32(if (context.config.concurrentWeakSweep) 1 else 0))
+    setRuntimeConstGlobal("Kotlin_gcMarkSingleThreaded", llvm.constInt32(if (config.gcMarkSingleThreaded) 1 else 0))
 
     return llvmModule
 }

kotlin-native/backend.native/tests/runtime/atomics/atomic_stress.kt

@@ -86,7 +86,12 @@ fun testAtomicReferenceStress(workers: Array<Worker>) {
 fun runStressTest() {
     val COUNT = 20
     val workers = Array(COUNT, { _ -> Worker.start()})
+
     testAtomicIntStress(workers)
     testAtomicLongStress(workers)
     testAtomicReferenceStress(workers)
+
+    workers.forEach {
+        it.requestTermination().result
+    }
 }

kotlin-native/runtime/src/custom_alloc/cpp/Heap.cpp

@@ -50,18 +50,22 @@ FinalizerQueue Heap::Sweep(gc::GCHandle gcHandle) noexcept {
         auto sweepHandle = gcHandle.sweepExtraObjects();
         extraObjectPages_.Sweep(sweepHandle, finalizerQueue);
     }
+    // wait for concurrent assistants to finish sweeping the last popped page
+    while (concurrentSweepersCount_.load(std::memory_order_acquire) > 0) {
+        std::this_thread::yield();
+    }
     CustomAllocDebug("Heap::Sweep done");
     return finalizerQueue;
 }
 
 NextFitPage* Heap::GetNextFitPage(uint32_t cellCount, FinalizerQueue& finalizerQueue) noexcept {
     CustomAllocDebug("Heap::GetNextFitPage()");
-    return nextFitPages_.GetPage(cellCount, finalizerQueue);
+    return nextFitPages_.GetPage(cellCount, finalizerQueue, concurrentSweepersCount_);
 }
 
 FixedBlockPage* Heap::GetFixedBlockPage(uint32_t cellCount, FinalizerQueue& finalizerQueue) noexcept {
     CustomAllocDebug("Heap::GetFixedBlockPage()");
-    return fixedBlockPages_[cellCount].GetPage(cellCount, finalizerQueue);
+    return fixedBlockPages_[cellCount].GetPage(cellCount, finalizerQueue, concurrentSweepersCount_);
 }
 
 SingleObjectPage* Heap::GetSingleObjectPage(uint64_t cellCount, FinalizerQueue& finalizerQueue) noexcept {
@@ -71,7 +75,7 @@ SingleObjectPage* Heap::GetSingleObjectPage(uint64_t cellCount, FinalizerQueue&
 
 ExtraObjectPage* Heap::GetExtraObjectPage(FinalizerQueue& finalizerQueue) noexcept {
     CustomAllocInfo("CustomAllocator::GetExtraObjectPage()");
-    return extraObjectPages_.GetPage(0, finalizerQueue);
+    return extraObjectPages_.GetPage(0, finalizerQueue, concurrentSweepersCount_);
 }
 
 std_support::vector<ObjHeader*> Heap::GetAllocatedObjects() noexcept {

kotlin-native/runtime/src/custom_alloc/cpp/Heap.hpp

@@ -45,6 +45,8 @@ private:
     PageStore<NextFitPage> nextFitPages_;
     PageStore<SingleObjectPage> singleObjectPages_;
     PageStore<ExtraObjectPage> extraObjectPages_;
+
+    std::atomic<std::size_t> concurrentSweepersCount_ = 0;
 };
 
 } // namespace kotlin::alloc

kotlin-native/runtime/src/custom_alloc/cpp/PageStore.hpp

@@ -44,18 +44,25 @@ public:
         }
     }
 
-    T* GetPage(uint32_t cellCount, FinalizerQueue& finalizerQueue) noexcept {
+    T* GetPage(uint32_t cellCount, FinalizerQueue& finalizerQueue, std::atomic<std::size_t>& concurrentSweepersCount_) noexcept {
         T* page;
         if ((page = ready_.Pop())) {
             used_.Push(page);
             return page;
         }
-        auto handle = gc::GCHandle::currentEpoch();
-        if ((page = unswept_.Pop())) {
-            // If there're unswept_ pages, the GC is in progress.
-            GCSweepScope sweepHandle = T::currentGCSweepScope(*handle);
-            if ((page = SweepSingle(sweepHandle, page, unswept_, used_, finalizerQueue))) {
-                return page;
+        {
+            auto handle = gc::GCHandle::currentEpoch();
+            ScopeGuard counterGuard(
+                    [&]() { ++concurrentSweepersCount_; },
+                    [&]() { --concurrentSweepersCount_; }
+            );
+
+            if ((page = unswept_.Pop())) {
+                // If there're unswept_ pages, the GC is in progress.
+                GCSweepScope sweepHandle = T::currentGCSweepScope(*handle);
+                if ((page = SweepSingle(sweepHandle, page, unswept_, used_, finalizerQueue))) {
+                    return page;
+                }
             }
         }
         if ((page = empty_.Pop())) {

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

@@ -16,7 +16,6 @@
 #include "MarkAndSweepUtils.hpp"
 #include "Memory.h"
 #include "ThreadData.hpp"
-#include "ThreadRegistry.hpp"
 #include "ThreadSuspension.hpp"
 #include "GCState.hpp"
 #include "GCStatistics.hpp"
@@ -28,10 +27,8 @@
 using namespace kotlin;
 
 namespace {
-    [[clang::no_destroy]] std::mutex markingMutex;
-    [[clang::no_destroy]] std::condition_variable markingCondVar;
-    [[clang::no_destroy]] std::atomic<bool> markingRequested = false;
-    [[clang::no_destroy]] std::atomic<uint64_t> markingEpoch = 0;
+
+[[clang::no_destroy]] std::mutex gcMutex;
 
 struct SweepTraits {
     using ObjectFactory = mm::ObjectFactory<gc::ConcurrentMarkAndSweep>;
@@ -57,6 +54,33 @@ struct ProcessWeaksTraits {
     }
 };
 
+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());
+    });
+}
+
+// TODO move to common
+[[maybe_unused]] inline void checkMarkCorrectness(mm::ObjectFactory<gc::ConcurrentMarkAndSweep>::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 =
+                            mm::ObjectFactory<gc::ConcurrentMarkAndSweep>::NodeRef::From(field).ObjectData();
+                    RuntimeAssert(fieldObjData.marked(), "Field %p of an alive obj %p must be alive", field, obj);
+                }
+            });
+        }
+    }
+}
+
 } // namespace
 
 void gc::ConcurrentMarkAndSweep::ThreadData::OnOOM(size_t size) noexcept {
@@ -68,48 +92,70 @@ void gc::ConcurrentMarkAndSweep::ThreadData::OnOOM(size_t size) noexcept {
 void gc::ConcurrentMarkAndSweep::ThreadData::OnSuspendForGC() noexcept {
     CallsCheckerIgnoreGuard guard;
 
-    std::unique_lock lock(markingMutex);
-    if (!markingRequested.load()) return;
-    AutoReset scopedAssignMarking(&marking_, true);
+    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::beginCooperation() {
+    cooperative_.store(true, std::memory_order_release);
+}
+
+bool gc::ConcurrentMarkAndSweep::ThreadData::cooperative() const {
+    return cooperative_.load(std::memory_order_relaxed);
+}
+
+void gc::ConcurrentMarkAndSweep::ThreadData::publish() {
     threadData_.Publish();
-    markingCondVar.wait(lock, []() { return !markingRequested.load(); });
-    // // Unlock while marking to allow mutliple threads to mark in parallel.
-    lock.unlock();
-    uint64_t epoch = markingEpoch.load();
-    GCLogDebug(epoch, "Parallel marking in thread %d", konan::currentThreadId());
-    MarkQueue markQueue;
-    auto handle = GCHandle::getByEpoch(epoch);
-    gc::collectRootSetForThread<internal::MarkTraits>(handle, markQueue, threadData_);
-    gc::Mark<internal::MarkTraits>(handle, markQueue);
+    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);
+    cooperative_.store(false, std::memory_order_relaxed);
+    rootSetLocked_.store(false, std::memory_order_release);
+}
+
+mm::ThreadData& gc::ConcurrentMarkAndSweep::ThreadData::commonThreadData() const {
+    return threadData_;
 }
 
 #ifndef CUSTOM_ALLOCATOR
 gc::ConcurrentMarkAndSweep::ConcurrentMarkAndSweep(
         mm::ObjectFactory<ConcurrentMarkAndSweep>& objectFactory,
         mm::ExtraObjectDataFactory& extraObjectDataFactory,
-        gcScheduler::GCScheduler& gcScheduler) noexcept :
+        gcScheduler::GCScheduler& gcScheduler,
+        bool mutatorsCooperate, std::size_t auxGCThreads) noexcept :
     objectFactory_(objectFactory),
     extraObjectDataFactory_(extraObjectDataFactory),
 #else
-gc::ConcurrentMarkAndSweep::ConcurrentMarkAndSweep(gcScheduler::GCScheduler& gcScheduler) noexcept :
+gc::ConcurrentMarkAndSweep::ConcurrentMarkAndSweep(
+        gcScheduler::GCScheduler& gcScheduler,
+        bool mutatorsCooperate, std::size_t auxGCThreads) noexcept :
 #endif
     gcScheduler_(gcScheduler),
     finalizerProcessor_([this](int64_t epoch) {
         GCHandle::getByEpoch(epoch).finalizersDone();
         state_.finalized(epoch);
-    }) {
-    gcThread_ = ScopedThread(ScopedThread::attributes().name("GC thread"), [this] {
-        while (true) {
-            auto epoch = state_.waitScheduled();
-            if (epoch.has_value()) {
-                PerformFullGC(*epoch);
-            } else {
-                break;
-            }
-        }
-    });
-    markingBehavior_ = kotlin::compiler::gcMarkSingleThreaded() ? MarkingBehavior::kDoNotMark : MarkingBehavior::kMarkOwnStack;
-    RuntimeLogDebug({kTagGC}, "Concurrent Mark & Sweep GC initialized");
+    }),
+    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() {
@@ -131,18 +177,42 @@ bool gc::ConcurrentMarkAndSweep::FinalizersThreadIsRunning() noexcept {
     return finalizerProcessor_.IsRunning();
 }
 
-void gc::ConcurrentMarkAndSweep::SetMarkingBehaviorForTests(MarkingBehavior markingBehavior) noexcept {
-    markingBehavior_ = markingBehavior;
+void gc::ConcurrentMarkAndSweep::mainGCThreadBody() {
+    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);
-    SetMarkingRequested(epoch);
+
+    markDispatcher_.beginMarkingEpoch(gcHandle);
+    GCLogDebug(epoch, "Main GC requested marking in mutators");
+
+    // Request STW
     bool didSuspend = mm::RequestThreadsSuspension();
     RuntimeAssert(didSuspend, "Only GC thread can request suspension");
     gcHandle.suspensionRequested();
 
-    WaitForThreadsReadyToMark();
+    // TODO (WaitForThreadsReadyToMark())
+    RuntimeAssert(!kotlin::mm::IsCurrentThreadRegistered(), "GC must run on unregistered thread");
+
+    markDispatcher_.waitForThreadsPauseMutation();
+    GCLogDebug(epoch, "All threads have paused mutation");
     gcHandle.threadsAreSuspended();
 
 #ifdef CUSTOM_ALLOCATOR
@@ -158,10 +228,9 @@ void gc::ConcurrentMarkAndSweep::PerformFullGC(int64_t epoch) noexcept {
 
     state_.start(epoch);
 
-    CollectRootSetAndStartMarking(gcHandle);
+    markDispatcher_.runMainInSTW();
 
-    // Can be unsafe, because we've stopped the world.
-    gc::Mark<internal::MarkTraits>(gcHandle, markQueue_);
+    markDispatcher_.endMarkingEpoch();
 
     mm::WaitForThreadsSuspension();
 
@@ -170,6 +239,7 @@ void gc::ConcurrentMarkAndSweep::PerformFullGC(int64_t epoch) noexcept {
     // would not publish into the global state at an unexpected time.
     std::optional extraObjectFactoryIterable = extraObjectDataFactory_.LockForIter();
     std::optional objectFactoryIterable = objectFactory_.LockForIter();
+    checkMarkCorrectness(*objectFactoryIterable);
 #endif
 
     if (compiler::concurrentWeakSweep()) {
@@ -214,28 +284,14 @@ void gc::ConcurrentMarkAndSweep::PerformFullGC(int64_t epoch) noexcept {
     finalizerProcessor_.ScheduleTasks(std::move(finalizerQueue), epoch);
 }
 
-void gc::ConcurrentMarkAndSweep::SetMarkingRequested(uint64_t epoch) noexcept {
-    markingRequested = markingBehavior_ == MarkingBehavior::kMarkOwnStack;
-    markingEpoch = epoch;
-}
-
-void gc::ConcurrentMarkAndSweep::WaitForThreadsReadyToMark() noexcept {
-    RuntimeAssert(!kotlin::mm::IsCurrentThreadRegistered(), "GC must run on unregistered thread");
-    mm::ThreadRegistry::Instance().waitAllThreads([](mm::ThreadData& thread) noexcept {
-        return thread.suspensionData().suspendedOrNative() || thread.gc().impl().gc().marking_.load();
-    });
-}
-
-void gc::ConcurrentMarkAndSweep::CollectRootSetAndStartMarking(GCHandle gcHandle) noexcept {
-        std::unique_lock lock(markingMutex);
-        markingRequested = false;
-        gc::collectRootSet<internal::MarkTraits>(
-                gcHandle,
-                markQueue_,
-                [](mm::ThreadData& thread) {
-                    return !thread.gc().impl().gc().marking_.load();
-                }
-            );
-        RuntimeLogDebug({kTagGC}, "Requesting marking in threads");
-        markingCondVar.notify_all();
-}
+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

@@ -23,6 +23,8 @@
 #include "Types.h"
 #include "Utils.hpp"
 #include "std_support/Memory.hpp"
+#include "MarkStack.hpp"
+#include "ParallelMark.hpp"
 
 #ifdef CUSTOM_ALLOCATOR
 #include "CustomAllocator.hpp"
@@ -37,48 +39,15 @@ namespace gc {
 // TODO: Also make marking run concurrently with Kotlin threads.
 class ConcurrentMarkAndSweep : private Pinned {
 public:
-    class ObjectData {
-    public:
-        bool tryMark() noexcept {
-            return trySetNext(reinterpret_cast<ObjectData*>(1));
-        }
-
-        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;
-    };
-
-    enum MarkingBehavior { kMarkOwnStack, kDoNotMark };
-
-    using MarkQueue = intrusive_forward_list<ObjectData>;
 
     class ThreadData : private Pinned {
     public:
-        using ObjectData = ConcurrentMarkAndSweep::ObjectData;
-
+        using ObjectData = mark::ObjectData;
         using Allocator = AllocatorWithGC<Allocator, ThreadData>;
 
-        explicit ThreadData(ConcurrentMarkAndSweep& gc, mm::ThreadData& threadData) noexcept : threadData_(threadData) {}
+        explicit ThreadData(ConcurrentMarkAndSweep& gc, mm::ThreadData& threadData) noexcept
+            : gc_(gc), threadData_(threadData) {}
+
         ~ThreadData() = default;
 
         void OnOOM(size_t size) noexcept;
@@ -91,13 +60,27 @@ public:
 
         BarriersThreadData& barriers() noexcept { return barriers_; }
 
+        bool tryLockRootSet();
+        void beginCooperation();
+        bool cooperative() const;
+        void publish();
+        bool published() const;
+        void clearMarkFlags();
+
+        mm::ThreadData& commonThreadData() const;
+
     private:
         friend ConcurrentMarkAndSweep;
+        ConcurrentMarkAndSweep& gc_;
         mm::ThreadData& threadData_;
-        std::atomic<bool> marking_;
         BarriersThreadData barriers_;
+
+        std::atomic<bool> rootSetLocked_ = false;
+        std::atomic<bool> published_ = false;
+        std::atomic<bool> cooperative_ = false;
     };
 
+    using ObjectData = ThreadData::ObjectData;
     using Allocator = ThreadData::Allocator;
 
 #ifndef CUSTOM_ALLOCATOR
@@ -109,22 +92,22 @@ public:
 #endif
 
 #ifdef CUSTOM_ALLOCATOR
-    explicit ConcurrentMarkAndSweep(gcScheduler::GCScheduler& scheduler) noexcept;
+    explicit ConcurrentMarkAndSweep(gcScheduler::GCScheduler& scheduler,
+                                    bool mutatorsCooperate, std::size_t auxGCThreads) noexcept;
 #else
     ConcurrentMarkAndSweep(
             mm::ObjectFactory<ConcurrentMarkAndSweep>& objectFactory,
             mm::ExtraObjectDataFactory& extraObjectDataFactory,
-            gcScheduler::GCScheduler& scheduler) noexcept;
+            gcScheduler::GCScheduler& scheduler,
+            bool mutatorsCooperate, std::size_t auxGCThreads) noexcept;
 #endif
     ~ConcurrentMarkAndSweep();
 
     void StartFinalizerThreadIfNeeded() noexcept;
     void StopFinalizerThreadIfRunning() noexcept;
     bool FinalizersThreadIsRunning() noexcept;
-    void SetMarkingBehaviorForTests(MarkingBehavior markingBehavior) noexcept;
-    void SetMarkingRequested(uint64_t epoch) noexcept;
-    void WaitForThreadsReadyToMark() noexcept;
-    void CollectRootSetAndStartMarking(GCHandle gcHandle) noexcept;
+
+    void reconfigure(std::size_t maxParallelism, bool mutatorsCooperate, size_t auxGCThreads) noexcept;
 
 #ifdef CUSTOM_ALLOCATOR
     alloc::Heap& heap() noexcept { return heap_; }
@@ -133,6 +116,8 @@ public:
     GCStateHolder& state() noexcept { return state_; }
 
 private:
+    void mainGCThreadBody();
+    void auxiliaryGCThreadBody();
     void PerformFullGC(int64_t epoch) noexcept;
 
 #ifndef CUSTOM_ALLOCATOR
@@ -144,44 +129,12 @@ private:
     gcScheduler::GCScheduler& gcScheduler_;
 
     GCStateHolder state_;
-    ScopedThread gcThread_;
     FinalizerProcessor<FinalizerQueue, FinalizerQueueTraits> finalizerProcessor_;
 
-    MarkQueue markQueue_;
-    MarkingBehavior markingBehavior_;
+    mark::ParallelMark markDispatcher_;
+    ScopedThread mainThread_;
+    std_support::vector<ScopedThread> auxThreads_;
 };
 
-namespace internal {
-struct MarkTraits {
-    using MarkQueue = gc::ConcurrentMarkAndSweep::MarkQueue;
-
-    static void clear(MarkQueue& queue) noexcept { queue.clear(); }
-
-    static ObjHeader* tryDequeue(MarkQueue& queue) noexcept {
-        if (auto* top = queue.try_pop_front()) {
-            auto node = mm::ObjectFactory<gc::ConcurrentMarkAndSweep>::NodeRef::From(*top);
-            return node->GetObjHeader();
-        }
-        return nullptr;
-    }
-
-    static bool tryEnqueue(MarkQueue& queue, ObjHeader* object) noexcept {
-        auto& objectData = mm::ObjectFactory<gc::ConcurrentMarkAndSweep>::NodeRef::From(object).ObjectData();
-        return queue.try_push_front(objectData);
-    }
-
-    static bool tryMark(ObjHeader* object) noexcept {
-        auto& objectData = mm::ObjectFactory<gc::ConcurrentMarkAndSweep>::NodeRef::From(object).ObjectData();
-        return objectData.tryMark();
-    }
-
-    static 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);
-    }
-};
-} // namespace internal
-
 } // namespace gc
 } // namespace kotlin

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

@@ -209,11 +209,27 @@ test_support::RegularWeakReferenceImpl& InstallWeakReference(mm::ThreadData& thr
     return weakReference;
 }
 
-class ConcurrentMarkAndSweepTest : public testing::TestWithParam<gc::ConcurrentMarkAndSweep::MarkingBehavior> {
+struct ParallelismOptions {
+    std::size_t maxParallelism;
+    bool cooperativeMutators;
+    std::size_t auxGCThreads;
+};
+
+class ConcurrentMarkAndSweepTest : public testing::TestWithParam<ParallelismOptions> {
 public:
 
     ConcurrentMarkAndSweepTest() {
-        mm::GlobalData::Instance().gc().impl().gc().SetMarkingBehaviorForTests(GetParam());
+        if (supportedConfiguration()) {
+            mm::GlobalData::Instance().gc().impl().gc().reconfigure(GetParam().maxParallelism,
+                                                                    GetParam().cooperativeMutators,
+                                                                    GetParam().auxGCThreads);
+        }
+    }
+
+    void SetUp() override {
+        if (!supportedConfiguration()) {
+            GTEST_SKIP() << "Unsupported parallelism configuration";
+        }
     }
 
     ~ConcurrentMarkAndSweepTest() {
@@ -225,6 +241,10 @@ public:
     testing::MockFunction<void(ObjHeader*)>& finalizerHook() { return finalizerHooks_.finalizerHook(); }
 
 private:
+    bool supportedConfiguration() const {
+        return !compiler::gcMarkSingleThreaded() || (!GetParam().cooperativeMutators && GetParam().auxGCThreads == 0);
+    }
+
     FinalizerHooksTestSupport finalizerHooks_;
 };
 
@@ -1172,60 +1192,23 @@ TEST_P(ConcurrentMarkAndSweepTest, FreeObjectWithFreeWeakReversedOrder) {
     f1.wait();
 }
 
-TEST_P(ConcurrentMarkAndSweepTest, MutatorsCanMarkOwnLocals) {
-    std_support::vector<Mutator> mutators(kDefaultThreadCount);
-    std_support::vector<ObjHeader*> globals(kDefaultThreadCount);
-    std_support::vector<ObjHeader*> locals(kDefaultThreadCount);
-    std_support::vector<ObjHeader*> reachablesLocals(kDefaultThreadCount);
-    std_support::vector<ObjHeader*> reachablesGlobals(kDefaultThreadCount);
-
-    auto expandRootSet = [&globals, &locals, &reachablesLocals, &reachablesGlobals](mm::ThreadData& threadData, Mutator& mutator, int i) {
-        auto& global = mutator.AddGlobalRoot();
-        auto& local = mutator.AddStackRoot();
-        auto& reachableLocal = AllocateObject(threadData);
-        auto& reachableGlobal = AllocateObject(threadData);
-        local->field1 = reachableLocal.header();
-        global->field1 = reachableGlobal.header();
-        globals[i] = global.header();
-        locals[i] = local.header();
-        reachablesLocals[i] = reachableLocal.header();
-        reachablesGlobals[i] = reachableGlobal.header();
-    };
-
-    for (int i = 0; i < kDefaultThreadCount; ++i) {
-        mutators[i]
-                .Execute([i, expandRootSet](mm::ThreadData& threadData, Mutator& mutator) { expandRootSet(threadData, mutator, i); })
-                .wait();
-    }
-
-    std_support::vector<std::future<void>> gcFutures(kDefaultThreadCount);
-
-    mm::GlobalData::Instance().gc().impl().gc().SetMarkingRequested(0);
-
-    for (int i = 0; i < kDefaultThreadCount; ++i) {
-        gcFutures[i] = mutators[i]
-            .Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().impl().gc().OnSuspendForGC(); });
-    }
-
-    if (GetParam() == gc::ConcurrentMarkAndSweep::kMarkOwnStack) {
-        mm::GlobalData::Instance().gc().impl().gc().WaitForThreadsReadyToMark();
-        mm::GlobalData::Instance().gc().impl().gc().CollectRootSetAndStartMarking(gc::GCHandle::createFakeForTests());
-    }
-
-    for (int i = 0; i < kDefaultThreadCount; ++i) {
-        gcFutures[i].wait();
-        // Verify that threads marked their own locals (but not their globals) according to the configured marking behavior:
-        ASSERT_THAT(IsMarked(reachablesLocals[i]), GetParam() == kotlin::gc::ConcurrentMarkAndSweep::kMarkOwnStack);
-        ASSERT_THAT(IsMarked(reachablesGlobals[i]), false);
-    }
-
-    for (auto& future : gcFutures) {
-        future.wait();
-    }
-}
-
 INSTANTIATE_TEST_SUITE_P(,
-                         ConcurrentMarkAndSweepTest,
-                         testing::Values(gc::ConcurrentMarkAndSweep::MarkingBehavior::kDoNotMark, gc::ConcurrentMarkAndSweep::MarkingBehavior::kMarkOwnStack),
-                         [] (const testing::TestParamInfo<gc::ConcurrentMarkAndSweep::MarkingBehavior>& behavior) { return (behavior.param == gc::ConcurrentMarkAndSweep::MarkingBehavior::kDoNotMark) ? "SingleThreadedMarking" : "MutatorsMarkOwnStack"; });
+    ConcurrentMarkAndSweepTest,
+    testing::Values(
+            ParallelismOptions{kDefaultThreadCount * 3, false, 0},
+            ParallelismOptions{kDefaultThreadCount * 3, true, 0},
+            ParallelismOptions{kDefaultThreadCount * 3, false, kDefaultThreadCount},
+            ParallelismOptions{kDefaultThreadCount * 3, true, kDefaultThreadCount},
+
+            ParallelismOptions{kDefaultThreadCount / 2, true, kDefaultThreadCount},
+            ParallelismOptions{kDefaultThreadCount / 2 * 3, true, kDefaultThreadCount}
+    ),
+    [] (const testing::TestParamInfo<ParallelismOptions>& paramInfo) {
+        using namespace std::string_literals;
+        auto base = "Mark"s;
+        auto parallelism = std::to_string(paramInfo.param.maxParallelism) + "Parallel";
+        auto withMutators = paramInfo.param.cooperativeMutators ? "WithMutators" : "";
+        auto withAux = paramInfo.param.auxGCThreads > 0 ? "WithGCThreads" : "";
+        return base + parallelism + withMutators + withAux;
+    });
 

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

@@ -11,6 +11,7 @@
 #include "MarkAndSweepUtils.hpp"
 #include "ObjectOps.hpp"
 #include "ThreadSuspension.hpp"
+#include "MarkStack.hpp"
 #include "std_support/Memory.hpp"
 
 using namespace kotlin;
@@ -118,17 +119,17 @@ bool gc::GC::FinalizersThreadIsRunning() noexcept {
 
 // static
 ALWAYS_INLINE void gc::GC::processObjectInMark(void* state, ObjHeader* object) noexcept {
-    gc::internal::processObjectInMark<gc::internal::MarkTraits>(state, object);
+    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::internal::MarkTraits>(state, array);
+    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::internal::MarkTraits>(state, field);
+    gc::internal::processFieldInMark<gc::mark::ParallelMark::MarkTraits>(state, field);
 }
 
 int64_t gc::GC::Schedule() noexcept {

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

@@ -23,9 +23,9 @@ using GCImpl = ConcurrentMarkAndSweep;
 class GC::Impl : private Pinned {
 public:
 #ifdef CUSTOM_ALLOCATOR
-    explicit Impl(gcScheduler::GCScheduler& gcScheduler) noexcept : gc_(gcScheduler) {}
+    explicit Impl(gcScheduler::GCScheduler& gcScheduler) noexcept : gc_(gcScheduler, compiler::gcMutatorsCooperate(), compiler::auxGCThreads()) {}
 #else
-    explicit Impl(gcScheduler::GCScheduler& gcScheduler) noexcept : gc_(objectFactory_, extraObjectDataFactory_, gcScheduler) {}
+    explicit Impl(gcScheduler::GCScheduler& gcScheduler) noexcept : gc_(objectFactory_, extraObjectDataFactory_, gcScheduler, compiler::gcMutatorsCooperate(), compiler::auxGCThreads()) {}
 #endif
 
 #ifndef CUSTOM_ALLOCATOR

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

@@ -0,0 +1,57 @@
+#pragma once
+
+#include <atomic>
+#include <cstddef>
+
+#include "Allocator.hpp"
+#include "IntrusiveList.hpp"
+#include "ObjectFactory.hpp"
+#include "Types.h"
+#include "Utils.hpp"
+#include "std_support/Memory.hpp"
+
+namespace kotlin::gc::mark {
+
+class ObjectData {
+    struct ObjectFactoryTraits {
+        using ObjectData = ObjectData;
+        class Allocator;
+    };
+public:
+    using ObjectFactory = mm::ObjectFactory<ObjectFactoryTraits>;
+
+    bool tryMark() noexcept {
+        return trySetNext(reinterpret_cast<ObjectData*>(1));
+    }
+
+    bool marked() const noexcept { return next() != nullptr; }
+
+    bool tryResetMark() noexcept {
+        if (next() == nullptr) return false;
+        next_.store(nullptr, std::memory_order_relaxed);
+        return true;
+    }
+
+    ObjHeader* objHeader() noexcept { // FIXME const
+        return ObjectFactory::NodeRef::From(*this).GetObjHeader();
+    }
+
+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;
+};
+
+
+} // namespace kotlin::gc::mark

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

@@ -0,0 +1,243 @@
+#include "ParallelMark.hpp"
+
+#include "MarkStack.hpp"
+#include "MarkAndSweepUtils.hpp"
+#include "GCStatistics.hpp"
+#include "Utils.hpp"
+#include "std_support/Memory.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 {
+    return phase_.load(std::memory_order_relaxed) == phase;
+}
+
+void gc::mark::MarkPacer::begin(gc::mark::MarkPacer::Phase phase) {
+    {
+        std::unique_lock lock(mutex_);
+        phase_.store(phase, std::memory_order_relaxed);
+    }
+    cond_.notify_all();
+}
+
+void gc::mark::MarkPacer::wait(gc::mark::MarkPacer::Phase phase) {
+    if (phase_.load(std::memory_order_relaxed) >= phase) return;
+    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_.load(std::memory_order_relaxed) >= Phase::kReady); });
+}
+
+void gc::mark::MarkPacer::waitEpochFinished(uint64_t currentEpoch) const {
+    std::unique_lock lock(mutex_);
+    cond_.wait(lock, [this, currentEpoch]() {
+        return is(Phase::kIdle) || is(Phase::kShutdown) || epoch_.load(std::memory_order_relaxed) > currentEpoch;
+    });
+}
+
+bool gc::mark::MarkPacer::acceptingNewWorkers() const {
+    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::waitForThreadsPauseMutation() noexcept {
+    RuntimeAssert(!kotlin::mm::IsCurrentThreadRegistered(), "Dispatcher thread must not be registered");
+    spinWait([this] {
+        return allMutators([](mm::ThreadData& mut) {
+            return mm::isSuspendedOrNative(mut) || mut.gc().impl().gc().cooperative();
+        });
+    });
+}
+
+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 epoch = gcHandle().getEpoch();
+    auto parallelWorker = createWorker();
+    if (parallelWorker) {
+        auto& gcData = mutatorThread.gc().impl().gc();
+        gcData.beginCooperation();
+        GCLogDebug(epoch, "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,181 @@
+#pragma once
+
+#include <mutex>
+#include <condition_variable>
+
+#include "GCStatistics.hpp"
+#include "MarkStack.hpp"
+#include "std_support/Vector.hpp"
+#include "ThreadRegistry.hpp"
+#include "Utils.hpp"
+#include "ParallelProcessor.hpp"
+#include "ManuallyScoped.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;
+    std::atomic<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<ObjectData>;
+    // work balancing parameters were chosen pretty arbitrary
+    using ParallelProcessor = ParallelProcessor<MarkStackImpl, 512, 4096>;
+public:
+    class MarkTraits {
+    public:
+        using MarkQueue = ParallelProcessor::Worker;
+        using ObjectFactory = ObjectData::ObjectFactory;
+
+        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) {
+                auto node = ObjectFactory::NodeRef::From(*obj);
+                return node->GetObjHeader();
+            }
+            return nullptr;
+        }
+
+        static ALWAYS_INLINE bool tryEnqueue(MarkQueue& queue, ObjHeader* object) noexcept {
+            auto& objectData = ObjectFactory::NodeRef::From(object).ObjectData();
+            return compiler::gcMarkSingleThreaded() ? queue.tryPushLocal(objectData) : queue.tryPush(objectData);
+        }
+
+        static ALWAYS_INLINE bool tryMark(ObjHeader* object) noexcept {
+            auto& objectData = ObjectFactory::NodeRef::From(object).ObjectData();
+            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 waitForThreadsPauseMutation() noexcept;
+    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
+

kotlin-native/runtime/src/gc/common/cpp/GCStatistics.cpp

@@ -12,7 +12,7 @@
 #include "ThreadData.hpp"
 #include "std_support/Optional.hpp"
 #include <cinttypes>
-#include <mutex>
+#include <limits>
 
 using namespace kotlin;
 
@@ -153,15 +153,18 @@ GCHandle GCHandle::create(uint64_t epoch) {
     current.epoch = static_cast<KLong>(epoch);
     current.startTime = static_cast<KLong>(konan::getTimeNanos());
     if (last.endTime) {
-        GCLogInfo(epoch, "Started. Time since last GC %" PRIu64 " microseconds.", *current.startTime - *last.endTime);
+        auto time = (*current.startTime - *last.endTime) / 1000;
+        GCLogInfo(epoch, "Started. Time since last GC %" PRIu64 " microseconds.", time);
     } else {
         GCLogInfo(epoch, "Started.");
     }
     current.memoryUsageBefore.heap = currentHeapUsage();
     return getByEpoch(epoch);
 }
-GCHandle GCHandle::createFakeForTests() { return getByEpoch(std::numeric_limits<uint64_t>::max()); }
+GCHandle GCHandle::createFakeForTests() { return getByEpoch(invalid().getEpoch() - 1); }
 GCHandle GCHandle::getByEpoch(uint64_t epoch) {
+    GCHandle handle{epoch};
+    RuntimeAssert(handle.isValid(), "Must be valid");
     return GCHandle{epoch};
 }
 
@@ -174,11 +177,17 @@ std::optional<gc::GCHandle> gc::GCHandle::currentEpoch() noexcept {
     return std::nullopt;
 }
 
+GCHandle GCHandle::invalid() {
+    return GCHandle{std::numeric_limits<uint64_t>::max()};
+}
 void GCHandle::ClearForTests() {
     std::lock_guard guard(lock);
     current = {};
     last = {};
 }
+bool GCHandle::isValid() const {
+    return epoch_ != GCHandle::invalid().epoch_;
+}
 void GCHandle::finished() {
     std::lock_guard guard(lock);
     if (auto* stat = statByEpoch(epoch_)) {

kotlin-native/runtime/src/gc/common/cpp/GCStatistics.hpp

@@ -9,6 +9,7 @@
 #include <pthread.h>
 
 #include "Common.h"
+#include "Logging.hpp"
 #include "Porting.h"
 #include "Utils.hpp"
 #include "std_support/Optional.hpp"
@@ -135,9 +136,11 @@ public:
     static GCHandle createFakeForTests();
     static GCHandle getByEpoch(uint64_t epoch);
     static std::optional<GCHandle> currentEpoch() noexcept;
+    static GCHandle invalid();
     static void ClearForTests();
 
     uint64_t getEpoch() { return epoch_; }
+    bool isValid() const;
     void finished();
     void finalizersDone();
     void finalizersScheduled(uint64_t finalizersCount);

kotlin-native/runtime/src/gc/common/cpp/MarkAndSweepUtils.hpp

@@ -35,25 +35,20 @@ void processFieldInMark(void* state, ObjHeader* field) noexcept {
 
 template <typename Traits>
 void processObjectInMark(void* state, ObjHeader* object) noexcept {
-    auto* typeInfo = object->type_info();
-    RuntimeAssert(typeInfo != theArrayTypeInfo, "Must not be an array of objects");
-    for (int i = 0; i < typeInfo->objOffsetsCount_; ++i) {
-        auto* field = *reinterpret_cast<ObjHeader**>(reinterpret_cast<uintptr_t>(object) + typeInfo->objOffsets_[i]);
-        if (!field) continue;
-        processFieldInMark<Traits>(state, field);
-    }
+    traverseClassObjectFields(object, [state] (ObjHeader** fieldLocation) noexcept {
+        if (auto field = *fieldLocation) {
+            processFieldInMark<Traits>(state, field);
+        }
+    });
 }
 
 template <typename Traits>
 void processArrayInMark(void* state, ArrayHeader* array) noexcept {
-    RuntimeAssert(array->type_info() == theArrayTypeInfo, "Must be an array of objects");
-    auto* begin = ArrayAddressOfElementAt(array, 0);
-    auto* end = ArrayAddressOfElementAt(array, array->count_);
-    for (auto* it = begin; it != end; ++it) {
-        auto* field = *it;
-        if (!field) continue;
-        processFieldInMark<Traits>(state, field);
-    }
+    traverseArrayOfObjectsElements(array, [state] (ObjHeader** elemLocation) noexcept {
+        if (auto elem = *elemLocation) {
+            processFieldInMark<Traits>(state, elem);
+        }
+    });
 }
 
 template <typename Traits>
@@ -93,6 +88,11 @@ void processExtraObjectData(GCHandle::GCMarkScope& markHandle, typename Traits::
 template <typename Traits>
 void Mark(GCHandle handle, typename Traits::MarkQueue& markQueue) noexcept {
     auto markHandle = handle.mark();
+    Mark<Traits>(markHandle, markQueue);
+}
+
+template <typename Traits>
+void Mark(GCHandle::GCMarkScope& markHandle, typename Traits::MarkQueue& markQueue) noexcept {
     while (ObjHeader* top = Traits::tryDequeue(markQueue)) {
         markHandle.addObject();
 

kotlin-native/runtime/src/main/cpp/BoundedQueue.hpp

@@ -0,0 +1,114 @@
+/*
+ * 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.
+ */
+
+
+/*
+ * An implementation of Dmitry Vyukov's Bounded Multi-producer/multi-consumer bounded queue.
+ *
+ *  Copyright (c) 2010-2011, Dmitry Vyukov. All rights reserved.
+ *  Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:
+ *     1. Redistributions of source code must retain the above copyright notice, this list of
+ *        conditions and the following disclaimer.
+ *     2. Redistributions in binary form must reproduce the above copyright notice, this list
+ *        of conditions and the following disclaimer in the documentation and/or other materials
+ *        provided with the distribution.
+ *  THIS SOFTWARE IS PROVIDED BY DMITRY VYUKOV "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
+ *  THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ *  DMITRY VYUKOV OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+ *  (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
+ *  HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ *  ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *  The views and conclusions contained in the software and documentation are those of the authors and should not be interpreted
+ *  as representing official policies, either expressed or implied, of Dmitry Vyukov.
+ *
+ *  TODO what about a binary distribution?
+ */
+
+#pragma once
+
+#include <atomic>
+
+#include "Utils.hpp"
+#include "ManuallyScoped.hpp"
+
+namespace kotlin {
+
+/**
+ * A fixed-size concurrent multi-producer/multi-consumer queue.
+ * @tparam kCapacity must be a power of 2.
+ */
+template<typename T, std::size_t kCapacity>
+class BoundedQueue : private Pinned {
+public:
+    BoundedQueue() {
+        static_assert((kCapacity >= 2) && ((kCapacity & (kCapacity - 1)) == 0), "Queue capacity must be a power of 2");
+
+        for (size_t i = 0; i < kCapacity; ++i) {
+            buffer_[i].sequence_.store(i, std::memory_order_relaxed);
+        }
+        enqueuePos_.store(0, std::memory_order_relaxed);
+        dequeuePos_.store(0, std::memory_order_relaxed);
+    }
+
+    bool enqueue(T&& value) {
+        Cell* cell;
+        std::size_t pos = enqueuePos_.load(std::memory_order_relaxed);
+        while (true) {
+            cell = &buffer_[pos & (kCapacity - 1)];
+            std::size_t seq = cell->sequence_.load(std::memory_order_acquire);
+            std::intptr_t dif = static_cast<std::intptr_t>(seq) - static_cast<std::intptr_t>(pos);
+            if (dif == 0) {
+                if (enqueuePos_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed)) {
+                    break;
+                }
+            } else if (dif < 0) {
+                return false;
+            } else {
+                pos = enqueuePos_.load(std::memory_order_relaxed);
+            }
+        }
+        cell->data_.construct(std::move(value));
+        cell->sequence_.store(pos + 1, std::memory_order_release);
+        return true;
+    }
+
+    std::optional<T> dequeue() {
+        Cell* cell;
+        std::size_t pos = dequeuePos_.load(std::memory_order_relaxed);
+        while (true) {
+            cell = &buffer_[pos & (kCapacity - 1)];
+            std::size_t seq = cell->sequence_.load(std::memory_order_acquire);
+            std::intptr_t dif = static_cast<std::intptr_t>(seq) - static_cast<std::intptr_t>(pos + 1);
+            if (dif == 0) {
+                if (dequeuePos_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed)) {
+                    break;
+                }
+            } else if (dif < 0) {
+                return std::nullopt;
+            } else {
+                pos = dequeuePos_.load(std::memory_order_relaxed);
+            }
+        }
+        auto result = std::move(*cell->data_);
+        cell->data_.destroy();
+        cell->sequence_.store(pos + kCapacity, std::memory_order_release);
+        return std::move(result);
+    }
+
+private:
+    struct Cell {
+        // TODO describe
+        std::atomic<size_t> sequence_;
+        ManuallyScoped<T> data_;
+    };
+
+    constexpr static auto kCacheLineSize = 128;
+
+    alignas(kCacheLineSize) Cell buffer_[kCapacity];
+    alignas(kCacheLineSize) std::atomic<size_t> enqueuePos_;
+    alignas(kCacheLineSize) std::atomic<size_t> dequeuePos_;
+};
+
+} // namespace kotlin

kotlin-native/runtime/src/main/cpp/BoundedQueueTest.cpp

@@ -0,0 +1,130 @@
+/*
+ * 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 "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+#include <list>
+
+#include "IntrusiveList.hpp"
+
+#include "ParallelProcessor.hpp"
+
+#include "std_support/Vector.hpp"
+#include "SingleThreadExecutor.hpp"
+#include "TestSupport.hpp"
+
+using ::testing::_;
+using namespace kotlin;
+
+namespace {
+
+class Element {
+public:
+    Element() : Element(0) {}
+    explicit Element(int value) : a(value), b(value), c(value), d(value) {}
+
+    bool isValid() const {
+        return a == b && b == c && c == d;
+    }
+
+private:
+    std::size_t a;
+    std::size_t b;
+    std::size_t c;
+    std::size_t d;
+};
+
+} // namespace
+
+
+TEST(BoundedQueueTest, ConcurrentEnqueue) {
+    constexpr auto kThreadCount = 16;
+    constexpr auto kElemsPerThread = 1024;
+    BoundedQueue<Element, kThreadCount * kElemsPerThread> queue;
+
+    std::atomic<bool> start = false;
+    std::list<ScopedThread> threads;
+    for (int t = 0; t < kThreadCount; ++t) {
+        threads.emplace_back([&, t]() {
+            while (!start) {
+                std::this_thread::yield();
+            }
+            for (int e = 0; e < kElemsPerThread; ++e) {
+                queue.enqueue(Element(t + e));
+            }
+        });
+    }
+    start = true;
+
+    threads.clear();
+
+    while (auto elem = queue.dequeue()) {
+        EXPECT_TRUE(elem->isValid());
+    }
+}
+
+TEST(BoundedQueueTest, ConcurrentDequeue) {
+    constexpr auto kThreadCount = 16;
+    constexpr auto kElemsPerThread = 1024;
+    BoundedQueue<Element, kThreadCount * kElemsPerThread> queue;
+
+    for (int i = 0; i < kThreadCount * kElemsPerThread; ++i) {
+        queue.enqueue(Element(i));
+    }
+
+    std::atomic<bool> start = false;
+    std::list<ScopedThread> threads;
+    for (int t = 0; t < kThreadCount; ++t) {
+        threads.emplace_back([&]() {
+            while (!start) {
+                std::this_thread::yield();
+            }
+            while (auto elem = queue.dequeue()) {
+                EXPECT_TRUE(elem->isValid());
+            }
+        });
+    }
+    start = true;
+}
+
+TEST(BoundedQueueTest, PingPongWithOverflow) {
+    constexpr auto kElemsPerThread = 1024;
+    BoundedQueue<Element, kElemsPerThread / 2> queue;
+
+    std::atomic<bool> start = false;
+    std::list<ScopedThread> writers;
+    for (std::size_t t = 0; t < kDefaultThreadCount; ++t) {
+        writers.emplace_back([&]() {
+            while (!start) {
+                std::this_thread::yield();
+            }
+            for (int i = 0; i < kElemsPerThread; ++i) {
+                while (!queue.enqueue(Element(i))) {
+                    std::this_thread::yield();
+                }
+            }
+        });
+    }
+
+    std::atomic<bool> allWritten = false;
+    std::list<ScopedThread> readers;
+    for (std::size_t t = 0; t < kDefaultThreadCount; ++t) {
+        readers.emplace_back([&]() {
+            while (!start) {
+                std::this_thread::yield();
+            }
+            while (!allWritten) {
+                while (auto elem = queue.dequeue()) {
+                    EXPECT_TRUE(elem->isValid());
+                }
+            }
+        });
+    }
+
+    start = true;
+    writers.clear();
+    allWritten = true;
+}

kotlin-native/runtime/src/main/cpp/CompilerConstants.cpp

@@ -19,7 +19,8 @@ using Kotlin_getSourceInfo_FunctionType = int(*)(void * /*addr*/, SourceInfo* /*
  * but can be changed after compiling caches. So use this way for variables, which will be rarely accessed.
  */
 RUNTIME_WEAK int32_t Kotlin_destroyRuntimeMode = 1;
-RUNTIME_WEAK int32_t Kotlin_gcMarkSingleThreaded = 1;
+RUNTIME_WEAK int32_t Kotlin_gcMutatorsCooperate = 0;
+RUNTIME_WEAK uint32_t Kotlin_auxGCThreads = 0;
 RUNTIME_WEAK int32_t Kotlin_workerExceptionHandling = 0;
 RUNTIME_WEAK int32_t Kotlin_suspendFunctionsFromAnyThreadFromObjC = 0;
 RUNTIME_WEAK Kotlin_getSourceInfo_FunctionType Kotlin_getSourceInfo_Function = nullptr;
@@ -36,8 +37,12 @@ ALWAYS_INLINE compiler::DestroyRuntimeMode compiler::destroyRuntimeMode() noexce
     return static_cast<compiler::DestroyRuntimeMode>(Kotlin_destroyRuntimeMode);
 }
 
-ALWAYS_INLINE bool compiler::gcMarkSingleThreaded() noexcept {
-    return Kotlin_gcMarkSingleThreaded != 0;
+ALWAYS_INLINE bool compiler::gcMutatorsCooperate() noexcept {
+    return Kotlin_gcMutatorsCooperate != 0;
+}
+
+ALWAYS_INLINE uint32_t compiler::auxGCThreads() noexcept {
+    return Kotlin_auxGCThreads;
 }
 
 ALWAYS_INLINE compiler::WorkerExceptionHandling compiler::workerExceptionHandling() noexcept {

kotlin-native/runtime/src/main/cpp/CompilerConstants.hpp

@@ -38,6 +38,7 @@ extern "C" const int32_t Kotlin_runtimeAssertsMode;
 extern "C" const int32_t Kotlin_disableMmap;
 extern "C" const char* const Kotlin_runtimeLogs;
 extern "C" const int32_t Kotlin_concurrentWeakSweep;
+extern "C" const int32_t Kotlin_gcMarkSingleThreaded;
 extern "C" const int32_t Kotlin_freezingEnabled;
 extern "C" const int32_t Kotlin_freezingChecksEnabled;
 
@@ -103,9 +104,15 @@ ALWAYS_INLINE inline bool concurrentWeakSweep() noexcept {
     return Kotlin_concurrentWeakSweep != 0;
 }
 
+ALWAYS_INLINE inline bool gcMarkSingleThreaded() noexcept {
+    return Kotlin_gcMarkSingleThreaded != 0;
+}
+
+
 WorkerExceptionHandling workerExceptionHandling() noexcept;
 DestroyRuntimeMode destroyRuntimeMode() noexcept;
-bool gcMarkSingleThreaded() noexcept;
+bool gcMutatorsCooperate() noexcept;
+uint32_t auxGCThreads() noexcept;
 bool suspendFunctionsFromAnyThreadFromObjCEnabled() noexcept;
 AppStateTracking appStateTracking() noexcept;
 int getSourceInfo(void* addr, SourceInfo *result, int result_size) noexcept;

kotlin-native/runtime/src/main/cpp/CooperativeIntrusiveList.hpp

@@ -1,116 +0,0 @@
-/*
- * 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 <mutex>
-#include "IntrusiveList.hpp"
-#include "Porting.h"
-#include "Mutex.hpp"
-
-namespace kotlin {
-
-/**
- * A list for thread-local use with a separate shared part that can be accessed by other threads.
- */
-template<typename T, typename Traits = DefaultIntrusiveForwardListTraits<T>>
-class CooperativeIntrusiveList : Pinned {
-    using ListImpl = intrusive_forward_list<T, Traits>;
-public:
-    using value_type = typename ListImpl::value_type;
-    using size_type = typename ListImpl::size_type;
-    using reference = typename ListImpl::reference;
-    using pointer = typename ListImpl::pointer;
-
-    CooperativeIntrusiveList() = default;
-
-    bool localEmpty() const {
-        return local_.empty();
-    }
-
-    size_type localSize() const {
-        return localSize_;
-    }
-
-    /**
-     * Tries to add `value` to the local list.
-     * See `intrusive_forward_list.try_push_front`.
-     */
-    bool tryPushLocal(reference value) {
-        auto pushed = local_.try_push_front(value);
-        if (pushed) ++localSize_;
-        return pushed;
-    }
-
-    /**
-     * Tries to pop a value from the local list.
-     * See `intrusive_forward_list.try_pop_front`.
-     */
-    pointer tryPopLocal() {
-        auto popped = local_.try_pop_front();
-        if (popped) {
-            --localSize_;
-        } else {
-            RuntimeAssert(localEmpty(), "Pop can only fail if the list is empty");
-        }
-        return popped;
-    }
-
-    void clearLocal() {
-        local_.clear();
-        localSize_ = 0;
-    }
-
-    bool sharedEmpty() const {
-        return shared_.empty();
-    }
-
-    /**
-     * Tries to move at most `maxAmount` elements from a from's shared list into `this`'s local list.
-     * In case some other thread is currently operating with the from's shared list, returns `0`.
-     * @return the number of elements stolen
-     */
-    size_type tryTransferFrom(CooperativeIntrusiveList<T, Traits>& from, size_type maxAmount) noexcept {
-        std::unique_lock guard(from.sharedLocked_, std::try_to_lock);
-        if (!guard || from.sharedEmpty()) {
-            return 0;
-        }
-
-        auto amount = local_.splice_after(local_.before_begin(),
-                                          from.shared_.before_begin(),
-                                          from.shared_.end(),
-                                          maxAmount);
-        from.sharedSize_ -= amount;
-        localSize_ += amount;
-
-        return amount;
-    }
-
-    /**
-     * Moves all of the local items into own shared list.
-     * @return `0` if the shared list is busy, amount of newly shared items otherwise.
-     */
-    size_type shareAll() noexcept {
-        RuntimeAssert(!localEmpty(), "Nothing to share");
-        std::unique_lock guard(sharedLocked_, std::try_to_lock);
-        if (!guard) return 0;
-
-        auto amount = shared_.splice_after(shared_.before_begin(), local_.before_begin(), local_.end(), localSize_);
-        sharedSize_ += amount;
-        localSize_ -= amount;
-
-        return amount;
-    }
-
-private:
-    ListImpl local_;
-    size_type localSize_ = 0;
-
-    ListImpl shared_;
-    size_type sharedSize_ = 0;
-    SpinLock<MutexThreadStateHandling::kIgnore> sharedLocked_;
-};
-
-}

kotlin-native/runtime/src/main/cpp/CooperativeIntrusiveListTest.cpp

@@ -1,236 +0,0 @@
-/*
- * 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 "CooperativeIntrusiveList.hpp"
-
-#include "gmock/gmock.h"
-#include "gtest/gtest.h"
-
-#include "ScopedThread.hpp"
-#include "TestSupport.hpp"
-
-#include "std_support/Deque.hpp"
-#include "std_support/Vector.hpp"
-#include "std_support/List.hpp"
-
-using namespace kotlin;
-
-using ::testing::_;
-
-namespace {
-
-class Node : private Pinned {
-public:
-    explicit Node(int value) : value_(value) {}
-
-    int& operator*() { return value_; }
-    const int& operator*() const { return value_; }
-
-    void clearNext() noexcept { next_ = nullptr; }
-
-    int value() const {
-        return value_;
-    }
-
-private:
-    friend struct DefaultIntrusiveForwardListTraits<Node>;
-
-    Node* next() const noexcept { return next_; }
-    void setNext(Node* next) noexcept {
-        RuntimeAssert(next, "next cannot be nullptr");
-        next_ = next;
-    }
-    bool trySetNext(Node* next) noexcept {
-        RuntimeAssert(next, "next cannot be nullptr");
-        if (next_) return false;
-        next_ = next;
-        return true;
-    }
-
-    int value_;
-    Node* next_ = nullptr;
-};
-
-using TestSubject = CooperativeIntrusiveList<Node>;
-
-std_support::vector<int> range(int first, int lastExclusive) {
-    std_support::vector<int> values;
-    for (int i = first; i < lastExclusive; ++i) {
-        values.push_back(i);
-    }
-    return values;
-}
-
-template<typename Values>
-[[nodiscard]] std_support::list<typename TestSubject::value_type> fill(TestSubject& list, Values&& values) {
-    std_support::list<typename TestSubject::value_type> nodesHandle;
-    for (int value: values) {
-        auto& elem = nodesHandle.emplace_back(value);
-        list.tryPushLocal(elem);
-    }
-    return nodesHandle;
-}
-
-void drainLocalInto(TestSubject& list, std_support::vector<int>& dest) {
-    while (auto elem = list.tryPopLocal()) {
-        dest.push_back(elem->value());
-    }
-}
-
-} // namespace
-
-TEST(CooperativeIntrusiveListTest, Init) {
-    TestSubject list;
-    EXPECT_THAT(list.localEmpty(), true);
-    EXPECT_THAT(list.localSize(), 0);
-    EXPECT_THAT(list.sharedEmpty(), true);
-}
-
-TEST(CooperativeIntrusiveListTest, TryPopLocalEmpty) {
-    TestSubject list;
-    auto res = list.tryPopLocal();
-    EXPECT_THAT(res, nullptr);
-}
-
-TEST(CooperativeIntrusiveListTest, TryPushLocalPopLocal) {
-    TestSubject list;
-    typename TestSubject::value_type value1(1);
-    typename TestSubject::value_type value2(2);
-    bool pushed1 = list.tryPushLocal(value1);
-    bool pushed2 = list.tryPushLocal(value2);
-    EXPECT_THAT(pushed1, true);
-    EXPECT_THAT(pushed2, true);
-    EXPECT_THAT(list.localEmpty(), false);
-    EXPECT_THAT(list.localSize(), 2);
-    EXPECT_THAT(list.sharedEmpty(), true);
-    std_support::vector<int> popped;
-    drainLocalInto(list, popped);
-    EXPECT_THAT(list.localEmpty(), true);
-    EXPECT_THAT(list.localSize(), 0);
-    EXPECT_THAT(list.sharedEmpty(), true);
-    EXPECT_THAT(popped, testing::UnorderedElementsAre(1, 2));
-}
-
-TEST(CooperativeIntrusiveListTest, TryPushLocalTwice) {
-    TestSubject list;
-    typename TestSubject::value_type value(1);
-    bool pushed1 = list.tryPushLocal(value);
-    EXPECT_THAT(pushed1, true);
-    bool pushed2 = list.tryPushLocal(value);
-    EXPECT_THAT(pushed2, false);
-    EXPECT_THAT(list.localEmpty(), false);
-    EXPECT_THAT(list.localSize(), 1);
-    EXPECT_THAT(list.sharedEmpty(), true);
-}
-
-TEST(CooperativeIntrusiveListTest, ShareSome) {
-    TestSubject list;
-    auto values = range(0, 10);
-    auto nodeHandle = fill(list, values);
-    EXPECT_THAT(list.localEmpty(), false);
-    EXPECT_THAT(list.localSize(), values.size());
-    EXPECT_THAT(list.sharedEmpty(), true);
-    auto sharedAmount = list.shareAll();
-    EXPECT_THAT(sharedAmount, values.size());
-    EXPECT_THAT(list.localEmpty(), true);
-    EXPECT_THAT(list.sharedEmpty(), false);
-}
-
-TEST(CooperativeIntrusiveListTest, TryTransferFromEmpty) {
-    TestSubject from;
-    TestSubject thief;
-    auto stolenAmount = thief.tryTransferFrom(from, 1);
-    EXPECT_THAT(stolenAmount, 0);
-}
-
-TEST(CooperativeIntrusiveListTest, TryTransferHalf) {
-    TestSubject from;
-    auto values = range(0, 10);
-    auto nodeHandle = fill(from, values);
-    from.shareAll();
-
-    TestSubject thief;
-    auto toTransfer = values.size() / 2;
-    auto stolenAmount = thief.tryTransferFrom(from, toTransfer);
-    EXPECT_THAT(stolenAmount, toTransfer);
-    EXPECT_THAT(thief.localSize(), stolenAmount);
-
-    from.tryTransferFrom(from, values.size());
-    EXPECT_THAT(from.sharedEmpty(), true);
-
-    std_support::vector<int> allTheElements;
-    drainLocalInto(from, allTheElements);
-    drainLocalInto(thief, allTheElements);
-    EXPECT_THAT(allTheElements, testing::UnorderedElementsAreArray(values));
-}
-
-TEST(CooperativeIntrusiveListTest, TryTransferAllEventually) {
-    TestSubject from;
-    auto values = range(0, 10);
-    auto nodeHandle = fill(from, values);
-    from.shareAll();
-
-    TestSubject thief;
-    for (std::size_t i = 0; i < values.size(); ++i) {
-        auto stolenAmount = thief.tryTransferFrom(from, 1);
-        EXPECT_THAT(stolenAmount, 1);
-    }
-    EXPECT_THAT(from.sharedEmpty(), true);
-    EXPECT_THAT(thief.localSize(), values.size());
-
-    std_support::vector<int> allTheElements;
-    drainLocalInto(from, allTheElements);
-    drainLocalInto(thief, allTheElements);
-    EXPECT_THAT(allTheElements, testing::UnorderedElementsAreArray(values));
-}
-
-TEST(CooperativeIntrusiveListTest, TransferingPingPong) {
-    TestSubject list1;
-    TestSubject list2;
-    const auto size = 100;
-    auto values = range(0, size);
-    auto nodesHandle1 = fill(list1, values);
-    auto nodesHandle2 = fill(list2, values);
-
-    std::atomic ready = false;
-    auto kIters = 10000;
-    std_support::vector<ScopedThread> threads;
-    for (int tIdx = 0; tIdx < 2; ++tIdx) {
-        threads.emplace_back([&ready, kIters, tIdx, &list1, &list2] {
-            TestSubject& self = tIdx % 2 == 0 ? list1 : list2;
-            TestSubject& from = tIdx % 2 == 0 ? list2 : list1;
-            while (!ready.load()) {
-                std::this_thread::yield();
-            }
-            for (int iter = 0; iter < kIters; ++iter) {
-                if (!self.localEmpty()) self.shareAll();
-                self.tryTransferFrom(from, size / 2);
-                if (!self.localEmpty()) self.shareAll();
-                self.tryTransferFrom(from, size);
-                if (auto popped = self.tryPopLocal()) {
-                    popped->clearNext();
-                    self.tryPushLocal(*popped);
-                }
-            }
-        });
-    }
-    ready = true;
-
-    for (auto& thr: threads) {
-        thr.join();
-    }
-
-    // check nothing is lost
-    list1.tryTransferFrom(list1, size * 2);
-    list2.tryTransferFrom(list2, size * 2);
-    std_support::vector<int> allTheElements;
-    drainLocalInto(list1, allTheElements);
-    drainLocalInto(list2, allTheElements);
-
-    std_support::vector<int> expected;
-    expected.insert(expected.end(), values.begin(), values.end());
-    expected.insert(expected.end(), values.begin(), values.end());
-    EXPECT_THAT(allTheElements, testing::UnorderedElementsAreArray(expected));
-}

kotlin-native/runtime/src/main/cpp/ManuallyScoped.hpp

@@ -31,7 +31,7 @@ public:
     const T* operator->() const noexcept { return impl(); }
 
 private:
-    T* impl() noexcept { return reinterpret_cast<T*>(implStorage_); }
+    __attribute__((used)) T* impl() noexcept { return reinterpret_cast<T*>(implStorage_); }
     const T* impl() const noexcept { return reinterpret_cast<const T*>(implStorage_); }
 
     alignas(T) char implStorage_[sizeof(T)];

kotlin-native/runtime/src/main/cpp/ObjectTraversal.hpp

@@ -16,6 +16,23 @@ namespace kotlin {
 
 // TODO: Consider an iterator/ranges based approaches for traversals.
 
+template <typename F>
+ALWAYS_INLINE void traverseClassObjectFields(ObjHeader* object, F process) noexcept(noexcept(process(std::declval<ObjHeader**>()))) {
+    const TypeInfo* typeInfo = object->type_info();
+    RuntimeAssert(typeInfo != theArrayTypeInfo, "Must not be an array of objects");
+    for (int index = 0; index < typeInfo->objOffsetsCount_; index++) {
+        process(reinterpret_cast<ObjHeader**>(reinterpret_cast<uintptr_t>(object) + typeInfo->objOffsets_[index]));
+    }
+}
+
+template <typename F>
+ALWAYS_INLINE void traverseArrayOfObjectsElements(ArrayHeader* array, F process) noexcept(noexcept(process(std::declval<ObjHeader**>()))) {
+    RuntimeAssert(array->type_info() == theArrayTypeInfo, "Must be an array of objects");
+    for (uint32_t index = 0; index < array->count_; index++) {
+        process(ArrayAddressOfElementAt(array, index));
+    }
+}
+
 template <typename F>
 void traverseObjectFields(ObjHeader* object, F process) noexcept(noexcept(process(std::declval<ObjHeader**>()))) {
     const TypeInfo* typeInfo = object->type_info();

kotlin-native/runtime/src/main/cpp/ParallelProcessor.hpp

@@ -0,0 +1,215 @@
+/*
+ * 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 "CompilerConstants.hpp"
+#include "KAssert.h"
+#include "Logging.hpp"
+#include "Utils.hpp"
+#include "Porting.h"
+#include "BoundedQueue.hpp"
+
+namespace kotlin {
+
+/**
+ * Coordinates a group of workers working in parallel on a large amounts of identical tasks.
+ * The dispatcher will try to balance the work among the workers.
+ *
+ * Requirements:
+ * -  Every instantiated worker must execute `tryPop` sooner or later;
+ * -  Every instantiated worker must finish execution before the destruction of the processor;
+ */
+template <typename ListImpl, std::size_t kBatchSize, std::size_t kBatchesPoolSize>
+class ParallelProcessor : private Pinned {
+    class Batch {
+    public:
+        ALWAYS_INLINE bool empty() const noexcept {
+            return elems_.empty();
+        }
+
+        ALWAYS_INLINE bool full() const noexcept {
+            return elemsCount_ == kBatchSize;
+        }
+
+        ALWAYS_INLINE std::size_t elementsCount() const noexcept {
+            return elemsCount_;
+        }
+
+        ALWAYS_INLINE bool tryPush(typename ListImpl::reference value) noexcept {
+            RuntimeAssert(!full(), "Batch overflow");
+            bool pushed = elems_.try_push_front(value);
+            if (pushed) {
+                ++elemsCount_;
+            }
+            return pushed;
+        }
+
+        ALWAYS_INLINE typename ListImpl::pointer tryPop() noexcept {
+            auto popped = elems_.try_pop_front();
+            if (popped) {
+                --elemsCount_;
+            }
+            return popped;
+        }
+
+        void transferAllInto(ListImpl& dst) noexcept {
+            dst.splice_after(dst.before_begin(), elems_.before_begin(), elems_.end(), std::numeric_limits<typename ListImpl::size_type>::max());
+            RuntimeAssert(empty(), "All the elements must be transferred");
+            elemsCount_ = 0;
+        }
+
+        void fillFrom(ListImpl& src) noexcept {
+            auto spliced = elems_.splice_after(elems_.before_begin(), src.before_begin(), src.end(), kBatchSize);
+            elemsCount_ = spliced;
+        }
+
+    private:
+        ListImpl elems_;
+        std::size_t elemsCount_ = 0;
+    };
+
+public:
+    class Worker : private Pinned {
+        friend ParallelProcessor;
+    public:
+        explicit Worker(ParallelProcessor& dispatcher) : dispatcher_(dispatcher) {
+            dispatcher_.registeredWorkers_.fetch_add(1, std::memory_order_relaxed);
+            RuntimeLogDebug({ "balancing" }, "Worker registered");
+        }
+
+        ALWAYS_INLINE bool localEmpty() const noexcept {
+            return batch_.empty() && overflowList_.empty();
+        }
+
+        ALWAYS_INLINE bool tryPushLocal(typename ListImpl::reference value) noexcept {
+            return overflowList_.try_push_front(value);
+        }
+
+        ALWAYS_INLINE typename ListImpl::pointer tryPopLocal() noexcept {
+            return overflowList_.try_pop_front();
+        }
+
+        ALWAYS_INLINE bool tryPush(typename ListImpl::reference value) noexcept {
+            if (batch_.full()) {
+                bool released = dispatcher_.releaseBatch(std::move(batch_));
+                if (!released) {
+                    RuntimeLogDebug({ "balancing" }, "Batches pool overflow");
+                    batch_.transferAllInto(overflowList_);
+                }
+                batch_ = Batch{};
+            }
+            return batch_.tryPush(value);
+        }
+
+        ALWAYS_INLINE typename ListImpl::pointer tryPop() noexcept {
+            if (batch_.empty()) {
+                while (true) {
+                    bool acquired = dispatcher_.acquireBatch(batch_);
+                    if (!acquired) {
+                        if (!overflowList_.empty()) {
+                            batch_.fillFrom(overflowList_);
+                            RuntimeLogDebug({ "balancing" }, "Acquired %zu elements from the overflow list", batch_.elementsCount());
+                        } else {
+                            bool newWorkAvailable = waitForMoreWork();
+                            if (newWorkAvailable) continue;
+                            return nullptr;
+                        }
+                    }
+                    RuntimeAssert(!batch_.empty(), "Must have acquired some elements");
+                    break;
+                }
+            }
+
+            return batch_.tryPop();
+        }
+
+    private:
+        bool waitForMoreWork() noexcept {
+            RuntimeAssert(batch_.empty(), "Local batch must be depleted before waiting for shared work");
+            RuntimeAssert(overflowList_.empty(), "Local overflow list must be depleted before waiting for shared work");
+
+            std::unique_lock lock(dispatcher_.waitMutex_);
+
+            auto nowWaiting = dispatcher_.waitingWorkers_.fetch_add(1, std::memory_order_relaxed) + 1;
+            RuntimeLogDebug({ "balancing" }, "Worker goes to sleep (now sleeping %zu of %zu)",
+                            nowWaiting, dispatcher_.registeredWorkers_.load(std::memory_order_relaxed));
+
+            if (dispatcher_.allDone_) {
+                dispatcher_.waitingWorkers_.fetch_sub(1, std::memory_order_relaxed);
+                return false;
+            }
+
+            if (nowWaiting == dispatcher_.registeredWorkers_.load(std::memory_order_relaxed)) {
+                // we are the last ones awake
+                RuntimeLogDebug({ "balancing" }, "Worker has detected termination");
+                dispatcher_.allDone_ = true;
+                lock.unlock();
+                dispatcher_.waitCV_.notify_all();
+                dispatcher_.waitingWorkers_.fetch_sub(1, std::memory_order_relaxed);
+                return false;
+            }
+
+            dispatcher_.waitCV_.wait(lock);
+            dispatcher_.waitingWorkers_.fetch_sub(1, std::memory_order_relaxed);
+            if (dispatcher_.allDone_) {
+                return false;
+            }
+            RuntimeLogDebug({ "balancing" }, "Worker woke up");
+
+            return true;
+        }
+
+        ParallelProcessor& dispatcher_;
+
+        Batch batch_;
+        ListImpl overflowList_;
+    };
+
+    ParallelProcessor() = default;
+
+    ~ParallelProcessor() {
+        RuntimeAssert(waitingWorkers_.load() == 0, "All the workers must terminate before dispatcher destruction");
+    }
+
+    size_t registeredWorkers() {
+        return registeredWorkers_.load(std::memory_order_relaxed);
+    }
+
+private:
+    bool releaseBatch(Batch&& batch) {
+        RuntimeAssert(!batch.empty(), "A batch to release into shared pool must be non-empty");
+        RuntimeLogDebug({ "balancing" }, "Releasing batch of %zu elements", batch.elementsCount());
+        bool shared = sharedBatches_.enqueue(std::move(batch));
+        if (shared) {
+            if (waitingWorkers_.load(std::memory_order_relaxed) > 0) {
+                waitCV_.notify_one();
+            }
+        }
+        return shared;
+    }
+
+    bool acquireBatch(Batch& dst) {
+        RuntimeAssert(dst.empty(), "Destination batch must be already depleted");
+        auto acquired = sharedBatches_.dequeue();
+        if (acquired) {
+            dst = std::move(*acquired);
+            RuntimeLogDebug({ "balancing" }, "Acquired a batch of %zu elements", dst.elementsCount());
+            return true;
+        }
+        return false;
+    }
+
+    BoundedQueue<Batch, kBatchesPoolSize> sharedBatches_;
+
+    std::atomic<size_t> registeredWorkers_ = 0;
+    std::atomic<size_t> waitingWorkers_ = 0;
+
+    std::atomic<bool> allDone_ = false;
+    mutable std::mutex waitMutex_;
+    mutable std::condition_variable waitCV_;
+};
+
+}

kotlin-native/runtime/src/main/cpp/ParallelProcessorTest.cpp

@@ -0,0 +1,115 @@
+/*
+ * 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 "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+#include <list>
+
+#include "IntrusiveList.hpp"
+
+#include "ParallelProcessor.hpp"
+
+#include "std_support/Vector.hpp"
+#include "SingleThreadExecutor.hpp"
+#include "TestSupport.hpp"
+
+using ::testing::_;
+using namespace kotlin;
+
+namespace {
+
+struct Task {
+    template <typename WorkList>
+    static void workLoop(WorkList& workList) {
+        while (Task* task = workList.tryPop()) {
+            RuntimeAssert(!task->done_.load(), "Tasks are not idempotent");
+            task->done_ = true;
+        }
+    }
+
+    Task* next() const noexcept { return next_; }
+
+    void setNext(Task* next) noexcept {
+        RuntimeAssert(next, "next cannot be nullptr");
+        next_ = next;
+    }
+
+    bool trySetNext(Task* next) noexcept {
+        RuntimeAssert(next, "next cannot be nullptr");
+        if (next_ == nullptr) {
+            next_ = next;
+            return true;
+        }
+        return false;
+    }
+
+    std::atomic<bool> done_ = false;
+    Task* next_ = nullptr;
+};
+
+auto workBatch(std::size_t size) {
+    std::list<Task> batch;
+    for (size_t i = 0; i < size; ++i) {
+        batch.emplace_back();
+    }
+    return batch;
+}
+
+template <typename WorkList, typename Iterable>
+void offerWork(WorkList& wl, Iterable& batch) {
+    for (auto& task: batch) {
+        bool accepted = wl.tryPush(task);
+        RuntimeAssert(accepted, "Must be accepted");
+    }
+}
+
+using ListImpl = intrusive_forward_list<Task>;
+static constexpr auto kBatchSize = 256;
+using Processor = ParallelProcessor<ListImpl, kBatchSize, 1024>;
+using Worker = typename Processor::Worker;
+
+} // namespace
+
+
+TEST(ParallelProcessorTest, ContededRegistration) {
+    Processor processor;
+    std::vector<std::unique_ptr<Worker>> workers(kDefaultThreadCount);
+
+    std::atomic<bool> start = false;
+    std::list<ScopedThread> threads;
+    for (int i = 0; i < kDefaultThreadCount; ++i) {
+        threads.emplace_back([i, &start, &workers, &processor] {
+            while (!start.load()) {}
+            workers[i] = std::make_unique<Worker>(processor);
+        });
+    }
+
+    start = true;
+
+    for (auto& t : threads) {
+        t.join();
+    }
+
+    EXPECT_THAT(processor.registeredWorkers(), kDefaultThreadCount);
+
+    workers.clear();
+}
+
+TEST(ParallelProcessorTest, Sharing) {
+    Processor processor;
+    Worker giver(processor);
+    Worker taker(processor);
+
+    auto work = workBatch(kBatchSize * 2);
+    offerWork(giver, work);
+
+    EXPECT_TRUE(taker.localEmpty());
+
+    // have to steal from giver
+    EXPECT_NE(taker.tryPop(), nullptr);
+
+    EXPECT_FALSE(taker.localEmpty());
+}

kotlin-native/runtime/src/main/cpp/Utils.hpp

@@ -56,6 +56,22 @@ protected:
     ~Pinned() = default;
 };
 
+// A helper that executes the action provided upon destruction of the ScopeGuard instance.
+template<typename FinalAction>
+class ScopeGuard final : private Pinned {
+public:
+    template<typename InitAction>
+    ScopeGuard(InitAction initAction, FinalAction finalAction) noexcept : finalAction_(finalAction) {
+        initAction();
+    }
+    ScopeGuard(FinalAction finalAction) noexcept : finalAction_(finalAction) {}
+    ~ScopeGuard() noexcept {
+        finalAction_();
+    }
+private:
+    FinalAction finalAction_;
+};
+
 // A helper that scopley assigns a value to a variable. The variable will
 // be set to its original value upon destruction of the AutoReset instance.
 // Note that an AutoReset instance must have a shorter lifetime than

kotlin-native/runtime/src/mm/cpp/ThreadSuspension.cpp

@@ -25,6 +25,11 @@ namespace {
 
 } // namespace
 
+bool kotlin::mm::isSuspendedOrNative(kotlin::mm::ThreadData& thread) noexcept {
+    auto& suspensionData = thread.suspensionData();
+    return suspensionData.suspended() || suspensionData.state() == kotlin::ThreadState::kNative;
+}
+
 std::atomic<bool> kotlin::mm::internal::gSuspensionRequested = false;
 
 kotlin::ThreadState kotlin::mm::ThreadSuspensionData::setState(kotlin::ThreadState newState) noexcept {
@@ -45,10 +50,10 @@ kotlin::ThreadState kotlin::mm::ThreadSuspensionData::setState(kotlin::ThreadSta
 
 NO_EXTERNAL_CALLS_CHECK void kotlin::mm::ThreadSuspensionData::suspendIfRequested() noexcept {
     if (IsThreadSuspensionRequested()) {
+        auto suspendStartMs = konan::getTimeMicros();
         threadData_.gc().OnSuspendForGC();
         std::unique_lock lock(gSuspensionMutex);
         auto threadId = konan::currentThreadId();
-        auto suspendStartMs = konan::getTimeMicros();
         RuntimeLogDebug({kTagGC, kTagMM}, "Suspending thread %d", threadId);
         AutoReset scopedAssignSuspended(&suspended_, true);
         gSuspensionCondVar.wait(lock, []() { return !IsThreadSuspensionRequested(); });

kotlin-native/runtime/src/mm/cpp/ThreadSuspension.hpp

@@ -52,6 +52,8 @@ private:
 bool RequestThreadsSuspension() noexcept;
 void WaitForThreadsSuspension() noexcept;
 
+bool isSuspendedOrNative(kotlin::mm::ThreadData& thread) noexcept;
+
 /**
  * Suspends all threads registered in ThreadRegistry except threads that are in the Native state.
  * Blocks until all such threads are suspended. Threads that are in the Native state on the moment

kotlin-native/runtime/src/test_support/cpp/CompilerGenerated.cpp

@@ -75,6 +75,12 @@ extern const int32_t Kotlin_disableMmap = 0;
 #endif
 extern const char* const Kotlin_runtimeLogs = nullptr;
 extern const int32_t Kotlin_concurrentWeakSweep = 1;
+#if KONAN_WINDOWS
+// parallel mark tests hang on mingw due to (presumably) a bug in winpthread
+extern const int32_t Kotlin_gcMarkSingleThreaded = 1;
+#else
+extern const int32_t Kotlin_gcMarkSingleThreaded = 0;
+#endif
 extern const int32_t Kotlin_freezingChecksEnabled = 1;
 extern const int32_t Kotlin_freezingEnabled = 1;