[K/N] Concurrent phase in GC mark (KT-58865)

+ Log Stop The World reasons

Merge-request: KT-MR-13759
Merged-by: Alexey Glushko <aleksei.glushko@jetbrains.com>
This commit is contained in:
Aleksei.Glushko
2024-01-09 18:34:36 +00:00
committed by Space Team
parent f8fd46a5b8
commit 8ed806ebe0
35 changed files with 539 additions and 599 deletions
@@ -186,7 +186,7 @@ class KonanConfig(val project: Project, val configuration: CompilerConfiguration
} ?: arg
}
private val defaultGcMarkSingleThreaded get() = target.family == Family.MINGW
private val defaultGcMarkSingleThreaded get() = target.family == Family.MINGW && gc == GC.PARALLEL_MARK_CONCURRENT_SWEEP
val gcMarkSingleThreaded: Boolean by lazy {
configuration.get(BinaryOptions.gcMarkSingleThreaded) ?: defaultGcMarkSingleThreaded
@@ -203,6 +203,12 @@ class KonanConfig(val project: Project, val configuration: CompilerConfiguration
"Mutators cooperation is not supported during single threaded mark")
}
false
} else if (gc == GC.CONCURRENT_MARK_AND_SWEEP) {
if (mutatorsCooperate == true) {
configuration.report(CompilerMessageSeverity.STRONG_WARNING,
"Mutators cooperation is not yet supported in CMS GC")
}
false
} else {
mutatorsCooperate ?: true
}
@@ -30,6 +30,7 @@ enum class LoggingTag(val ord: Int) {
Alloc(6),
Balancing(7),
Barriers(8),
GCMark(9),
;
companion object {
@@ -481,7 +481,6 @@ internal class CodegenLlvmHelpers(private val generationState: NativeGenerationS
val Kotlin_processObjectInMark by lazyRtFunction
val Kotlin_processArrayInMark by lazyRtFunction
val Kotlin_processFieldInMark by lazyRtFunction
val Kotlin_processEmptyObjectInMark by lazyRtFunction
val UpdateVolatileHeapRef by lazyRtFunction
@@ -1828,6 +1828,10 @@ internal object EscapeAnalysis {
IntraproceduralAnalysis(context, moduleDFG, externalModulesDFG, callGraph).analyze()
InterproceduralAnalysis(context, generationState, callGraph, intraproceduralAnalysisResult, externalModulesDFG, lifetimes,
propagateExiledToHeapObjects = context.config.memoryModel != MemoryModel.EXPERIMENTAL
// The GC must be careful not to scan exiled objects, that have already became dead,
// as they may reference other already destroyed stack-allocated objects.
// TODO somehow tag these object, so that GC could handle them properly.
|| context.config.gc == GC.CONCURRENT_MARK_AND_SWEEP
).analyze()
} catch (t: Throwable) {
val extraUserInfo =
@@ -81,7 +81,6 @@ touchFunction(Kotlin_mm_safePointWhileLoopBody)
touchFunction(Kotlin_processObjectInMark)
touchFunction(Kotlin_processArrayInMark)
touchFunction(Kotlin_processFieldInMark)
touchFunction(Kotlin_processEmptyObjectInMark)
touchFunction(Kotlin_arrayGetElementAddress)
@@ -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.
*/
@@ -33,12 +33,14 @@ void gc::barriers::BarriersThreadData::onThreadRegistration() noexcept {
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");
RuntimeAssert(markBarriersEnabled.load(std::memory_order_relaxed),
"New allocations marking may only be requested by mark barriers");
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");
RuntimeAssert(!markBarriersEnabled.load(std::memory_order_relaxed),
"New allocations marking could only been requested by mark barriers");
markHandle_ = std::nullopt;
}
@@ -48,7 +50,8 @@ 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");
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) {
auto& objectData = alloc::objectDataForObject(allocated);
@@ -87,7 +90,7 @@ NO_INLINE void beforeHeapRefUpdateSlowPath(mm::DirectRefAccessor ref, ObjHeader*
// TODO perhaps it would be better to path the thread data from outside
auto& threadData = *mm::ThreadRegistry::Instance().CurrentThreadData();
auto& markQueue = *threadData.gc().impl().gc().mark().markQueue();
gc::mark::ParallelMark::MarkTraits::tryEnqueue(markQueue, prev);
gc::mark::ConcurrentMark::MarkTraits::tryEnqueue(markQueue, prev);
// No need to add the marked object in statistics here.
// Objects will be counted on dequeue.
}
@@ -102,3 +105,30 @@ ALWAYS_INLINE void gc::barriers::beforeHeapRefUpdate(mm::DirectRefAccessor ref,
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
NO_INLINE void weakRefReadInMarkSlowPath(ObjHeader* weakReferee) noexcept {
auto& threadData = *mm::ThreadRegistry::Instance().CurrentThreadData();
auto& markQueue = *threadData.gc().impl().gc().mark().markQueue();
gc::mark::ConcurrentMark::MarkTraits::tryEnqueue(markQueue, weakReferee);
}
} // namespace
ALWAYS_INLINE OBJ_GETTER(gc::barriers::weakRefReadBarrier, std::atomic<ObjHeader*>& 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);
}
return weak;
}
@@ -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.
*/
@@ -12,13 +12,14 @@
#include "GCStatistics.hpp"
#include "ReferenceOps.hpp"
/** See. `ConcurrentMark` */
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;
@@ -34,6 +35,6 @@ void disableMarkBarriers() noexcept;
void beforeHeapRefUpdate(mm::DirectRefAccessor ref, ObjHeader* value) noexcept;
// TODO re-introduce weak barriers again
OBJ_GETTER(weakRefReadBarrier, std::atomic<ObjHeader*>& weakReferee) noexcept;
} // namespace kotlin::gc
@@ -7,9 +7,8 @@
#include "gtest/gtest.h"
#include "Barriers.hpp"
#include "ParallelMark.hpp"
#include "ConcurrentMark.hpp"
#include "GCImpl.hpp"
#include "ManuallyScoped.hpp"
#include "ObjectTestSupport.hpp"
#include "ObjectOps.hpp"
#include "ReferenceOps.hpp"
@@ -57,7 +56,7 @@ public:
}
private:
gc::mark::ParallelMark::ParallelProcessor parProc_;
gc::mark::ConcurrentMark::ParallelProcessor parProc_;
};
} // namespace
@@ -0,0 +1,112 @@
/*
* 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.
*/
#include "ConcurrentMark.hpp"
#include "MarkAndSweepUtils.hpp"
#include "GCStatistics.hpp"
#include "Utils.hpp"
#include "GCImpl.hpp"
using namespace kotlin;
void gc::mark::ConcurrentMark::beginMarkingEpoch(gc::GCHandle gcHandle) {
gcHandle_ = gcHandle;
lockedMutatorsList_ = mm::ThreadRegistry::Instance().LockForIter();
parallelProcessor_.construct();
}
void gc::mark::ConcurrentMark::endMarkingEpoch() {
parallelProcessor_.destroy();
resetMutatorFlags();
lockedMutatorsList_ = std::nullopt;
}
void gc::mark::ConcurrentMark::runMainInSTW() {
ParallelProcessor::Worker mainWorker(*parallelProcessor_);
GCLogDebug(gcHandle().getEpoch(), "Creating main (#0) mark worker");
// create mutator mark queues
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 <MarkTraits>(gcHandle(), mainWorker);
barriers::enableMarkBarriers(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;
do {
for (auto& mutator: *lockedMutatorsList_) {
const bool markQueueNowEmpty = mutator.gc().impl().gc().mark().markQueue()->forceFlush();
if (!markQueueNowEmpty) {
refsRemainInMutatorQueues = true;
}
}
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();
}
}
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_) {
tryCollectRootSet(thread, markQueue);
}
}
void gc::mark::ConcurrentMark::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::ConcurrentMark::parallelMark(ParallelProcessor::Worker& worker) {
GCLogDebug(gcHandle().getEpoch(), "Mark loop has begun");
Mark <MarkTraits>(gcHandle(), worker);
}
void gc::mark::ConcurrentMark::resetMutatorFlags() {
for (auto& mut: *lockedMutatorsList_) {
mut.gc().impl().gc().clearMarkFlags();
}
}
@@ -0,0 +1,125 @@
/*
* Copyright 2010-2023 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
* that can be found in the LICENSE file.
*/
#pragma once
#include <condition_variable>
#include <mutex>
#include "GCStatistics.hpp"
#include "ManuallyScoped.hpp"
#include "ObjectData.hpp"
#include "ParallelProcessor.hpp"
#include "SafePoint.hpp"
#include "ThreadRegistry.hpp"
#include "Utils.hpp"
namespace kotlin::gc::mark {
/**
* Implementation of a Mark GC phase, that runs concurrent with mutator threads, using a "Snapshot at the Beginning" approach.
* Steps:
* 1. Pause all mutators, collect their root sets.
* 2. Mutators resume, maintaining a weak tri-collor invariant with the help of:
* - Deletion write barrier (Dijkstra et al):
* Remembers overwritten values in a thread-local mark queue.
* Barrier+write combination doesn't need to be atomic,
* 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
*/
class ConcurrentMark : private Pinned {
using MarkStackImpl = intrusive_forward_list<GC::ObjectData>;
public:
// work balancing parameters were chosen pretty arbitrary
using ParallelProcessor = ParallelProcessor<MarkStackImpl, 512, 4096>;
using MutatorQueue = ParallelProcessor::WorkSource;
class MarkTraits {
public:
using MarkQueue = ParallelProcessor::Worker;
using AnyQueue = ParallelProcessor::WorkSource;
static constexpr auto kAllowHeapToStackRefs = false;
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();
if (obj) {
auto object = alloc::objectForObjectData(*obj);
RuntimeLogDebug({logging::Tag::kGCMark}, "Dequeued %p", object);
return object;
}
return nullptr;
}
static ALWAYS_INLINE bool tryEnqueue(AnyQueue& queue, ObjHeader* object) noexcept {
auto& objectData = alloc::objectDataForObject(object);
bool pushed = queue.tryPush(objectData);
if (pushed) {
RuntimeLogDebug({logging::Tag::kGCMark}, "Enqueued %p", object);
}
return pushed;
}
static ALWAYS_INLINE bool tryMark(ObjHeader* object) noexcept {
auto& objectData = alloc::objectDataForObject(object);
bool pushed = objectData.tryMark();
if (pushed) {
RuntimeLogDebug({logging::Tag::kGCMark}, "Marked %p", object);
}
return pushed;
}
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);
}
};
class ThreadData : private Pinned {
public:
auto& markQueue() noexcept { return markQueue_; }
private:
ManuallyScoped<MutatorQueue> markQueue_{};
};
void beginMarkingEpoch(GCHandle gcHandle);
void endMarkingEpoch();
/** To be run by a single "main" GC thread during STW. */
void runMainInSTW();
/**
* To be run by mutator threads that would like to participate in mark.
* Will wait for STW detection by a "main" routine.
*/
void runOnMutator(mm::ThreadData& mutatorThread);
private:
GCHandle& gcHandle();
void completeMutatorsRootSet(MarkTraits::MarkQueue& markQueue);
void tryCollectRootSet(mm::ThreadData& thread, ParallelProcessor::Worker& markQueue);
void parallelMark(ParallelProcessor::Worker& worker);
void resetMutatorFlags();
GCHandle gcHandle_ = GCHandle::invalid();
std::optional<mm::ThreadRegistry::Iterable> lockedMutatorsList_;
ManuallyScoped<ParallelProcessor> parallelProcessor_{};
};
} // namespace kotlin::gc::mark
@@ -55,8 +55,6 @@ ScopedThread createGCThread(const char* name, Body&& body) {
} // namespace
void gc::ConcurrentMarkAndSweep::ThreadData::OnSuspendForGC() noexcept {
CallsCheckerIgnoreGuard guard;
gc_.markDispatcher_.runOnMutator(commonThreadData());
}
@@ -91,12 +89,10 @@ gc::ConcurrentMarkAndSweep::ConcurrentMarkAndSweep(
GCHandle::getByEpoch(epoch).finalizersDone();
state_.finalized(epoch);
}),
markDispatcher_(mutatorsCooperate),
mainThread_(createGCThread("Main GC thread", [this] { mainGCThreadBody(); })) {
for (std::size_t i = 0; i < auxGCThreads; ++i) {
auxThreads_.emplace_back(createGCThread("Auxiliary GC thread", [this] { auxiliaryGCThreadBody(); }));
}
RuntimeLogInfo({kTagGC}, "Parallel Mark & Concurrent Sweep GC initialized");
RuntimeAssert(!mutatorsCooperate, "Cooperative mutators aren't supported yet");
RuntimeAssert(auxGCThreads == 0, "Auxiliary GC threads aren't supported yet");
RuntimeLogInfo({kTagGC}, "Concurrent Mark & Sweep GC initialized");
}
gc::ConcurrentMarkAndSweep::~ConcurrentMarkAndSweep() {
@@ -119,7 +115,7 @@ bool gc::ConcurrentMarkAndSweep::FinalizersThreadIsRunning() noexcept {
}
void gc::ConcurrentMarkAndSweep::mainGCThreadBody() {
RuntimeLogWarning({kTagGC}, "Initializing Concurrent Mark and Sweep GC. Concurrent mark is not implemented yet.");
RuntimeLogWarning({kTagGC}, "Initializing Concurrent Mark and Sweep GC.");
while (true) {
auto epoch = state_.waitScheduled();
if (epoch.has_value()) {
@@ -128,14 +124,6 @@ void gc::ConcurrentMarkAndSweep::mainGCThreadBody() {
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 {
@@ -145,7 +133,7 @@ void gc::ConcurrentMarkAndSweep::PerformFullGC(int64_t epoch) noexcept {
markDispatcher_.beginMarkingEpoch(gcHandle);
GCLogDebug(epoch, "Main GC requested marking in mutators");
stopTheWorld(gcHandle);
stopTheWorld(gcHandle, "GC stop the world #1: collect root set");
auto& scheduler = gcScheduler_;
scheduler.onGCStart();
@@ -207,15 +195,3 @@ void gc::ConcurrentMarkAndSweep::PerformFullGC(int64_t epoch) noexcept {
// TODO: Consider having an always on sleeping finalizer thread.
finalizerProcessor_.ScheduleTasks(std::move(finalizerQueue), epoch);
}
void gc::ConcurrentMarkAndSweep::reconfigure(std::size_t maxParallelism, bool mutatorsCooperate, std::size_t auxGCThreads) noexcept {
if (compiler::gcMarkSingleThreaded()) {
RuntimeCheck(auxGCThreads == 0, "Auxiliary GC threads must not be created with gcMarkSingleThread");
return;
}
std::unique_lock mainGCLock(gcMutex);
markDispatcher_.reset(maxParallelism, mutatorsCooperate, [this] { auxThreads_.clear(); });
for (std::size_t i = 0; i < auxGCThreads; ++i) {
auxThreads_.emplace_back(createGCThread("Auxiliary GC thread", [this] { auxiliaryGCThreadBody(); }));
}
}
@@ -19,7 +19,7 @@
#include "IntrusiveList.hpp"
#include "MarkAndSweepUtils.hpp"
#include "ObjectData.hpp"
#include "ParallelMark.hpp"
#include "ConcurrentMark.hpp"
#include "ScopedThread.hpp"
#include "ThreadData.hpp"
#include "Types.h"
@@ -57,7 +57,7 @@ public:
ConcurrentMarkAndSweep& gc_;
mm::ThreadData& threadData_;
barriers::BarriersThreadData barriers_;
mark::ParallelMark::ThreadData mark_;
mark::ConcurrentMark::ThreadData mark_;
std::atomic<bool> rootSetLocked_ = false;
std::atomic<bool> published_ = false;
@@ -71,13 +71,10 @@ public:
void StopFinalizerThreadIfRunning() noexcept;
bool FinalizersThreadIsRunning() noexcept;
void reconfigure(std::size_t maxParallelism, bool mutatorsCooperate, size_t auxGCThreads) noexcept;
GCStateHolder& state() noexcept { return state_; }
private:
void mainGCThreadBody();
void auxiliaryGCThreadBody();
void PerformFullGC(int64_t epoch) noexcept;
alloc::Allocator& allocator_;
@@ -86,7 +83,7 @@ private:
GCStateHolder state_;
FinalizerProcessor<alloc::FinalizerQueue, alloc::FinalizerQueueTraits> finalizerProcessor_;
mark::ParallelMark markDispatcher_;
mark::ConcurrentMark markDispatcher_;
ScopedThread mainThread_;
std::vector<ScopedThread> auxThreads_;
};
@@ -11,7 +11,6 @@
#include "GCImpl.hpp"
#include "GlobalData.hpp"
#include "SafePoint.hpp"
#include "StableRef.hpp"
#include "TestSupport.hpp"
#include "TracingGCTest.hpp"
@@ -34,65 +33,5 @@ public:
} // namespace
TYPED_TEST_P(TracingGCTest, CMSMultipleMutatorsWeak) {
std::vector<Mutator> mutators(kDefaultThreadCount);
ObjHeader* globalRoot = nullptr;
test_support::RegularWeakReferenceImpl* weak = nullptr;
mutators[0]
.Execute([&weak, &globalRoot](mm::ThreadData& threadData, Mutator& mutator) {
auto& global = mutator.AddGlobalRoot();
auto& object = AllocateObject(threadData);
auto& objectWeak = ([&threadData, &object]() -> test_support::RegularWeakReferenceImpl& {
ObjHolder holder;
return test_support::InstallWeakReference(threadData, object.header(), holder.slot());
})();
global->field1 = objectWeak.header();
weak = &objectWeak;
globalRoot = global.header();
})
.wait();
// Make sure all mutators are initialized.
for (int i = 1; i < kDefaultThreadCount; ++i) {
mutators[i].Execute([](mm::ThreadData& threadData, Mutator& mutator) {}).wait();
}
std::vector<std::future<void>> gcFutures;
auto epoch = mm::GlobalData::Instance().gc().Schedule();
std::atomic<bool> gcDone = false;
// Spin until thread suspension is requested.
while (!mm::IsThreadSuspensionRequested()) {}
for (auto& mutator : mutators) {
gcFutures.emplace_back(mutator.Execute([&](mm::ThreadData& threadData, Mutator& mutator) noexcept {
bool dead = false;
while (!gcDone.load(std::memory_order_relaxed)) {
mm::safePoint(threadData);
auto weakReferee = weak->get();
if (dead) {
EXPECT_THAT(weakReferee, nullptr);
} else if (weakReferee == nullptr) {
dead = true;
}
}
EXPECT_THAT(weak->get(), nullptr);
}));
}
mm::GlobalData::Instance().gc().WaitFinalizers(epoch);
gcDone.store(true, std::memory_order_relaxed);
for (auto& future : gcFutures) {
future.wait();
}
for (auto& mutator : mutators) {
EXPECT_THAT(mutator.Alive(), testing::UnorderedElementsAre(globalRoot, weak->header()));
}
}
REGISTER_TYPED_TEST_SUITE_WITH_LISTS(TracingGCTest, TRACING_GC_TEST_LIST, CMSMultipleMutatorsWeak);
REGISTER_TYPED_TEST_SUITE_WITH_LISTS(TracingGCTest, TRACING_GC_TEST_LIST);
INSTANTIATE_TYPED_TEST_SUITE_P(CMS, TracingGCTest, ConcurrentMarkAndSweepTest);
@@ -59,17 +59,12 @@ bool gc::GC::FinalizersThreadIsRunning() noexcept {
// static
ALWAYS_INLINE void gc::GC::processObjectInMark(void* state, ObjHeader* object) noexcept {
gc::internal::processObjectInMark<gc::mark::ParallelMark::MarkTraits>(state, object);
gc::internal::processObjectInMark<gc::mark::ConcurrentMark::MarkTraits>(state, object);
}
// static
ALWAYS_INLINE void gc::GC::processArrayInMark(void* state, ArrayHeader* array) noexcept {
gc::internal::processArrayInMark<gc::mark::ParallelMark::MarkTraits>(state, array);
}
// static
ALWAYS_INLINE void gc::GC::processFieldInMark(void* state, ObjHeader* field) noexcept {
gc::internal::processFieldInMark<gc::mark::ParallelMark::MarkTraits>(state, field);
gc::internal::processArrayInMark<gc::mark::ConcurrentMark::MarkTraits>(state, array);
}
int64_t gc::GC::Schedule() noexcept {
@@ -89,7 +84,7 @@ ALWAYS_INLINE void gc::beforeHeapRefUpdate(mm::DirectRefAccessor ref, ObjHeader*
}
ALWAYS_INLINE OBJ_GETTER(gc::weakRefReadBarrier, std::atomic<ObjHeader*>& weakReferee) noexcept {
RETURN_OBJ(weakReferee.load(std::memory_order_relaxed));
RETURN_RESULT_OF(gc::barriers::weakRefReadBarrier, weakReferee);
}
bool gc::isMarked(ObjHeader* object) noexcept {
@@ -1,235 +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 "ParallelMark.hpp"
#include "MarkAndSweepUtils.hpp"
#include "GCStatistics.hpp"
#include "Utils.hpp"
// required to access gc thread data
#include "GCImpl.hpp"
using namespace kotlin;
namespace {
template<typename Cond>
void spinWait(Cond&& until) {
while (!until()) {
std::this_thread::yield();
}
}
} // namespace
bool gc::mark::MarkPacer::is(gc::mark::MarkPacer::Phase phase) const {
std::unique_lock lock(mutex_);
return phase_ == phase;
}
void gc::mark::MarkPacer::begin(gc::mark::MarkPacer::Phase phase) {
{
std::unique_lock lock(mutex_);
phase_ = phase;
}
cond_.notify_all();
}
void gc::mark::MarkPacer::wait(gc::mark::MarkPacer::Phase phase) {
std::unique_lock lock(mutex_);
cond_.wait(lock, [=]() { return phase_ >= phase; });
}
void gc::mark::MarkPacer::beginEpoch(uint64_t epoch) {
epoch_ = epoch;
begin(Phase::kReady);
GCLogDebug(epoch_.load(), "Mark is ready to recruit workers in a new epoch.");
}
void gc::mark::MarkPacer::waitNewEpochReadyOrShutdown() const {
std::unique_lock lock(mutex_);
cond_.wait(lock, [this]() { return phase_ >= Phase::kReady; });
}
void gc::mark::MarkPacer::waitEpochFinished(uint64_t currentEpoch) const {
std::unique_lock lock(mutex_);
cond_.wait(lock, [this, currentEpoch]() {
return phase_ == Phase::kIdle || phase_ == Phase::kShutdown || epoch_.load(std::memory_order_relaxed) > currentEpoch;
});
}
bool gc::mark::MarkPacer::acceptingNewWorkers() const {
std::unique_lock lock(mutex_);
return Phase::kReady <= phase_ && phase_ <= Phase::kParallelMark;
}
gc::mark::ParallelMark::ParallelMark(bool mutatorsCooperate) {
std::size_t maxParallelism = std::thread::hardware_concurrency();
if (maxParallelism == 0) {
maxParallelism = std::numeric_limits<std::size_t>::max();
}
setParallelismLevel(maxParallelism, mutatorsCooperate);
}
void gc::mark::ParallelMark::beginMarkingEpoch(gc::GCHandle gcHandle) {
gcHandle_ = gcHandle;
lockedMutatorsList_ = mm::ThreadRegistry::Instance().LockForIter();
parallelProcessor_.construct();
if (!compiler::gcMarkSingleThreaded()) {
std::unique_lock guard(workerCreationMutex_);
pacer_.beginEpoch(gcHandle.getEpoch());
// main worker is always accounted, so others would not be able to exhaust all the parallelism before main is instantiated
activeWorkersCount_ = 1;
}
}
void gc::mark::ParallelMark::endMarkingEpoch() {
if (!compiler::gcMarkSingleThreaded()) {
// We must now wait for every worker to finish the Mark procedure:
// wake up from possible waiting, publish statistics, etc.
// Only then it's safe to destroy the parallelProcessor and proceed to other GC tasks such as sweep.
spinWait([=]() { return activeWorkersCount_.load(std::memory_order_relaxed) == 0; });
std::unique_lock guard(workerCreationMutex_);
RuntimeAssert(activeWorkersCount_ == 0, "All the workers must already finish");
pacer_.begin(MarkPacer::Phase::kIdle);
}
parallelProcessor_.destroy();
resetMutatorFlags();
lockedMutatorsList_ = std::nullopt;
}
void gc::mark::ParallelMark::runMainInSTW() {
if (compiler::gcMarkSingleThreaded()) {
ParallelProcessor::Worker worker(*parallelProcessor_);
gc::collectRootSet<MarkTraits>(gcHandle(), worker, [] (mm::ThreadData&) { return true; });
gc::Mark<MarkTraits>(gcHandle(), worker);
} else {
RuntimeAssert(activeWorkersCount_ > 0, "Main worker must always be accounted");
ParallelProcessor::Worker mainWorker(*parallelProcessor_);
GCLogDebug(gcHandle().getEpoch(), "Creating main (#0) mark worker");
pacer_.begin(MarkPacer::Phase::kRootSet);
completeMutatorsRootSet(mainWorker);
spinWait([this] {
return allMutators([](mm::ThreadData& mut) { return mut.gc().impl().gc().published(); });
});
// global root set must be collected after all the mutator's global data have been published
collectRootSetGlobals<MarkTraits>(gcHandle(), mainWorker);
pacer_.begin(MarkPacer::Phase::kParallelMark);
parallelMark(mainWorker);
}
}
void gc::mark::ParallelMark::runOnMutator(mm::ThreadData& mutatorThread) {
if (compiler::gcMarkSingleThreaded() || !mutatorsCooperate_) return;
auto parallelWorker = createWorker();
if (parallelWorker) {
GCLogDebug(gcHandle().getEpoch(), "Mutator thread cooperates in marking");
tryCollectRootSet(mutatorThread, *parallelWorker);
completeRootSetAndMark(*parallelWorker);
}
}
void gc::mark::ParallelMark::runAuxiliary() {
RuntimeAssert(!compiler::gcMarkSingleThreaded(), "Should not reach here during single threaded mark");
pacer_.waitNewEpochReadyOrShutdown();
if (pacer_.is(MarkPacer::Phase::kShutdown)) return;
auto curEpoch = gcHandle().getEpoch();
auto parallelWorker = createWorker();
if (parallelWorker) {
completeRootSetAndMark(*parallelWorker);
}
pacer_.waitEpochFinished(curEpoch);
}
void gc::mark::ParallelMark::requestShutdown() {
pacer_.begin(MarkPacer::Phase::kShutdown);
}
bool gc::mark::ParallelMark::shutdownRequested() const {
return pacer_.is(MarkPacer::Phase::kShutdown);
}
gc::GCHandle& gc::mark::ParallelMark::gcHandle() {
RuntimeAssert(gcHandle_.isValid(), "GCHandle must be initialized");
return gcHandle_;
}
void gc::mark::ParallelMark::setParallelismLevel(size_t maxParallelism, bool mutatorsCooperate) {
RuntimeCheck(maxParallelism > 0, "Parallelism level can't be 0");
maxParallelism_ = maxParallelism;
mutatorsCooperate_ = mutatorsCooperate;
RuntimeLogInfo({kTagGC},
"Set up parallel mark with maxParallelism = %zu and %s" "cooperative mutators",
maxParallelism_, (mutatorsCooperate_ ? "" : "non-"));
}
void gc::mark::ParallelMark::completeRootSetAndMark(ParallelProcessor::Worker& parallelWorker) {
pacer_.wait(MarkPacer::Phase::kRootSet);
completeMutatorsRootSet(parallelWorker);
pacer_.wait(MarkPacer::Phase::kParallelMark);
parallelMark(parallelWorker);
}
void gc::mark::ParallelMark::completeMutatorsRootSet(MarkTraits::MarkQueue& markQueue) {
// workers compete for mutators to collect their root set
for (auto& thread: *lockedMutatorsList_) {
tryCollectRootSet(thread, markQueue);
}
}
void gc::mark::ParallelMark::tryCollectRootSet(mm::ThreadData& thread, MarkTraits::MarkQueue& markQueue) {
auto& gcData = thread.gc().impl().gc();
if (!gcData.tryLockRootSet()) return;
GCLogDebug(gcHandle().getEpoch(), "Root set collection on thread %d for thread %d",
konan::currentThreadId(), thread.threadId());
gcData.publish();
collectRootSetForThread<MarkTraits>(gcHandle(), markQueue, thread);
}
void gc::mark::ParallelMark::parallelMark(ParallelProcessor::Worker& worker) {
GCLogDebug(gcHandle().getEpoch(), "Mark loop has begun");
Mark<MarkTraits>(gcHandle(), worker);
std::unique_lock guard(workerCreationMutex_);
activeWorkersCount_.fetch_sub(1, std::memory_order_relaxed);
}
std::optional<gc::mark::ParallelMark::ParallelProcessor::Worker> gc::mark::ParallelMark::createWorker() {
std::unique_lock guard(workerCreationMutex_);
if (!pacer_.acceptingNewWorkers() ||
activeWorkersCount_.load(std::memory_order_relaxed) >= maxParallelism_ ||
activeWorkersCount_.load(std::memory_order_relaxed) == 0) return std::nullopt;
auto num = activeWorkersCount_.fetch_add(1, std::memory_order_relaxed);
GCLogDebug(gcHandle().getEpoch(), "Creating mark worker #%zu", num);
return std::make_optional<ParallelProcessor::Worker>(*parallelProcessor_);
}
void gc::mark::ParallelMark::resetMutatorFlags() {
for (auto& mut: *lockedMutatorsList_) {
auto& gcData = mut.gc().impl().gc();
if (!compiler::gcMarkSingleThreaded()) {
// single threaded mark do not use this flag
RuntimeAssert(gcData.published(), "Must have been published during mark");
}
gcData.clearMarkFlags();
}
}
@@ -1,193 +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 <condition_variable>
#include <mutex>
#include "GCStatistics.hpp"
#include "ManuallyScoped.hpp"
#include "ObjectData.hpp"
#include "ParallelProcessor.hpp"
#include "ThreadRegistry.hpp"
#include "Utils.hpp"
namespace kotlin::gc::mark {
class MarkPacer : private Pinned {
public:
enum class Phase {
/** Mark is not in progress. */
kIdle,
/**
* MarkDispatcher is ready to recruit new workers.
*
* In case of cooperative mark mutator threads are welcome to mark their own root sets.
* Each thread is free to start as soon as it reaches a safe point.
* No need to wait for others.
*/
kReady,
/**
* All mutator threads must be in a safe state at this point:
* 1) Suspended on a safe point;
* 2) In the native code;
* 3) Registered as cooperative markers during previous phase.
*
* Now all the GC workers are summoned to participate in a root set collection.
*/
kRootSet,
/**
* Root set is collected. No more workers can be instantiated, time to begin parallel mark.
* Parallel mark can't stop before all the created workers begin the marking.
*/
kParallelMark,
/** A shutdown was requested. There is nothing more to wait for. */
kShutdown,
};
bool is(Phase phase) const;
void begin(Phase phase);
void wait(Phase phase);
void beginEpoch(uint64_t epoch);
void waitNewEpochReadyOrShutdown() const;
void waitEpochFinished(uint64_t epoch) const;
bool acceptingNewWorkers() const;
private:
std::atomic<uint64_t> epoch_ = 0;
Phase phase_ = Phase::kIdle;
mutable std::mutex mutex_;
mutable std::condition_variable cond_;
};
/**
* Parallel mark dispatcher.
* Mark can be performed on one or more threads.
* Each threads wanting to participate have to execute an appropriate run- routine when ready to mark.
* There must be exactly one executor of a `runMainInSTW()`.
*
* Mark workers are able to balance work between each other through sharing/stealing.
*/
class ParallelMark : private Pinned {
using MarkStackImpl = intrusive_forward_list<GC::ObjectData>;
// work balancing parameters were chosen pretty arbitrary
public:
using ParallelProcessor = ParallelProcessor<MarkStackImpl, 512, 4096>;
using MutatorQueue = ParallelProcessor::WorkSource;
class MarkTraits {
public:
using MarkQueue = ParallelProcessor::Worker;
using AnyQueue = ParallelProcessor::WorkSource;
static void clear(AnyQueue& queue) noexcept {
RuntimeAssert(queue.localEmpty(), "Mark queue must be empty");
}
static ALWAYS_INLINE ObjHeader* tryDequeue(MarkQueue& queue) noexcept {
auto* obj = compiler::gcMarkSingleThreaded() ? queue.tryPopLocal() : queue.tryPop();
if (obj) {
return alloc::objectForObjectData(*obj);
}
return nullptr;
}
static ALWAYS_INLINE bool tryEnqueue(AnyQueue& queue, ObjHeader* object) noexcept {
auto& objectData = alloc::objectDataForObject(object);
return compiler::gcMarkSingleThreaded() ? queue.tryPushLocal(objectData) : queue.tryPush(objectData);
}
static ALWAYS_INLINE bool tryMark(ObjHeader* object) noexcept {
auto& objectData = alloc::objectDataForObject(object);
return objectData.tryMark();
}
static ALWAYS_INLINE void processInMark(MarkQueue& markQueue, ObjHeader* object) noexcept {
auto process = object->type_info()->processObjectInMark;
RuntimeAssert(process != nullptr, "Got null processObjectInMark for object %p", object);
process(static_cast<void*>(&markQueue), object);
}
};
class ThreadData : private Pinned {
public:
auto& markQueue() noexcept { return markQueue_; }
private:
ManuallyScoped<MutatorQueue> markQueue_{};
};
explicit ParallelMark(bool mutatorsCooperate);
void beginMarkingEpoch(gc::GCHandle gcHandle);
void endMarkingEpoch();
/** To be run by a single "main" GC thread during STW. */
void runMainInSTW();
/**
* To be run by mutator threads that would like to participate in mark.
* Will wait for STW detection by a "main" routine.
*/
void runOnMutator(mm::ThreadData& mutatorThread);
/**
* To be run by auxiliary GC threads.
* Will wait for STW detection by a "main" routine.
*/
void runAuxiliary();
void requestShutdown();
bool shutdownRequested() const;
template<typename Pred>
void reset(std::size_t maxParallelism, bool mutatorsCooperate, Pred waitForWorkersToFinish) {
pacer_.begin(MarkPacer::Phase::kShutdown);
waitForWorkersToFinish();
pacer_.begin(MarkPacer::Phase::kIdle);
setParallelismLevel(maxParallelism, mutatorsCooperate);
}
private:
GCHandle& gcHandle();
void setParallelismLevel(size_t maxParallelism, bool mutatorsCooperate);
template <typename Pred>
bool allMutators(Pred predicate) noexcept {
for (auto& thread : *lockedMutatorsList_) {
if (!predicate(thread)) {
return false;
}
}
return true;
}
void completeRootSetAndMark(ParallelProcessor::Worker& parallelWorker);
void completeMutatorsRootSet(MarkTraits::MarkQueue& markQueue);
void tryCollectRootSet(mm::ThreadData& thread, ParallelProcessor::Worker& markQueue);
void parallelMark(ParallelProcessor::Worker& worker);
std::optional<ParallelProcessor::Worker> createWorker();
void resetMutatorFlags();
std::size_t maxParallelism_ = 1;
bool mutatorsCooperate_ = false;
GCHandle gcHandle_ = GCHandle::invalid();
MarkPacer pacer_;
std::optional<mm::ThreadRegistry::Iterable> lockedMutatorsList_;
ManuallyScoped<ParallelProcessor> parallelProcessor_{};
std::mutex workerCreationMutex_;
std::atomic<std::size_t> activeWorkersCount_ = 0;
};
} // namespace kotlin::gc::mark
@@ -72,7 +72,6 @@ public:
static void processObjectInMark(void* state, ObjHeader* object) noexcept;
static void processArrayInMark(void* state, ArrayHeader* array) noexcept;
static void processFieldInMark(void* state, ObjHeader* field) noexcept;
// TODO: These should exist only in the scheduler.
int64_t Schedule() noexcept;
@@ -5,8 +5,8 @@
#include "MarkAndSweepUtils.hpp"
void kotlin::gc::stopTheWorld(kotlin::gc::GCHandle gcHandle) noexcept {
bool didSuspend = mm::RequestThreadsSuspension();
void kotlin::gc::stopTheWorld(GCHandle gcHandle, const char* reason) noexcept {
bool didSuspend = mm::RequestThreadsSuspension(reason);
RuntimeAssert(didSuspend, "Only GC thread can request suspension");
gcHandle.suspensionRequested();
@@ -26,27 +26,32 @@ namespace gc {
namespace internal {
template <typename Traits>
void processFieldInMark(void* state, ObjHeader* field) noexcept {
void processFieldInMark(void* state, ObjHeader* object, ObjHeader* field) noexcept {
auto& markQueue = *static_cast<typename Traits::MarkQueue*>(state);
if (field->heap()) {
Traits::tryEnqueue(markQueue, field);
}
if constexpr (!Traits::kAllowHeapToStackRefs) {
if (object->heap()) {
RuntimeAssert(!field->local(), "Heap object %p references stack object %p[typeInfo=%p]", object, field, field->type_info());
}
}
}
template <typename Traits>
void processObjectInMark(void* state, ObjHeader* object) noexcept {
traverseClassObjectFields(object, [state] (auto fieldAccessor) noexcept {
traverseClassObjectFields(object, [=] (auto fieldAccessor) noexcept {
if (ObjHeader* field = fieldAccessor.direct()) {
processFieldInMark<Traits>(state, field);
processFieldInMark<Traits>(state, object, field);
}
});
}
template <typename Traits>
void processArrayInMark(void* state, ArrayHeader* array) noexcept {
traverseArrayOfObjectsElements(array, [state] (auto elemAccessor) noexcept {
traverseArrayOfObjectsElements(array, [=] (auto elemAccessor) noexcept {
if (ObjHeader* elem = elemAccessor.direct()) {
processFieldInMark<Traits>(state, elem);
processFieldInMark<Traits>(state, array->obj(), elem);
}
});
}
@@ -180,11 +185,11 @@ struct DefaultProcessWeaksTraits {
static bool IsMarked(ObjHeader* obj) noexcept { return gc::isMarked(obj); }
};
void stopTheWorld(GCHandle gcHandle) noexcept;
void stopTheWorld(GCHandle gcHandle, const char* reason) noexcept;
void resumeTheWorld(GCHandle gcHandle) noexcept;
[[nodiscard]] inline auto stopTheWorldInScope(GCHandle gcHandle) noexcept {
return ScopeGuard([=]() { stopTheWorld(gcHandle); }, [=]() { resumeTheWorld(gcHandle); });
return ScopeGuard([=]() { stopTheWorld(gcHandle, "GC stop the world"); }, [=]() { resumeTheWorld(gcHandle); });
}
} // namespace gc
@@ -75,6 +75,8 @@ class ScopedMarkTraits : private Pinned {
public:
using MarkQueue = std::vector<ObjHeader*>;
static constexpr auto kAllowHeapToStackRefs = true;
ScopedMarkTraits() {
RuntimeAssert(instance_ == nullptr, "Only one ScopedMarkTraits is allowed");
instance_ = this;
@@ -1073,6 +1073,180 @@ TYPED_TEST_P(TracingGCTest, FreeObjectWithFreeWeakReversedOrder) {
f1.wait();
}
TYPED_TEST_P(TracingGCTest, MutateBetweenSafePoints) {
constexpr auto kQuant = 5;
constexpr auto kGcNumber = 5;
constexpr auto kMaxItersPerGC = 10;
std::atomic<bool> stopMutation = false;
std::atomic<bool> startMutation = false;
std::atomic<std::size_t> initializedMutators = 0;
std::atomic<int> gcEpoch = 0;
Mutator scheduler;
std::future<void> schedulerFuture = scheduler.Execute([&](mm::ThreadData& threadData, Mutator& mutator) {
while (initializedMutators < kDefaultThreadCount) { /* wait */ }
startMutation = true;
for (int i = 0; i < kGcNumber; ++i) {
gcEpoch = i;
mm::GlobalData::Instance().gcScheduler().scheduleAndWaitFinalized();
}
stopMutation = true;
});
std::vector<Mutator> mutators(kDefaultThreadCount);
std::vector<std::future<void>> mutatorFutures;
for (int i = 0; i < kDefaultThreadCount; ++i) {
mutatorFutures.emplace_back(mutators[i].Execute([&](mm::ThreadData& threadData, Mutator& mutator) {
StackObjectHolder head{threadData};
ObjHeader* tail = head.header();
auto append = [&](int count) {
for (int i = 0; i < count; ++i) {
auto& next = AllocateObject(threadData);
auto& tailObj = test_support::Object<Payload>::FromObjHeader(tail);
tailObj->field1 = next.header();
tail = next.header();
}
};
auto cut = [&](ObjHeader* first, int count) {
ObjHeader* last = first;
for (int i = 0; i < count; ++i) {
auto& lastObj = test_support::Object<Payload>::FromObjHeader(last);
last = lastObj->field1.accessor().load();
}
auto& firstObj = test_support::Object<Payload>::FromObjHeader(first);
auto& lastObj = test_support::Object<Payload>::FromObjHeader(last);
firstObj->field1.accessor() = lastObj->field1.accessor().load();
};
// Initialize
append(kQuant * 2);
++initializedMutators;
while (!startMutation.load()) { /* wait */ };
int iter = 0;
while (!stopMutation.load()) {
// do not let mutator outpace gc to much
while (iter > (gcEpoch.load() * kMaxItersPerGC) && !stopMutation.load()) {
mm::safePoint(threadData);
}
auto oldTail = tail;
append(kQuant);
// [head]->(a)->...->(b)->...->(oldTail)->(c)->...->(d)
mm::safePoint(threadData);
append(kQuant);
// [head]->(a)->...->(b)->...->(oldTail)->(c)->...->(d)->...->(tail)
mm::safePoint(threadData);
cut(head.header(), kQuant);
// (a)->...-+
// v
// [head]->(b)->...->(oldTail)->(c)->...->(d)->...->(tail)
mm::safePoint(threadData);
cut(oldTail, kQuant);
// (a)->...-+ (c)->...-+
// v v
// [head]->(b)->...->(oldTail)->(d)->...->(tail)
mm::safePoint(threadData);
++iter;
}
std::size_t length = 0;
auto* node = &test_support::Object<Payload>::FromObjHeader(head.header());
while ((*node)->field1.accessor().load() != nullptr) {
node = &test_support::Object<Payload>::FromObjHeader((*node)->field1.accessor().load());
++length;
}
EXPECT_THAT(length, kQuant * 2);
}));
}
schedulerFuture.wait();
for (auto& future : mutatorFutures) {
future.wait();
}
}
TYPED_TEST_P(TracingGCTest, WeakResuractionInMark) {
constexpr auto kObjectsPerThread = 100;
std::vector<Mutator> mutators(kDefaultThreadCount);
std::vector<test_support::RegularWeakReferenceImpl*> weaks(kDefaultThreadCount);
std::vector<test_support::Object<Payload>*> roots(kDefaultThreadCount);
// initialize threads
for (int i = 0; i < kDefaultThreadCount; ++i) {
mutators[i].Execute([&, i](mm::ThreadData& threadData, Mutator& mutator) {
// make sure GC will have somthing to do whil we play with weak references
for (int j = 0; j < kObjectsPerThread; ++j) {
auto& object = AllocateObject(threadData);
object->field1 = roots[i]->header();
roots[i] = &object;
}
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<std::future<void>> mutatorFutures;
for (int i = 0; i < kDefaultThreadCount; ++i) {
mutatorFutures.emplace_back(mutators[i].Execute([&, i](mm::ThreadData& threadData, Mutator& mutator) noexcept {
safePoint(threadData);
auto weakReferee = weaks[i]->get();
(*roots[i])->field2 = weakReferee;
bool resurected = weakReferee != nullptr;
while (!gcDone.load(std::memory_order_relaxed)) {
safePoint(threadData);
}
if (resurected) {
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();
}
}
#define TRACING_GC_TEST_LIST \
RootSet, \
InterconnectedRootSet, \
@@ -1092,7 +1266,9 @@ TYPED_TEST_P(TracingGCTest, FreeObjectWithFreeWeakReversedOrder) {
MultipleMutatorsAddToRootSetAfterCollectionRequested, \
CrossThreadReference, \
NewThreadsWhileRequestingCollection, \
FreeObjectWithFreeWeakReversedOrder
FreeObjectWithFreeWeakReversedOrder, \
MutateBetweenSafePoints, \
WeakResuractionInMark
template <typename FixtureImpl>
class STWMarkGCTest : public TracingGCTest<FixtureImpl> {};
@@ -48,9 +48,6 @@ ALWAYS_INLINE void gc::GC::processObjectInMark(void* state, ObjHeader* object) n
// static
ALWAYS_INLINE void gc::GC::processArrayInMark(void* state, ArrayHeader* array) noexcept {}
// static
ALWAYS_INLINE void gc::GC::processFieldInMark(void* state, ObjHeader* field) noexcept {}
int64_t gc::GC::Schedule() noexcept {
return 0;
}
@@ -67,11 +67,6 @@ ALWAYS_INLINE void gc::GC::processArrayInMark(void* state, ArrayHeader* array) n
gc::internal::processArrayInMark<gc::mark::ParallelMark::MarkTraits>(state, array);
}
// static
ALWAYS_INLINE void gc::GC::processFieldInMark(void* state, ObjHeader* field) noexcept {
gc::internal::processFieldInMark<gc::mark::ParallelMark::MarkTraits>(state, field);
}
int64_t gc::GC::Schedule() noexcept {
return impl_->gc().state().schedule();
}
@@ -82,6 +82,8 @@ public:
public:
using MarkQueue = ParallelProcessor::Worker;
static constexpr auto kAllowHeapToStackRefs = true;
static void clear(MarkQueue& queue) noexcept {
RuntimeAssert(queue.localEmpty(), "Mark queue must be empty");
}
@@ -148,7 +148,7 @@ void gc::ParallelMarkConcurrentSweep::PerformFullGC(int64_t epoch) noexcept {
markDispatcher_.beginMarkingEpoch(gcHandle);
GCLogDebug(epoch, "Main GC requested marking in mutators");
stopTheWorld(gcHandle);
stopTheWorld(gcHandle, "GC stop the world #1: mark");
auto& scheduler = gcScheduler_;
scheduler.onGCStart();
@@ -168,7 +168,7 @@ void gc::ParallelMarkConcurrentSweep::PerformFullGC(int64_t epoch) noexcept {
gc::processWeaks<DefaultProcessWeaksTraits>(gcHandle, mm::SpecialRefRegistry::instance());
if (compiler::concurrentWeakSweep()) {
stopTheWorld(gcHandle);
stopTheWorld(gcHandle, "GC stop the world #2: prepare heap for sweep");
gc::DisableWeakRefBarriers();
}
@@ -60,11 +60,6 @@ ALWAYS_INLINE void gc::GC::processArrayInMark(void* state, ArrayHeader* array) n
gc::internal::processArrayInMark<gc::internal::MarkTraits>(state, array);
}
// static
ALWAYS_INLINE void gc::GC::processFieldInMark(void* state, ObjHeader* field) noexcept {
gc::internal::processFieldInMark<gc::internal::MarkTraits>(state, field);
}
int64_t gc::GC::Schedule() noexcept {
return impl_->gc().state().schedule();
}
@@ -57,7 +57,7 @@ bool gc::SameThreadMarkAndSweep::FinalizersThreadIsRunning() noexcept {
void gc::SameThreadMarkAndSweep::PerformFullGC(int64_t epoch) noexcept {
auto gcHandle = GCHandle::create(epoch);
stopTheWorld(gcHandle);
stopTheWorld(gcHandle, "GC stop the world");
auto& scheduler = gcScheduler_;
scheduler.onGCStart();
@@ -68,6 +68,8 @@ namespace internal {
struct MarkTraits {
using MarkQueue = gc::SameThreadMarkAndSweep::MarkQueue;
static constexpr auto kAllowHeapToStackRefs = true;
static void clear(MarkQueue& queue) noexcept { queue.clear(); }
static ObjHeader* tryDequeue(MarkQueue& queue) noexcept {
@@ -38,8 +38,9 @@ enum class Tag : int32_t {
kAlloc = 6,
kBalancing = 7,
kBarriers = 8,
kGCMark = 9,
kEnumSize = 9
kEnumSize = 10
};
namespace internal {
@@ -65,6 +66,7 @@ inline const char* name(Tag tag) {
case Tag::kAlloc: return "alloc";
case Tag::kBalancing: return "balancing";
case Tag::kBarriers: return "barriers";
case Tag::kGCMark: return "gcMark";
case Tag::kEnumSize: break;
}
+3 -2
View File
@@ -77,7 +77,9 @@ struct ObjHeader {
* Hardware guaranties on many supported platforms doesn't allow this to happen.
*/
const TypeInfo* type_info() const {
return atomicGetRelaxed(&clearPointerBits(typeInfoOrMetaRelaxed(), OBJECT_TAG_MASK)->typeInfo_);
const TypeInfo* typeInfo = atomicGetRelaxed(&clearPointerBits(typeInfoOrMetaRelaxed(), OBJECT_TAG_MASK)->typeInfo_);
RuntimeAssert(typeInfo != nullptr, "TypeInfo ptr in object %p in null", this);
return typeInfo;
}
bool has_meta_object() const {
@@ -599,7 +601,6 @@ void compactObjectPoolInCurrentThread() noexcept;
RUNTIME_NOTHROW ALWAYS_INLINE extern "C" void Kotlin_processObjectInMark(void* state, ObjHeader* object);
RUNTIME_NOTHROW ALWAYS_INLINE extern "C" void Kotlin_processArrayInMark(void* state, ObjHeader* object);
RUNTIME_NOTHROW ALWAYS_INLINE extern "C" void Kotlin_processFieldInMark(void* state, ObjHeader* field);
RUNTIME_NOTHROW ALWAYS_INLINE extern "C" void Kotlin_processEmptyObjectInMark(void* state, ObjHeader* object);
#endif // RUNTIME_MEMORY_H
@@ -95,7 +95,7 @@ public:
explicit WorkSource(ParallelProcessor& dispatcher) : dispatcher_(dispatcher) {}
ALWAYS_INLINE bool retainsNoWork() const noexcept {
return batch_.empty() && this->localEmpty();
return batch_.empty() && overflowList().empty();
}
ALWAYS_INLINE bool tryPush(typename ListImpl::reference value) noexcept {
@@ -119,7 +119,7 @@ public:
if (!batch_.empty()) {
bool released = dispatcher_.releaseBatch(std::move(batch_));
if (released) {
RuntimeLogDebug({ kTagBalancing }, "Work butch flushed");
RuntimeLogDebug({ kTagBalancing }, "Work batch flushed");
batch_ = Batch{};
} else {
RuntimeLogDebug({ kTagBalancing }, "Failed to force flush work queue");
@@ -130,7 +130,7 @@ public:
if (overflowList().empty()) {
return true;
} else {
RuntimeLogDebug({ kTagBalancing }, "Refiling batch from overflow list");
RuntimeLogDebug({ kTagBalancing }, "Refilling batch from overflow list");
batch_.fillFrom(overflowList());
}
}
@@ -141,6 +141,10 @@ public:
return this->localQueue_;
}
const ListImpl& overflowList() const noexcept {
return this->localQueue_;
}
ParallelProcessor& dispatcher_;
Batch batch_;
};
@@ -222,6 +226,13 @@ public:
return registeredWorkers_.load(std::memory_order_relaxed);
}
/** Prepare for a new round of work processing. Must be called only after a previous round is fully finished. */
void resetForNewWork() noexcept {
RuntimeAssert(allDone_, "A work processing iteration must be finished");
RuntimeAssert(waitingWorkers_ == 0, "There must be no workers sleeping inside processing loop");
allDone_ = false;
}
private:
bool releaseBatch(Batch&& batch) {
RuntimeAssert(!batch.empty(), "A batch to release into shared pool must be non-empty");
@@ -575,10 +575,6 @@ RUNTIME_NOTHROW ALWAYS_INLINE extern "C" void Kotlin_processArrayInMark(void* st
gc::GC::processArrayInMark(state, object->array());
}
RUNTIME_NOTHROW ALWAYS_INLINE extern "C" void Kotlin_processFieldInMark(void* state, ObjHeader* field) {
gc::GC::processFieldInMark(state, field);
}
RUNTIME_NOTHROW ALWAYS_INLINE extern "C" void Kotlin_processEmptyObjectInMark(void* state, ObjHeader* object) {
// Empty object. Nothing to do.
// TODO: Try to generate it in the code generator.
@@ -23,7 +23,7 @@ namespace {
} // namespace
std::atomic<bool> kotlin::mm::internal::gSuspensionRequested = false;
std::atomic<mm::internal::SuspensionReason> mm::internal::gSuspensionReauestReason = nullptr;
ALWAYS_INLINE mm::ThreadSuspensionData::MutatorPauseHandle::MutatorPauseHandle(const char* reason, mm::ThreadData& threadData) noexcept
: reason_(reason), threadData_(threadData), pauseStartTimeMicros_(konan::getTimeMicros())
@@ -65,7 +65,7 @@ kotlin::ThreadState kotlin::mm::ThreadSuspensionData::setState(kotlin::ThreadSta
NO_EXTERNAL_CALLS_CHECK void kotlin::mm::ThreadSuspensionData::suspendIfRequested() noexcept {
if (IsThreadSuspensionRequested()) {
auto pauseHandle = pauseMutationInScope("stop the world");
auto pauseHandle = pauseMutationInScope(internal::gSuspensionReauestReason.load(std::memory_order_relaxed));
threadData_.gc().OnSuspendForGC();
std::unique_lock lock(gSuspensionMutex);
@@ -80,18 +80,18 @@ ALWAYS_INLINE mm::ThreadSuspensionData::MutatorPauseHandle mm::ThreadSuspensionD
return MutatorPauseHandle(reason, threadData_);
}
bool kotlin::mm::RequestThreadsSuspension() noexcept {
bool kotlin::mm::RequestThreadsSuspension(internal::SuspensionReason reason) noexcept {
CallsCheckerIgnoreGuard guard;
RuntimeAssert(gSafePointActivator == std::nullopt, "Current thread already suspended threads.");
{
std::unique_lock lock(gSuspensionMutex);
// Someone else has already suspended threads.
if (internal::gSuspensionRequested.load(std::memory_order_relaxed)) {
if (internal::gSuspensionReauestReason.load(std::memory_order_relaxed) != nullptr) {
return false;
}
gSafePointActivator = mm::SafePointActivator();
internal::gSuspensionRequested.store(true);
internal::gSuspensionReauestReason.store(reason);
}
return true;
@@ -116,7 +116,7 @@ void kotlin::mm::ResumeThreads() noexcept {
// https://en.cppreference.com/w/cpp/thread/condition_variable
{
std::unique_lock lock(gSuspensionMutex);
internal::gSuspensionRequested.store(false);
internal::gSuspensionReauestReason.store(nullptr);
}
gSuspensionCondVar.notify_all();
}
@@ -9,6 +9,7 @@
#include <atomic>
#include "Memory.h"
#include "Utils.hpp"
namespace kotlin {
namespace mm {
@@ -17,14 +18,15 @@ class ThreadData;
namespace internal {
extern std::atomic<bool> gSuspensionRequested;
using SuspensionReason = const char*;
extern std::atomic<SuspensionReason> gSuspensionReauestReason;
} // namespace internal
inline bool IsThreadSuspensionRequested() noexcept {
// Must use seq_cst ordering for synchronization with GC
// in native->runnable transition.
return internal::gSuspensionRequested.load();
return internal::gSuspensionReauestReason.load();
}
class ThreadSuspensionData : private Pinned {
@@ -71,7 +73,7 @@ private:
mm::ThreadData& threadData_;
};
bool RequestThreadsSuspension() noexcept;
bool RequestThreadsSuspension(const char* reason) noexcept;
void WaitForThreadsSuspension() noexcept;
/**
@@ -80,8 +82,8 @@ void WaitForThreadsSuspension() noexcept;
* of this call will be suspended on exit from the Native state.
* Returns false if some other thread has suspended the threads.
*/
inline bool SuspendThreads() noexcept {
if (!RequestThreadsSuspension()) {
inline bool SuspendThreads(internal::SuspensionReason reason) noexcept {
if (!RequestThreadsSuspension(reason)) {
return false;
}
WaitForThreadsSuspension();
@@ -128,7 +128,7 @@ TEST_F(ThreadSuspensionTest, SimpleStartStop) {
reportProgress(i, kIterations);
canStart = true;
mm::SuspendThreads();
mm::SuspendThreads("test");
auto suspended = collectSuspended();
EXPECT_THAT(suspended, testing::Each(true));
EXPECT_EQ(mm::IsThreadSuspensionRequested(), true);
@@ -171,7 +171,7 @@ TEST_F(ThreadSuspensionTest, SwitchStateToNative) {
reportProgress(i, kIterations);
canStart = true;
mm::SuspendThreads();
mm::SuspendThreads("test");
EXPECT_EQ(mm::IsThreadSuspensionRequested(), true);
mm::ResumeThreads();
@@ -196,7 +196,7 @@ TEST_F(ThreadSuspensionTest, ConcurrentSuspend) {
ready[i] = true;
waitUntilThreadsAreReady();
bool success = mm::SuspendThreads();
bool success = mm::SuspendThreads("test");
if (success) {
successCount++;
auto allThreadData = collectThreadData();
@@ -248,7 +248,7 @@ TEST_F(ThreadSuspensionTest, FileInitializationWithSuspend) {
waitUntilThreadsAreReady();
auto gcThread = std::async(std::launch::async, [] {
mm::RequestThreadsSuspension();
mm::RequestThreadsSuspension("test");
mm::WaitForThreadsSuspension();
mm::ResumeThreads();
});