diff --git a/kotlin-native/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/BinaryOptions.kt b/kotlin-native/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/BinaryOptions.kt index b7c51b46932..306567cc488 100644 --- a/kotlin-native/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/BinaryOptions.kt +++ b/kotlin-native/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/BinaryOptions.kt @@ -47,6 +47,8 @@ object BinaryOptions : BinaryOptionRegistry() { val concurrentWeakSweep by booleanOption() + val concurrentMarkMaxIterations by uintOption() + val gcMutatorsCooperate by booleanOption() val auxGCThreads by uintOption() diff --git a/kotlin-native/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/KonanConfig.kt b/kotlin-native/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/KonanConfig.kt index bbaaf26aec3..a26a9c4dbca 100644 --- a/kotlin-native/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/KonanConfig.kt +++ b/kotlin-native/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/KonanConfig.kt @@ -197,6 +197,9 @@ class KonanConfig(val project: Project, val configuration: CompilerConfiguration val concurrentWeakSweep: Boolean get() = configuration.get(BinaryOptions.concurrentWeakSweep) ?: true + val concurrentMarkMaxIterations: UInt + get() = configuration.get(BinaryOptions.concurrentMarkMaxIterations) ?: 100U + val gcMutatorsCooperate: Boolean by lazy { val mutatorsCooperate = configuration.get(BinaryOptions.gcMutatorsCooperate) if (gcMarkSingleThreaded) { diff --git a/kotlin-native/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/llvm/IrToBitcode.kt b/kotlin-native/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/llvm/IrToBitcode.kt index fb1a7d42060..28b30643480 100644 --- a/kotlin-native/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/llvm/IrToBitcode.kt +++ b/kotlin-native/backend.native/compiler/ir/backend.native/src/org/jetbrains/kotlin/backend/konan/llvm/IrToBitcode.kt @@ -2795,6 +2795,7 @@ internal class CodeGeneratorVisitor( overrideRuntimeGlobal("Kotlin_destroyRuntimeMode", llvm.constInt32(context.config.destroyRuntimeMode.value)) overrideRuntimeGlobal("Kotlin_gcMutatorsCooperate", llvm.constInt32(if (context.config.gcMutatorsCooperate) 1 else 0)) overrideRuntimeGlobal("Kotlin_auxGCThreads", llvm.constInt32(context.config.auxGCThreads.toInt())) + overrideRuntimeGlobal("Kotlin_concurrentMarkMaxIterations", llvm.constInt32(context.config.concurrentMarkMaxIterations.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) { diff --git a/kotlin-native/runtime/src/gc/cms/cpp/Barriers.cpp b/kotlin-native/runtime/src/gc/cms/cpp/Barriers.cpp index 461c48a94e1..29105fe5b2f 100644 --- a/kotlin-native/runtime/src/gc/cms/cpp/Barriers.cpp +++ b/kotlin-native/runtime/src/gc/cms/cpp/Barriers.cpp @@ -14,33 +14,78 @@ using namespace kotlin; -inline constexpr auto kTagBarriers = logging::Tag::kBarriers; -#define BarriersLogDebug(active, format, ...) RuntimeLogDebug({kTagBarriers}, "%s" format, active ? "[active] " : "", ##__VA_ARGS__) - namespace { -std::atomic markBarriersEnabled = false; +enum class BarriersPhase { + /** Normal execution */ + kDisabled, + /** During mark closure building */ + kMarkClosure, + /** After the mark closure is built, but before the mark completed (during weak ref processing) */ + kWeakProcessing +}; + +const char* toString(BarriersPhase barriersPhase) { + switch (barriersPhase) { + case BarriersPhase::kDisabled: + return "none"; + case BarriersPhase::kMarkClosure: + return "mark"; + case BarriersPhase::kWeakProcessing: + return "weak-processing"; + } +} + +std::atomic barriersPhase = BarriersPhase::kDisabled; std::atomic markingEpoch = 0; +BarriersPhase currentPhase() noexcept { + return barriersPhase.load(std::memory_order_acquire); +} + +BarriersPhase currentPhaseRelaxed() noexcept { + return barriersPhase.load(std::memory_order_relaxed); +} + +ALWAYS_INLINE void assertPhase(BarriersPhase actual, BarriersPhase expected) noexcept { + RuntimeAssert(actual == expected, "Barriers phase: expected %s but observed %s", toString(expected), toString(actual)); +} + +ALWAYS_INLINE void assertPhase(BarriersPhase expected) noexcept { + assertPhase(currentPhaseRelaxed(), expected); +} + +ALWAYS_INLINE void assertPhaseNot(BarriersPhase expected) noexcept { + RuntimeAssert(currentPhaseRelaxed() != expected, "Barriers phase: phase %s not expected", toString(expected)); +} + +void switchPhase(BarriersPhase from, BarriersPhase to) noexcept { + auto prev = barriersPhase.exchange(to, std::memory_order_release); + assertPhase(prev, from); +} + +auto& markDispatcher() noexcept { + return mm::GlobalData::Instance().gc().impl().gc().mark(); +} + +inline constexpr auto kTagBarriers = logging::Tag::kBarriers; +#define BarriersLogDebug(phase, format, ...) RuntimeLogDebug({kTagBarriers}, "[%s]" format, toString(phase), ##__VA_ARGS__) + } // namespace void gc::barriers::BarriersThreadData::onThreadRegistration() noexcept { - if (markBarriersEnabled.load(std::memory_order_acquire)) { + if (currentPhase() != BarriersPhase::kDisabled) { startMarkingNewObjects(GCHandle::getByEpoch(markingEpoch.load(std::memory_order_relaxed))); } } -ALWAYS_INLINE void gc::barriers::BarriersThreadData::onSafePoint() noexcept {} - void gc::barriers::BarriersThreadData::startMarkingNewObjects(gc::GCHandle gcHandle) noexcept { - RuntimeAssert(markBarriersEnabled.load(std::memory_order_relaxed), - "New allocations marking may only be requested by mark barriers"); + assertPhaseNot(BarriersPhase::kDisabled); markHandle_ = gcHandle.mark(); } void gc::barriers::BarriersThreadData::stopMarkingNewObjects() noexcept { - RuntimeAssert(!markBarriersEnabled.load(std::memory_order_relaxed), - "New allocations marking could only been requested by mark barriers"); + assertPhase(BarriersPhase::kDisabled); markHandle_ = std::nullopt; } @@ -49,30 +94,32 @@ bool gc::barriers::BarriersThreadData::shouldMarkNewObjects() const noexcept { } ALWAYS_INLINE void gc::barriers::BarriersThreadData::onAllocation(ObjHeader* allocated) { - bool shouldMark = shouldMarkNewObjects(); - RuntimeAssert(shouldMark == markBarriersEnabled.load(std::memory_order_relaxed), - "New allocations marking must happen with and only with mark barriers"); - BarriersLogDebug(shouldMark, "Allocation %p", allocated); - if (shouldMark) { + BarriersLogDebug(currentPhaseRelaxed(), "Allocation %p", allocated); + if (shouldMarkNewObjects()) { auto& objectData = alloc::objectDataForObject(allocated); objectData.markUncontended(); markHandle_->addObject(); } } -void gc::barriers::enableMarkBarriers(int64_t epoch) noexcept { +void gc::barriers::enableBarriers(int64_t epoch) noexcept { auto mutators = mm::ThreadRegistry::Instance().LockForIter(); markingEpoch.store(epoch, std::memory_order_relaxed); - markBarriersEnabled.store(true, std::memory_order_release); - for (auto& mutator: mutators) { + switchPhase(BarriersPhase::kDisabled, BarriersPhase::kMarkClosure); + for (auto& mutator : mutators) { mutator.gc().impl().gc().barriers().startMarkingNewObjects(GCHandle::getByEpoch(epoch)); } } -void gc::barriers::disableMarkBarriers() noexcept { +void gc::barriers::switchToWeakProcessingBarriers() noexcept { + // TODO markDispatcher().assertWeakReadForbiden(); + switchPhase(BarriersPhase::kMarkClosure, BarriersPhase::kWeakProcessing); +} + +void gc::barriers::disableBarriers() noexcept { auto mutators = mm::ThreadRegistry::Instance().LockForIter(); - markBarriersEnabled.store(false, std::memory_order_release); - for (auto& mutator: mutators) { + switchPhase(BarriersPhase::kWeakProcessing, BarriersPhase::kDisabled); + for (auto& mutator : mutators) { mutator.gc().impl().gc().barriers().stopMarkingNewObjects(); } } @@ -82,12 +129,11 @@ namespace { // TODO decide whether it's really beneficial to NO_INLINE the slow path NO_INLINE void beforeHeapRefUpdateSlowPath(mm::DirectRefAccessor ref, ObjHeader* value) noexcept { auto prev = ref.load(); - BarriersLogDebug(true, "Write *%p <- %p (%p overwritten)", ref.location(), value, prev); if (prev != nullptr && prev->heap()) { // TODO Redundant if the destination object is black. // Yet at the moment there is now efficient way to distinguish black and gray objects. - // TODO perhaps it would be better to path the thread data from outside + // TODO perhaps it would be better to pass the thread data from outside auto& threadData = *mm::ThreadRegistry::Instance().CurrentThreadData(); auto& markQueue = *threadData.gc().impl().gc().mark().markQueue(); gc::mark::ConcurrentMark::MarkTraits::tryEnqueue(markQueue, prev); @@ -99,36 +145,59 @@ NO_INLINE void beforeHeapRefUpdateSlowPath(mm::DirectRefAccessor ref, ObjHeader* } // namespace ALWAYS_INLINE void gc::barriers::beforeHeapRefUpdate(mm::DirectRefAccessor ref, ObjHeader* value) noexcept { - if (__builtin_expect(markBarriersEnabled.load(std::memory_order_acquire), false)) { + auto phase = currentPhase(); + BarriersLogDebug(phase, "Write *%p <- %p (%p overwritten)", ref.location(), value, ref.load()); + if (__builtin_expect(phase == BarriersPhase::kMarkClosure, false)) { beforeHeapRefUpdateSlowPath(ref, value); - } else { - BarriersLogDebug(false, "Write *%p <- %p (%p overwritten)", ref.location(), value, ref.load()); } } namespace { -// TODO decide whether it's really beneficial to NO_INLINE the slow path +/** + * Before the mark closure is built, every weak read may resurrect a weakly-reachable object. + * Thus, the referent must be pushed in a mark queue, in case it wold be resureceted behind the mark front. + */ NO_INLINE void weakRefReadInMarkSlowPath(ObjHeader* weakReferee) noexcept { + assertPhase(BarriersPhase::kMarkClosure); auto& threadData = *mm::ThreadRegistry::Instance().CurrentThreadData(); auto& markQueue = *threadData.gc().impl().gc().mark().markQueue(); gc::mark::ConcurrentMark::MarkTraits::tryEnqueue(markQueue, weakReferee); } +/** After the mark closure is built, but weak refs are not yet nulled out, every weak read shouuld check if the weak referent is marked. */ +NO_INLINE ObjHeader* weakRefReadInWeakSweepSlowPath(ObjHeader* weakReferee) noexcept { + assertPhase(BarriersPhase::kWeakProcessing); + if (!gc::isMarked(weakReferee)) { + return nullptr; + } + return weakReferee; +} + } // namespace +ALWAYS_INLINE ObjHeader* gc::barriers::weakRefReadBarrier(std::atomic& weakReferee) noexcept { + if (__builtin_expect(currentPhase() != BarriersPhase::kDisabled, false)) { + // Mark dispatcher requires weak reads be protected by the following: + auto weakReadProtector = markDispatcher().weakReadProtector(); -ALWAYS_INLINE OBJ_GETTER(gc::barriers::weakRefReadBarrier, std::atomic& weakReferee) noexcept { - // TODO be careful with atomics when conucrrent weak sweep is supported - auto weak = weakReferee.load(std::memory_order_relaxed); - if (!weak) return nullptr; - bool mark = markBarriersEnabled.load(std::memory_order_acquire); - if (__builtin_expect(mark, false)) { - BarriersLogDebug(true, "[mark] Weak read %p", weak); - weakRefReadInMarkSlowPath(weak); - } else { - // TODO reintroduce after-mark barriers that check mark bit like in PMCS, when concurrent weak sweep is supported - BarriersLogDebug(false, "Weak read %p", weak); + auto weak = weakReferee.load(std::memory_order_relaxed); + if (!weak) return nullptr; + + auto phase = currentPhase(); + BarriersLogDebug(phase, "Weak read %p", weak); + + if (__builtin_expect(phase == BarriersPhase::kMarkClosure, false)) { + weakRefReadInMarkSlowPath(weak); + } else { + if (__builtin_expect(phase == BarriersPhase::kWeakProcessing, false)) { + // TODO reread the referee here under the barrier guard + // if `disableBarriers` would be possible outside of STW + return weakRefReadInWeakSweepSlowPath(weakReferee); + } + } + return weak; } - return weak; + + return weakReferee.load(std::memory_order_relaxed); } diff --git a/kotlin-native/runtime/src/gc/cms/cpp/Barriers.hpp b/kotlin-native/runtime/src/gc/cms/cpp/Barriers.hpp index 238f50c105b..00e2e0eb54d 100644 --- a/kotlin-native/runtime/src/gc/cms/cpp/Barriers.hpp +++ b/kotlin-native/runtime/src/gc/cms/cpp/Barriers.hpp @@ -18,23 +18,24 @@ namespace kotlin::gc::barriers { class BarriersThreadData : private Pinned { public: void onThreadRegistration() noexcept; - void onSafePoint() noexcept; void startMarkingNewObjects(GCHandle gcHandle) noexcept; void stopMarkingNewObjects() noexcept; bool shouldMarkNewObjects() const noexcept; void onAllocation(ObjHeader* allocated); + private: std::optional markHandle_{}; }; // Must be called during STW. -void enableMarkBarriers(int64_t epoch) noexcept; -void disableMarkBarriers() noexcept; +void enableBarriers(int64_t epoch) noexcept; +void switchToWeakProcessingBarriers() noexcept; +void disableBarriers() noexcept; void beforeHeapRefUpdate(mm::DirectRefAccessor ref, ObjHeader* value) noexcept; -OBJ_GETTER(weakRefReadBarrier, std::atomic& weakReferee) noexcept; +ObjHeader* weakRefReadBarrier(std::atomic& weakReferee) noexcept; -} // namespace kotlin::gc +} // namespace kotlin::gc::barriers diff --git a/kotlin-native/runtime/src/gc/cms/cpp/BarriersTest.cpp b/kotlin-native/runtime/src/gc/cms/cpp/BarriersTest.cpp index 0d6e21d3125..fda600246ee 100644 --- a/kotlin-native/runtime/src/gc/cms/cpp/BarriersTest.cpp +++ b/kotlin-native/runtime/src/gc/cms/cpp/BarriersTest.cpp @@ -44,7 +44,6 @@ test_support::Object& AllocateObject(mm::ThreadData& threadData) { class BarriersTest : public testing::Test { public: - ~BarriersTest() override { mm::SpecialRefRegistry::instance().clearForTests(); mm::GlobalData::Instance().allocator().clearForTests(); @@ -74,7 +73,7 @@ TEST_F(BarriersTest, Deletion) { { ThreadStateGuard guard(ThreadState::kNative); // pretend to be the GC thread - gc::barriers::enableMarkBarriers(gcHandle.getEpoch()); + gc::barriers::enableBarriers(gcHandle.getEpoch()); } UpdateHeapRef(&ref, newObj.header()); @@ -84,14 +83,14 @@ TEST_F(BarriersTest, Deletion) { { ThreadStateGuard guard(ThreadState::kNative); // pretend to be the GC thread - gc::barriers::disableMarkBarriers(); + gc::barriers::switchToWeakProcessingBarriers(); + gc::barriers::disableBarriers(); } - }); } TEST_F(BarriersTest, AllocationDuringMarkBarreirs) { - gc::barriers::enableMarkBarriers(gcHandle.getEpoch()); + gc::barriers::enableBarriers(gcHandle.getEpoch()); RunInNewThread([this](mm::ThreadData& threadData) { initMutatorMarkQueue(threadData); @@ -99,7 +98,8 @@ TEST_F(BarriersTest, AllocationDuringMarkBarreirs) { EXPECT_THAT(gc::isMarked(obj.header()), true); }); - gc::barriers::disableMarkBarriers(); + gc::barriers::switchToWeakProcessingBarriers(); + gc::barriers::disableBarriers(); } TEST_F(BarriersTest, ConcurrentDeletion) { @@ -118,7 +118,7 @@ TEST_F(BarriersTest, ConcurrentDeletion) { std::atomic canStart = false; std::atomic finished = 0; - gc::barriers::enableMarkBarriers(gcHandle.getEpoch()); + gc::barriers::enableBarriers(gcHandle.getEpoch()); std::vector threads; for (int i = 0; i < kDefaultThreadCount; ++i) { @@ -148,7 +148,8 @@ TEST_F(BarriersTest, ConcurrentDeletion) { std::this_thread::yield(); } - gc::barriers::disableMarkBarriers(); + gc::barriers::switchToWeakProcessingBarriers(); + gc::barriers::disableBarriers(); EXPECT_THAT(gc::isMarked(ref), true); } diff --git a/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMark.cpp b/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMark.cpp index e4e31db2e85..26e8f09a72f 100644 --- a/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMark.cpp +++ b/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMark.cpp @@ -12,6 +12,20 @@ using namespace kotlin; +namespace { + +class FlushActionActivator final : public mm::ExtraSafePointActionActivator {}; + +} // namespace + +void gc::mark::ConcurrentMark::ThreadData::ensureFlushActionExecuted() noexcept { + flushAction_->ensureExecuted([this] { markQueue()->forceFlush(); }); +} + +void gc::mark::ConcurrentMark::ThreadData::onSafePoint() noexcept { + FlushActionActivator::doIfActive([this] { ensureFlushActionExecuted(); }); +} + void gc::mark::ConcurrentMark::beginMarkingEpoch(gc::GCHandle gcHandle) { gcHandle_ = gcHandle; @@ -31,62 +45,66 @@ void gc::mark::ConcurrentMark::runMainInSTW() { GCLogDebug(gcHandle().getEpoch(), "Creating main (#0) mark worker"); // create mutator mark queues - for (auto& thread: *lockedMutatorsList_) { + for (auto& thread : *lockedMutatorsList_) { thread.gc().impl().gc().mark().markQueue().construct(*parallelProcessor_); } completeMutatorsRootSet(mainWorker); // global root set must be collected after all the mutator's global data have been published - collectRootSetGlobals (gcHandle(), mainWorker); - - barriers::enableMarkBarriers(gcHandle().getEpoch()); + collectRootSetGlobals(gcHandle(), mainWorker); + barriers::enableBarriers(gcHandle().getEpoch()); resumeTheWorld(gcHandle()); - // build mark closure - parallelMark(mainWorker); - - // TODO resume the world much later when the mark closure is completed - stopTheWorld(gcHandle(), "GC stop the world #2: complete mark closure"); - - barriers::disableMarkBarriers(); - - bool refsRemainInMutatorQueues = false; + // Mutator threads might release their internal batch at a pretty arbitrary moment (during a barrier execution with overflow). + // So there are not so many reliable ways to track releases of new work. + // The number of batches sharad inside a parallel processor may only grow, + // we use this number to decide when to finish the mark. + auto everSharedBatches = parallelProcessor_->batchesEverShared(); + size_t iter = 0; + bool terminateInSTW = false; do { - for (auto& mutator: *lockedMutatorsList_) { - const bool markQueueNowEmpty = mutator.gc().impl().gc().mark().markQueue()->forceFlush(); - if (!markQueueNowEmpty) { - refsRemainInMutatorQueues = true; - } + GCLogDebug(gcHandle().getEpoch(), "Building mark closure (attempt #%zu)", iter); + Mark(gcHandle(), mainWorker); + + RuntimeCheck(iter <= compiler::concurrentMarkMaxIterations(), "Failed to terminate mark in STW in a single iteration"); + ++iter; + if (iter == compiler::concurrentMarkMaxIterations()) { + fprintf(stderr, "EMERGENCY MARK TERMINATION\n"); + GCLogWarning(gcHandle().getEpoch(), "Finishing mark closure in STW after (%zu concurrent attempts)", iter); + stopTheWorld(gcHandle(), "GC stop the world #2: concurrent mark took too long"); + terminateInSTW = true; } + } while (!tryTerminateMark(everSharedBatches)); - parallelProcessor_->resetForNewWork(); - // complete mark closure form newly found objects - parallelMark(mainWorker); - } while (refsRemainInMutatorQueues); - - for (auto& thread: *lockedMutatorsList_) { - auto& markQueue = thread.gc().impl().gc().mark().markQueue(); - RuntimeAssert(markQueue->retainsNoWork(), ""); // TODO move into queue's destuctor? - markQueue.destroy(); + for (auto& thread : *lockedMutatorsList_) { + thread.gc().impl().gc().mark().markQueue().destroy(); } + + endMarkingEpoch(); + + gc::processWeaks(gcHandle(), mm::SpecialRefRegistry::instance()); + + if (!terminateInSTW) { + stopTheWorld(gcHandle(), "GC stop the world #2: prepare to sweep"); + } + + barriers::disableBarriers(); } void gc::mark::ConcurrentMark::runOnMutator(mm::ThreadData&) { // no-op } - gc::GCHandle& gc::mark::ConcurrentMark::gcHandle() { RuntimeAssert(gcHandle_.isValid(), "GCHandle must be initialized"); return gcHandle_; } - void gc::mark::ConcurrentMark::completeMutatorsRootSet(MarkTraits::MarkQueue& markQueue) { // workers compete for mutators to collect their root set - for (auto& thread: *lockedMutatorsList_) { + for (auto& thread : *lockedMutatorsList_) { tryCollectRootSet(thread, markQueue); } } @@ -97,16 +115,70 @@ void gc::mark::ConcurrentMark::tryCollectRootSet(mm::ThreadData& thread, MarkTra GCLogDebug(gcHandle().getEpoch(), "Root set collection on thread %d for thread %d", konan::currentThreadId(), thread.threadId()); gcData.publish(); - collectRootSetForThread (gcHandle(), markQueue, thread); + collectRootSetForThread(gcHandle(), markQueue, thread); } -void gc::mark::ConcurrentMark::parallelMark(ParallelProcessor::Worker& worker) { - GCLogDebug(gcHandle().getEpoch(), "Mark loop has begun"); - Mark (gcHandle(), worker); +/** Terminates the mark loop if possible, otherwise returns `false`. */ +bool gc::mark::ConcurrentMark::tryTerminateMark(std::size_t& everSharedBatches) noexcept { + // prevent unwanted mutations (such as weak-reachable resurrection) during termination detection + std::unique_lock markTerminationGuard(markTerminationMutex_); + + // has to happen under the termination lock guard + flushMutatorQueues(); + + // After the mutators have been forced to flush their local queues, + // there is only on possibility for this counter to remain the same as on a previous iteration: + // 1. Mutator local queues are empty, + // 2. AND were empty before the flush request was made, + // 3. AND the last attempt at completing mark closure encountered 0 new objects // FIXME this is actually redundant + const auto nowSharedBatches = parallelProcessor_->batchesEverShared(); + if (nowSharedBatches > everSharedBatches) { + everSharedBatches = nowSharedBatches; + parallelProcessor_->resetForNewWork(); + return false; + } + RuntimeAssert(nowSharedBatches == everSharedBatches, "This number must decrease"); + + barriers::switchToWeakProcessingBarriers(); + return true; +} + +void gc::mark::ConcurrentMark::flushMutatorQueues() noexcept { + for (auto& mutator : *lockedMutatorsList_) { + mutator.gc().impl().gc().mark().flushAction_.construct(); + } + + { + FlushActionActivator flushActivator{}; + + // wait all mutators flushed + while (true) { + bool allDone = true; + for (auto& mutator : *lockedMutatorsList_) { + auto& markData = mutator.gc().impl().gc().mark(); + if (mutator.suspensionData().suspendedOrNative()) { + markData.ensureFlushActionExecuted(); + } else if (!markData.flushAction_->executed()) { + allDone = false; + } + } + if (allDone) break; + std::this_thread::yield(); + } + } + + // It's guaranteed by the activator that no mutator thread would access somethingFlushed_ at this point. + for (auto& mutator : *lockedMutatorsList_) { + mutator.gc().impl().gc().mark().flushAction_.destroy(); + } } void gc::mark::ConcurrentMark::resetMutatorFlags() { - for (auto& mut: *lockedMutatorsList_) { + for (auto& mut : *lockedMutatorsList_) { mut.gc().impl().gc().clearMarkFlags(); } } + +bool gc::mark::test_support::flushActionRequested() { + return FlushActionActivator::isActive(); +} diff --git a/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMark.hpp b/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMark.hpp index 61a43e99e2e..07872c4e91b 100644 --- a/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMark.hpp +++ b/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMark.hpp @@ -1,5 +1,5 @@ /* - * Copyright 2010-2023 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license + * Copyright 2010-2024 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. */ @@ -13,8 +13,11 @@ #include "ObjectData.hpp" #include "ParallelProcessor.hpp" #include "SafePoint.hpp" +#include "ThreadData.hpp" #include "ThreadRegistry.hpp" #include "Utils.hpp" +#include "concurrent/Once.hpp" + namespace kotlin::gc::mark { @@ -29,12 +32,14 @@ namespace kotlin::gc::mark { * as only references from mark phase beginning matter for the SatB approach. * - Read barrier for weak refs: * Remembers each object read via a weak reference in a thread-local mark queue. - * Prevents the possibility of inserting a strong reference to a ewakly-reachable object behind the mark front. - * 3. Concurrently, the marker thread builds a mark closure. Unmarked objects hidden in a mutator mark queues may still exist. - * 4. Pause mutators once more, drain local mark queues, and complete the mark closure, this time non-concurrently. - * // TODO build closure fully concurrent - * 5. Process and clean weak references. // TODO process weak refs concurrently - * 6. Prepare mared heap for sweeping and resume mutation. // TODO prepare heap without a pause + * Prevents the possibility of inserting a strong reference to a weakly-reachable object behind the mark front. + * 3. Concurrently, the marker thread builds a mark closure. + * From time to time the mutator threads flush their mark queues into the global one. + * 4. In case the mark closure is complete, replace the remembering weak read barrier with + * the barrier that hides unmarked (dead) referents. + * 5. Process and clean weak references. + * 6. Pause mutators once again and disable all the barreirs. + * 7. Prepare marked heap for sweeping and resume mutation. // TODO prepare heap without a pause */ class ConcurrentMark : private Pinned { using MarkStackImpl = intrusive_forward_list; @@ -51,9 +56,7 @@ public: static constexpr auto kAllowHeapToStackRefs = false; - static void clear(AnyQueue& queue) noexcept { - RuntimeAssert(queue.localEmpty(), "Mark queue must be empty"); - } + static void clear(AnyQueue& queue) noexcept { RuntimeAssert(queue.localEmpty(), "Mark queue must be empty"); } static ALWAYS_INLINE ObjHeader* tryDequeue(MarkQueue& queue) noexcept { auto* obj = queue.tryPop(); @@ -91,10 +94,17 @@ public: }; class ThreadData : private Pinned { + friend ConcurrentMark; + public: auto& markQueue() noexcept { return markQueue_; } + void onSafePoint() noexcept; + private: + void ensureFlushActionExecuted() noexcept; + ManuallyScoped markQueue_{}; + ManuallyScoped flushAction_{}; }; void beginMarkingEpoch(GCHandle gcHandle); @@ -109,17 +119,33 @@ public: */ void runOnMutator(mm::ThreadData& mutatorThread); + /** + * Weak reference reads may be mutually exclusive with certain parts of mark oprocess. + * Every read must be guarded by the object returned by this method. + */ + auto weakReadProtector() noexcept { + return std::pair{ThreadStateGuard{ThreadState::kNative}, std::shared_lock{markTerminationMutex_}}; + } + private: GCHandle& gcHandle(); void completeMutatorsRootSet(MarkTraits::MarkQueue& markQueue); void tryCollectRootSet(mm::ThreadData& thread, ParallelProcessor::Worker& markQueue); - void parallelMark(ParallelProcessor::Worker& worker); + bool tryTerminateMark(std::size_t& everSharedBatches) noexcept; + void flushMutatorQueues() noexcept; void resetMutatorFlags(); GCHandle gcHandle_ = GCHandle::invalid(); std::optional lockedMutatorsList_; ManuallyScoped parallelProcessor_{}; + + RWSpinLock markTerminationMutex_; }; + +namespace test_support { +bool flushActionRequested(); +} + } // namespace kotlin::gc::mark diff --git a/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.cpp b/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.cpp index 1a973f71918..ec36a2560fa 100644 --- a/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.cpp +++ b/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.cpp @@ -24,7 +24,7 @@ namespace { [[clang::no_destroy]] std::mutex gcMutex; -template +template ScopedThread createGCThread(const char* name, Body&& body) { return ScopedThread(ScopedThread::attributes().name(name), [name, body] { RuntimeLogDebug({kTagGC}, "%s %d starts execution", name, konan::currentThreadId()); @@ -37,7 +37,7 @@ ScopedThread createGCThread(const char* name, Body&& body) { // TODO move to common [[maybe_unused]] inline void checkMarkCorrectness(alloc::ObjectFactoryImpl::Iterable& heap) { if (compiler::runtimeAssertsMode() == compiler::RuntimeAssertsMode::kIgnore) return; - for (auto objRef: heap) { + for (auto objRef : heap) { auto obj = objRef.GetObjHeader(); auto& objData = objRef.ObjectData(); if (objData.marked()) { @@ -62,7 +62,8 @@ 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()); + RuntimeLogDebug( + {kTagGC}, "Thread %d have exclusively acquired thread %d's root set", konan::currentThreadId(), threadData_.threadId()); } return locked; } @@ -142,12 +143,6 @@ void gc::ConcurrentMarkAndSweep::PerformFullGC(int64_t epoch) noexcept { markDispatcher_.runMainInSTW(); - markDispatcher_.endMarkingEpoch(); - - // TODO re-enable concurrent weak sweep again - - gc::processWeaks(gcHandle, mm::SpecialRefRegistry::instance()); - // TODO outline as mark_.isolateMarkedHeapAndFinishMark() // By this point all the alive heap must be marked. // All the mutations (incl. allocations) after this method will be subject for the next GC. diff --git a/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.hpp b/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.hpp index cbf5e89709f..9a8db89967e 100644 --- a/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.hpp +++ b/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.hpp @@ -40,7 +40,7 @@ public: void OnSuspendForGC() noexcept; - void safePoint() noexcept { barriers_.onSafePoint(); } + void safePoint() noexcept { mark_.onSafePoint(); } void onThreadRegistration() noexcept { barriers_.onThreadRegistration(); } @@ -77,6 +77,8 @@ public: return mainThreadFinalizerProcessor_; } + auto& mark() noexcept { return markDispatcher_; } + private: void mainGCThreadBody(); void PerformFullGC(int64_t epoch) noexcept; diff --git a/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweepTest.cpp b/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweepTest.cpp index 64ed1cadcb9..07dea9a52a5 100644 --- a/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweepTest.cpp +++ b/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweepTest.cpp @@ -33,5 +33,82 @@ public: } // namespace -REGISTER_TYPED_TEST_SUITE_WITH_LISTS(TracingGCTest, TRACING_GC_TEST_LIST); +TYPED_TEST_P(TracingGCTest, WeakResurrectionAtMarkTermination) { + std::vector mutators(kDefaultThreadCount); + std::vector weaks(kDefaultThreadCount); + std::vector*> roots(kDefaultThreadCount); + + // initialize threads + for (int i = 0; i < kDefaultThreadCount; ++i) { + mutators[i] + .Execute([&, i](mm::ThreadData& threadData, Mutator& mutator) { + roots[i] = &AllocateObject(threadData); + mutator.AddGlobalRoot(roots[i]->header()); + + auto& weakReferee = AllocateObject(threadData); + auto& weakRef = [&threadData, &weakReferee]() -> test_support::RegularWeakReferenceImpl& { + ObjHolder holder; + return test_support::InstallWeakReference(threadData, weakReferee.header(), holder.slot()); + }(); + EXPECT_NE(weakRef.get(), nullptr); + weaks[i] = &weakRef; + mutator.AddGlobalRoot(weakRef.header()); + }) + .wait(); + } + + auto epoch = mm::GlobalData::Instance().gc().Schedule(); + std::atomic gcDone = false; + + // Spin until thread suspension is requested. + while (!mm::IsThreadSuspensionRequested()) { + } + + std::vector> mutatorFutures; + for (int i = 0; i < kDefaultThreadCount; ++i) { + mutatorFutures.emplace_back(mutators[i].Execute([&, i](mm::ThreadData& threadData, Mutator& mutator) noexcept { + while (!gc::mark::test_support::flushActionRequested() && !gcDone.load(std::memory_order_relaxed)) { + safePoint(threadData); + } + + threadData.gc().impl().gc().mark().onSafePoint(); + + auto weakReferee = weaks[i]->get(); + (*roots[i])->field2 = weakReferee; + bool resurrected = weakReferee != nullptr; + + while (!gcDone.load(std::memory_order_relaxed)) { + safePoint(threadData); + } + + if (resurrected) { + EXPECT_NE(weaks[i]->get(), nullptr); + } else { + EXPECT_EQ(weaks[i]->get(), nullptr); + } + })); + } + + mm::GlobalData::Instance().gc().WaitFinalizers(epoch); + gcDone = true; + + for (auto& future : mutatorFutures) { + future.wait(); + } + + for (int i = 0; i < kDefaultThreadCount; ++i) { + mutators[i] + .Execute([&, i](mm::ThreadData&, Mutator&) noexcept { + if (auto weakReferee = weaks[i]->get()) { + auto& extraObj = *mm::ExtraObjectData::Get(weakReferee); + extraObj.ClearRegularWeakReferenceImpl(); + extraObj.Uninstall(); + alloc::destroyExtraObjectData(extraObj); + } + }) + .wait(); + } +} + +REGISTER_TYPED_TEST_SUITE_WITH_LISTS(TracingGCTest, TRACING_GC_TEST_LIST, WeakResurrectionAtMarkTermination); INSTANTIATE_TYPED_TEST_SUITE_P(CMS, TracingGCTest, ConcurrentMarkAndSweepTest); diff --git a/kotlin-native/runtime/src/gc/cms/cpp/GCImpl.cpp b/kotlin-native/runtime/src/gc/cms/cpp/GCImpl.cpp index c3af42b518b..42dbd634f8b 100644 --- a/kotlin-native/runtime/src/gc/cms/cpp/GCImpl.cpp +++ b/kotlin-native/runtime/src/gc/cms/cpp/GCImpl.cpp @@ -92,7 +92,7 @@ ALWAYS_INLINE void gc::beforeHeapRefUpdate(mm::DirectRefAccessor ref, ObjHeader* } ALWAYS_INLINE OBJ_GETTER(gc::weakRefReadBarrier, std::atomic& weakReferee) noexcept { - RETURN_RESULT_OF(gc::barriers::weakRefReadBarrier, weakReferee); + RETURN_OBJ(gc::barriers::weakRefReadBarrier(weakReferee)); } bool gc::isMarked(ObjHeader* object) noexcept { diff --git a/kotlin-native/runtime/src/gc/common/cpp/TracingGCTest.hpp b/kotlin-native/runtime/src/gc/common/cpp/TracingGCTest.hpp index 563db0ea5ab..f160097e304 100644 --- a/kotlin-native/runtime/src/gc/common/cpp/TracingGCTest.hpp +++ b/kotlin-native/runtime/src/gc/common/cpp/TracingGCTest.hpp @@ -1174,7 +1174,7 @@ TYPED_TEST_P(TracingGCTest, MutateBetweenSafePoints) { } } -TYPED_TEST_P(TracingGCTest, WeakResuractionInMark) { +TYPED_TEST_P(TracingGCTest, WeakResurrectionInMark) { constexpr auto kObjectsPerThread = 100; std::vector mutators(kDefaultThreadCount); std::vector weaks(kDefaultThreadCount); @@ -1215,12 +1215,13 @@ TYPED_TEST_P(TracingGCTest, WeakResuractionInMark) { auto weakReferee = weaks[i]->get(); (*roots[i])->field2 = weakReferee; - bool resurected = weakReferee != nullptr; + bool resurrected = weakReferee != nullptr; while (!gcDone.load(std::memory_order_relaxed)) { safePoint(threadData); } - if (resurected) { + + if (resurrected) { EXPECT_NE(weaks[i]->get(), nullptr); } else { EXPECT_EQ(weaks[i]->get(), nullptr); @@ -1257,7 +1258,7 @@ TYPED_TEST_P(TracingGCTest, WeakResuractionInMark) { NewThreadsWhileRequestingCollection, \ FreeObjectWithFreeWeakReversedOrder, \ MutateBetweenSafePoints, \ - WeakResuractionInMark + WeakResurrectionInMark template class STWMarkGCTest : public TracingGCTest {}; diff --git a/kotlin-native/runtime/src/gc/pmcs/cpp/Barriers.cpp b/kotlin-native/runtime/src/gc/pmcs/cpp/Barriers.cpp index 54d7ed49b76..104efb3aebd 100644 --- a/kotlin-native/runtime/src/gc/pmcs/cpp/Barriers.cpp +++ b/kotlin-native/runtime/src/gc/pmcs/cpp/Barriers.cpp @@ -33,7 +33,23 @@ ObjHeader* weakRefBarrierImpl(ObjHeader* weakReferee) noexcept { NO_INLINE ObjHeader* weakRefReadSlowPath(std::atomic& weakReferee) noexcept { // reread an action to avoid register pollution outside the function auto barrier = weakRefBarrier.load(std::memory_order_seq_cst); + + // The referee must be reread here to avoid a possible race with the concurrent barrier disablement. + // NOTE: at the moment the barriers are disabled in STW, meaning this all is not the case. + // + // Consider the following situation: + // 1. GC thread enables the barriers and resumes mutators. + // 2. A mutator reads obj_ into local variable and sleeps. + // 3. GC thread finishes weak processing and disables the barriers. + // 4. The mutator continues executing, sees that there's no more barriers (but have not yet communicated this fact with the GC) + // and returns object from the local variable. + // Why reading inside the barrier code fixes it: + // 1. We read the barrier with seq_cst and do it before reading obj_. + // 2. If there's a barrier, we execute it, and the GC is definitely waiting for us to finish, + // and the object is still alive but is not marked (if it's marked, then there's nothing to go wrong) + // 3. If there's no barrier, then the GC has already cleaned up all the weaks and we read obj_ after that happening. auto* weak = weakReferee.load(std::memory_order_relaxed); + return barrier ? barrier(weak) : weak; } diff --git a/kotlin-native/runtime/src/main/cpp/BoundedQueue.hpp b/kotlin-native/runtime/src/main/cpp/BoundedQueue.hpp index 105b7f031a6..e8eaca060d4 100644 --- a/kotlin-native/runtime/src/main/cpp/BoundedQueue.hpp +++ b/kotlin-native/runtime/src/main/cpp/BoundedQueue.hpp @@ -93,6 +93,11 @@ public: return std::move(result); } + /** Returns the number of items ever added to the queue: both present in the queue and already dequeed. */ + size_t cumulativeThroughput() const noexcept { + return enqueuePos_.load(); + } + private: struct Cell { // TODO describe diff --git a/kotlin-native/runtime/src/main/cpp/CompilerConstants.cpp b/kotlin-native/runtime/src/main/cpp/CompilerConstants.cpp index 997fc664d7c..87a5f36aaa2 100644 --- a/kotlin-native/runtime/src/main/cpp/CompilerConstants.cpp +++ b/kotlin-native/runtime/src/main/cpp/CompilerConstants.cpp @@ -21,6 +21,7 @@ using Kotlin_getSourceInfo_FunctionType = int(*)(void * /*addr*/, SourceInfo* /* RUNTIME_WEAK int32_t Kotlin_destroyRuntimeMode = 1; RUNTIME_WEAK int32_t Kotlin_gcMutatorsCooperate = 0; RUNTIME_WEAK uint32_t Kotlin_auxGCThreads = 0; +RUNTIME_WEAK uint32_t Kotlin_concurrentMarkMaxIterations = 100; RUNTIME_WEAK int32_t Kotlin_workerExceptionHandling = 0; RUNTIME_WEAK int32_t Kotlin_suspendFunctionsFromAnyThreadFromObjC = 0; RUNTIME_WEAK Kotlin_getSourceInfo_FunctionType Kotlin_getSourceInfo_Function = nullptr; @@ -48,6 +49,10 @@ ALWAYS_INLINE uint32_t compiler::auxGCThreads() noexcept { return Kotlin_auxGCThreads; } +ALWAYS_INLINE uint32_t compiler::concurrentMarkMaxIterations() noexcept { + return Kotlin_concurrentMarkMaxIterations; +} + ALWAYS_INLINE compiler::WorkerExceptionHandling compiler::workerExceptionHandling() noexcept { return static_cast(Kotlin_workerExceptionHandling); } diff --git a/kotlin-native/runtime/src/main/cpp/CompilerConstants.hpp b/kotlin-native/runtime/src/main/cpp/CompilerConstants.hpp index 59b6bfea514..7b6676cff1c 100644 --- a/kotlin-native/runtime/src/main/cpp/CompilerConstants.hpp +++ b/kotlin-native/runtime/src/main/cpp/CompilerConstants.hpp @@ -107,6 +107,7 @@ WorkerExceptionHandling workerExceptionHandling() noexcept; DestroyRuntimeMode destroyRuntimeMode() noexcept; bool gcMutatorsCooperate() noexcept; uint32_t auxGCThreads() noexcept; +uint32_t concurrentMarkMaxIterations() noexcept; bool suspendFunctionsFromAnyThreadFromObjCEnabled() noexcept; AppStateTracking appStateTracking() noexcept; int getSourceInfo(void* addr, SourceInfo *result, int result_size) noexcept; diff --git a/kotlin-native/runtime/src/main/cpp/ParallelProcessor.hpp b/kotlin-native/runtime/src/main/cpp/ParallelProcessor.hpp index e8bab4a6536..4ce09eff0b3 100644 --- a/kotlin-native/runtime/src/main/cpp/ParallelProcessor.hpp +++ b/kotlin-native/runtime/src/main/cpp/ParallelProcessor.hpp @@ -65,6 +65,11 @@ private: elemsCount_ = spliced; } + void clear() noexcept { + elems_.clear(); + elemsCount_ = 0; + } + private: ListImpl elems_; std::size_t elemsCount_ = 0; @@ -94,6 +99,10 @@ public: public: explicit WorkSource(ParallelProcessor& dispatcher) : dispatcher_(dispatcher) {} + ~WorkSource() noexcept { + RuntimeAssert(retainsNoWork(), "A queue must be empty"); + } + ALWAYS_INLINE bool retainsNoWork() const noexcept { return batch_.empty() && overflowList().empty(); } @@ -136,6 +145,11 @@ public: } } + void clear() noexcept { + batch_.clear(); + overflowList().clear(); + } + protected: ListImpl& overflowList() noexcept { return this->localQueue_; @@ -222,7 +236,7 @@ public: RuntimeAssert(waitingWorkers_.load() == 0, "All the workers must terminate before dispatcher destruction"); } - size_t registeredWorkers() { + size_t registeredWorkers() const noexcept { return registeredWorkers_.load(std::memory_order_relaxed); } @@ -233,6 +247,11 @@ public: allDone_ = false; } + /** Returns a total cumulatiove number of bathces that have ever been shared by any of the work sources so far. */ + size_t batchesEverShared() const noexcept { + return sharedBatches_.cumulativeThroughput(); + } + private: bool releaseBatch(Batch&& batch) { RuntimeAssert(!batch.empty(), "A batch to release into shared pool must be non-empty"); diff --git a/kotlin-native/runtime/src/main/cpp/ParallelProcessorTest.cpp b/kotlin-native/runtime/src/main/cpp/ParallelProcessorTest.cpp index 2b4a30b32b2..125cc915e81 100644 --- a/kotlin-native/runtime/src/main/cpp/ParallelProcessorTest.cpp +++ b/kotlin-native/runtime/src/main/cpp/ParallelProcessorTest.cpp @@ -59,7 +59,7 @@ auto createWork(std::size_t size) { template void offerWork(WorkList& wl, Iterable& batch) { - for (auto& task: batch) { + for (auto& task : batch) { bool accepted = wl.tryPush(task); RuntimeAssert(accepted, "Must be accepted"); } @@ -74,7 +74,6 @@ using WorkSource = typename Processor::WorkSource; } // namespace - TEST(ParallelProcessorTest, ContededRegistration) { Processor processor; std::vector> workers(kDefaultThreadCount); @@ -83,7 +82,8 @@ TEST(ParallelProcessorTest, ContededRegistration) { std::list threads; for (int i = 0; i < kDefaultThreadCount; ++i) { threads.emplace_back([i, &start, &workers, &processor] { - while (!start.load()) {} + while (!start.load()) { + } workers[i] = std::make_unique(processor); }); } @@ -115,6 +115,8 @@ TEST(ParallelProcessorTest, Sharing) { EXPECT_NE(taker.tryPop(), nullptr); EXPECT_FALSE(taker.retainsNoWork()); + taker.clear(); + giver.clear(); } TEST(ParallelProcessorTest, SharingFromNonWorkerSource) { @@ -135,6 +137,8 @@ TEST(ParallelProcessorTest, SharingFromNonWorkerSource) { EXPECT_NE(taker.tryPop(), nullptr); EXPECT_FALSE(taker.retainsNoWork()); + taker.clear(); + giver.clear(); } TEST(ParallelProcessorTest, Overflow) {