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 00171c05f91..7079b77bcbd 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 @@ -41,6 +41,10 @@ object BinaryOptions : BinaryOptionRegistry() { val concurrentWeakSweep by booleanOption() + val gcMutatorsCooperate by booleanOption() + + val auxGCThreads by uintOption() + val linkRuntime by option() val bundleId by stringOption() @@ -95,6 +99,15 @@ open class BinaryOptionRegistry { } } + protected fun uintOption(): PropertyDelegateProvider>> = + PropertyDelegateProvider { _, property -> + val option = BinaryOption(property.name, UIntValueParser) + register(option) + ReadOnlyProperty { _, _ -> + option.compilerConfigurationKey + } + } + protected fun stringOption(): PropertyDelegateProvider>> = PropertyDelegateProvider { _, property -> val option = BinaryOption(property.name, StringValueParser) @@ -121,6 +134,13 @@ private object BooleanValueParser : BinaryOption.ValueParser { get() = "true|false" } +private object UIntValueParser : BinaryOption.ValueParser { + override fun parse(value: String): UInt? = value.toUIntOrNull() + + override val validValuesHint: String? + get() = "non-negative-number" +} + private object StringValueParser : BinaryOption.ValueParser { override fun parse(value: String) = value override val validValuesHint: String? 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 e6bd80c8003..e3a741bd4ea 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 @@ -143,12 +143,41 @@ class KonanConfig(val project: Project, val configuration: CompilerConfiguration } ?: arg } - val gcMarkSingleThreaded: Boolean - get() = configuration.get(BinaryOptions.gcMarkSingleThreaded) ?: false + private val defaultGcMarkSingleThreaded get() = target.family == Family.MINGW + + val gcMarkSingleThreaded: Boolean by lazy { + configuration.get(BinaryOptions.gcMarkSingleThreaded) ?: defaultGcMarkSingleThreaded + } val concurrentWeakSweep: Boolean get() = configuration.get(BinaryOptions.concurrentWeakSweep) ?: false + val gcMutatorsCooperate: Boolean by lazy { + val mutatorsCooperate = configuration.get(BinaryOptions.gcMutatorsCooperate) + if (gcMarkSingleThreaded) { + if (mutatorsCooperate == true) { + configuration.report(CompilerMessageSeverity.STRONG_WARNING, + "Mutators cooperation is not supported during single threaded mark") + } + false + } else { + mutatorsCooperate ?: true + } + } + + val auxGCThreads: UInt by lazy { + val auxGCThreads = configuration.get(BinaryOptions.auxGCThreads) + if (gcMarkSingleThreaded) { + if (auxGCThreads != null && auxGCThreads != 0U) { + configuration.report(CompilerMessageSeverity.STRONG_WARNING, + "Auxiliary GC workers are not supported during single threaded mark") + } + 0U + } else { + auxGCThreads ?: 0U + } + } + val irVerificationMode: IrVerificationMode get() = configuration.getNotNull(KonanConfigKeys.VERIFY_IR) @@ -408,6 +437,8 @@ class KonanConfig(val project: Project, val configuration: CompilerConfiguration append("-runtime_asserts=${runtimeAssertsMode.name}") if (disableMmap != defaultDisableMmap) append("-disable_mmap=${disableMmap}") + if (gcMarkSingleThreaded != defaultGcMarkSingleThreaded) + append("-gc_mark_single_threaded=${gcMarkSingleThreaded}") } private val userCacheFlavorString = buildString { 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 2b76888fe7e..05fee14aaad 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 @@ -2780,7 +2780,8 @@ internal class CodeGeneratorVisitor( return overrideRuntimeGlobal("Kotlin_destroyRuntimeMode", llvm.constInt32(context.config.destroyRuntimeMode.value)) - overrideRuntimeGlobal("Kotlin_gcMarkSingleThreaded", llvm.constInt32(if (context.config.gcMarkSingleThreaded) 1 else 0)) + overrideRuntimeGlobal("Kotlin_gcMutatorsCooperate", llvm.constInt32(if (context.config.gcMutatorsCooperate) 1 else 0)) + overrideRuntimeGlobal("Kotlin_auxGCThreads", llvm.constInt32(context.config.auxGCThreads.toInt())) overrideRuntimeGlobal("Kotlin_workerExceptionHandling", llvm.constInt32(context.config.workerExceptionHandling.value)) overrideRuntimeGlobal("Kotlin_suspendFunctionsFromAnyThreadFromObjC", llvm.constInt32(if (context.config.suspendFunctionsFromAnyThreadFromObjC) 1 else 0)) val getSourceInfoFunctionName = when (context.config.sourceInfoType) { @@ -3015,6 +3016,7 @@ internal fun NativeGenerationState.generateRuntimeConstantsModule() : LLVMModule setRuntimeConstGlobal("Kotlin_freezingEnabled", llvm.constInt32(if (config.freezing.enableFreezeAtRuntime) 1 else 0)) setRuntimeConstGlobal("Kotlin_freezingChecksEnabled", llvm.constInt32(if (config.freezing.enableFreezeChecks) 1 else 0)) setRuntimeConstGlobal("Kotlin_concurrentWeakSweep", llvm.constInt32(if (context.config.concurrentWeakSweep) 1 else 0)) + setRuntimeConstGlobal("Kotlin_gcMarkSingleThreaded", llvm.constInt32(if (config.gcMarkSingleThreaded) 1 else 0)) return llvmModule } diff --git a/kotlin-native/backend.native/tests/runtime/atomics/atomic_stress.kt b/kotlin-native/backend.native/tests/runtime/atomics/atomic_stress.kt index 9e96baedc14..d981372df41 100644 --- a/kotlin-native/backend.native/tests/runtime/atomics/atomic_stress.kt +++ b/kotlin-native/backend.native/tests/runtime/atomics/atomic_stress.kt @@ -86,7 +86,12 @@ fun testAtomicReferenceStress(workers: Array) { fun runStressTest() { val COUNT = 20 val workers = Array(COUNT, { _ -> Worker.start()}) + testAtomicIntStress(workers) testAtomicLongStress(workers) testAtomicReferenceStress(workers) + + workers.forEach { + it.requestTermination().result + } } diff --git a/kotlin-native/runtime/src/custom_alloc/cpp/Heap.cpp b/kotlin-native/runtime/src/custom_alloc/cpp/Heap.cpp index 15095f991a2..593ae317297 100644 --- a/kotlin-native/runtime/src/custom_alloc/cpp/Heap.cpp +++ b/kotlin-native/runtime/src/custom_alloc/cpp/Heap.cpp @@ -50,18 +50,22 @@ FinalizerQueue Heap::Sweep(gc::GCHandle gcHandle) noexcept { auto sweepHandle = gcHandle.sweepExtraObjects(); extraObjectPages_.Sweep(sweepHandle, finalizerQueue); } + // wait for concurrent assistants to finish sweeping the last popped page + while (concurrentSweepersCount_.load(std::memory_order_acquire) > 0) { + std::this_thread::yield(); + } CustomAllocDebug("Heap::Sweep done"); return finalizerQueue; } NextFitPage* Heap::GetNextFitPage(uint32_t cellCount, FinalizerQueue& finalizerQueue) noexcept { CustomAllocDebug("Heap::GetNextFitPage()"); - return nextFitPages_.GetPage(cellCount, finalizerQueue); + return nextFitPages_.GetPage(cellCount, finalizerQueue, concurrentSweepersCount_); } FixedBlockPage* Heap::GetFixedBlockPage(uint32_t cellCount, FinalizerQueue& finalizerQueue) noexcept { CustomAllocDebug("Heap::GetFixedBlockPage()"); - return fixedBlockPages_[cellCount].GetPage(cellCount, finalizerQueue); + return fixedBlockPages_[cellCount].GetPage(cellCount, finalizerQueue, concurrentSweepersCount_); } SingleObjectPage* Heap::GetSingleObjectPage(uint64_t cellCount, FinalizerQueue& finalizerQueue) noexcept { @@ -71,7 +75,7 @@ SingleObjectPage* Heap::GetSingleObjectPage(uint64_t cellCount, FinalizerQueue& ExtraObjectPage* Heap::GetExtraObjectPage(FinalizerQueue& finalizerQueue) noexcept { CustomAllocInfo("CustomAllocator::GetExtraObjectPage()"); - return extraObjectPages_.GetPage(0, finalizerQueue); + return extraObjectPages_.GetPage(0, finalizerQueue, concurrentSweepersCount_); } std_support::vector Heap::GetAllocatedObjects() noexcept { diff --git a/kotlin-native/runtime/src/custom_alloc/cpp/Heap.hpp b/kotlin-native/runtime/src/custom_alloc/cpp/Heap.hpp index d2fdf4f6db8..671d4944457 100644 --- a/kotlin-native/runtime/src/custom_alloc/cpp/Heap.hpp +++ b/kotlin-native/runtime/src/custom_alloc/cpp/Heap.hpp @@ -45,6 +45,8 @@ private: PageStore nextFitPages_; PageStore singleObjectPages_; PageStore extraObjectPages_; + + std::atomic concurrentSweepersCount_ = 0; }; } // namespace kotlin::alloc diff --git a/kotlin-native/runtime/src/custom_alloc/cpp/PageStore.hpp b/kotlin-native/runtime/src/custom_alloc/cpp/PageStore.hpp index bf989d4ce29..9bd8b2a981a 100644 --- a/kotlin-native/runtime/src/custom_alloc/cpp/PageStore.hpp +++ b/kotlin-native/runtime/src/custom_alloc/cpp/PageStore.hpp @@ -44,18 +44,25 @@ public: } } - T* GetPage(uint32_t cellCount, FinalizerQueue& finalizerQueue) noexcept { + T* GetPage(uint32_t cellCount, FinalizerQueue& finalizerQueue, std::atomic& concurrentSweepersCount_) noexcept { T* page; if ((page = ready_.Pop())) { used_.Push(page); return page; } - auto handle = gc::GCHandle::currentEpoch(); - if ((page = unswept_.Pop())) { - // If there're unswept_ pages, the GC is in progress. - GCSweepScope sweepHandle = T::currentGCSweepScope(*handle); - if ((page = SweepSingle(sweepHandle, page, unswept_, used_, finalizerQueue))) { - return page; + { + auto handle = gc::GCHandle::currentEpoch(); + ScopeGuard counterGuard( + [&]() { ++concurrentSweepersCount_; }, + [&]() { --concurrentSweepersCount_; } + ); + + if ((page = unswept_.Pop())) { + // If there're unswept_ pages, the GC is in progress. + GCSweepScope sweepHandle = T::currentGCSweepScope(*handle); + if ((page = SweepSingle(sweepHandle, page, unswept_, used_, finalizerQueue))) { + return page; + } } } if ((page = empty_.Pop())) { diff --git a/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.cpp b/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.cpp index 2084b8e1dde..f664fc72954 100644 --- a/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.cpp +++ b/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.cpp @@ -16,7 +16,6 @@ #include "MarkAndSweepUtils.hpp" #include "Memory.h" #include "ThreadData.hpp" -#include "ThreadRegistry.hpp" #include "ThreadSuspension.hpp" #include "GCState.hpp" #include "GCStatistics.hpp" @@ -28,10 +27,8 @@ using namespace kotlin; namespace { - [[clang::no_destroy]] std::mutex markingMutex; - [[clang::no_destroy]] std::condition_variable markingCondVar; - [[clang::no_destroy]] std::atomic markingRequested = false; - [[clang::no_destroy]] std::atomic markingEpoch = 0; + +[[clang::no_destroy]] std::mutex gcMutex; struct SweepTraits { using ObjectFactory = mm::ObjectFactory; @@ -57,6 +54,33 @@ struct ProcessWeaksTraits { } }; +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()); + body(); + RuntimeLogDebug({kTagGC}, "%s %d finishes execution", name, konan::currentThreadId()); + }); +} + +// TODO move to common +[[maybe_unused]] inline void checkMarkCorrectness(mm::ObjectFactory::Iterable& heap) { + if (compiler::runtimeAssertsMode() == compiler::RuntimeAssertsMode::kIgnore) return; + for (auto objRef: heap) { + auto obj = objRef.GetObjHeader(); + auto& objData = objRef.ObjectData(); + if (objData.marked()) { + traverseReferredObjects(obj, [obj](ObjHeader* field) { + if (field->heap()) { + auto& fieldObjData = + mm::ObjectFactory::NodeRef::From(field).ObjectData(); + RuntimeAssert(fieldObjData.marked(), "Field %p of an alive obj %p must be alive", field, obj); + } + }); + } + } +} + } // namespace void gc::ConcurrentMarkAndSweep::ThreadData::OnOOM(size_t size) noexcept { @@ -68,48 +92,70 @@ void gc::ConcurrentMarkAndSweep::ThreadData::OnOOM(size_t size) noexcept { void gc::ConcurrentMarkAndSweep::ThreadData::OnSuspendForGC() noexcept { CallsCheckerIgnoreGuard guard; - std::unique_lock lock(markingMutex); - if (!markingRequested.load()) return; - AutoReset scopedAssignMarking(&marking_, true); + gc_.markDispatcher_.runOnMutator(commonThreadData()); +} + +bool gc::ConcurrentMarkAndSweep::ThreadData::tryLockRootSet() { + bool expected = false; + bool locked = rootSetLocked_.compare_exchange_strong(expected, true, std::memory_order_acq_rel); + if (locked) { + RuntimeLogDebug({kTagGC}, "Thread %d have exclusively acquired thread %d's root set", konan::currentThreadId(), threadData_.threadId()); + } + return locked; +} + +void gc::ConcurrentMarkAndSweep::ThreadData::beginCooperation() { + cooperative_.store(true, std::memory_order_release); +} + +bool gc::ConcurrentMarkAndSweep::ThreadData::cooperative() const { + return cooperative_.load(std::memory_order_relaxed); +} + +void gc::ConcurrentMarkAndSweep::ThreadData::publish() { threadData_.Publish(); - markingCondVar.wait(lock, []() { return !markingRequested.load(); }); - // // Unlock while marking to allow mutliple threads to mark in parallel. - lock.unlock(); - uint64_t epoch = markingEpoch.load(); - GCLogDebug(epoch, "Parallel marking in thread %d", konan::currentThreadId()); - MarkQueue markQueue; - auto handle = GCHandle::getByEpoch(epoch); - gc::collectRootSetForThread(handle, markQueue, threadData_); - gc::Mark(handle, markQueue); + published_.store(true, std::memory_order_release); +} + +bool gc::ConcurrentMarkAndSweep::ThreadData::published() const { + return published_.load(std::memory_order_acquire); +} + +void gc::ConcurrentMarkAndSweep::ThreadData::clearMarkFlags() { + published_.store(false, std::memory_order_relaxed); + cooperative_.store(false, std::memory_order_relaxed); + rootSetLocked_.store(false, std::memory_order_release); +} + +mm::ThreadData& gc::ConcurrentMarkAndSweep::ThreadData::commonThreadData() const { + return threadData_; } #ifndef CUSTOM_ALLOCATOR gc::ConcurrentMarkAndSweep::ConcurrentMarkAndSweep( mm::ObjectFactory& objectFactory, mm::ExtraObjectDataFactory& extraObjectDataFactory, - gcScheduler::GCScheduler& gcScheduler) noexcept : + gcScheduler::GCScheduler& gcScheduler, + bool mutatorsCooperate, std::size_t auxGCThreads) noexcept : objectFactory_(objectFactory), extraObjectDataFactory_(extraObjectDataFactory), #else -gc::ConcurrentMarkAndSweep::ConcurrentMarkAndSweep(gcScheduler::GCScheduler& gcScheduler) noexcept : +gc::ConcurrentMarkAndSweep::ConcurrentMarkAndSweep( + gcScheduler::GCScheduler& gcScheduler, + bool mutatorsCooperate, std::size_t auxGCThreads) noexcept : #endif gcScheduler_(gcScheduler), finalizerProcessor_([this](int64_t epoch) { GCHandle::getByEpoch(epoch).finalizersDone(); state_.finalized(epoch); - }) { - gcThread_ = ScopedThread(ScopedThread::attributes().name("GC thread"), [this] { - while (true) { - auto epoch = state_.waitScheduled(); - if (epoch.has_value()) { - PerformFullGC(*epoch); - } else { - break; - } - } - }); - markingBehavior_ = kotlin::compiler::gcMarkSingleThreaded() ? MarkingBehavior::kDoNotMark : MarkingBehavior::kMarkOwnStack; - RuntimeLogDebug({kTagGC}, "Concurrent Mark & Sweep GC initialized"); + }), + markDispatcher_(mutatorsCooperate), + mainThread_(createGCThread("Main GC thread", [this] { mainGCThreadBody(); })) +{ + for (std::size_t i = 0; i < auxGCThreads; ++i) { + auxThreads_.emplace_back(createGCThread("Auxiliary GC thread", [this] { auxiliaryGCThreadBody(); })); + } + RuntimeLogInfo({kTagGC}, "Parallel Mark & Concurrent Sweep GC initialized"); } gc::ConcurrentMarkAndSweep::~ConcurrentMarkAndSweep() { @@ -131,18 +177,42 @@ bool gc::ConcurrentMarkAndSweep::FinalizersThreadIsRunning() noexcept { return finalizerProcessor_.IsRunning(); } -void gc::ConcurrentMarkAndSweep::SetMarkingBehaviorForTests(MarkingBehavior markingBehavior) noexcept { - markingBehavior_ = markingBehavior; +void gc::ConcurrentMarkAndSweep::mainGCThreadBody() { + while (true) { + auto epoch = state_.waitScheduled(); + if (epoch.has_value()) { + PerformFullGC(*epoch); + } else { + break; + } + } + markDispatcher_.requestShutdown(); +} + +void gc::ConcurrentMarkAndSweep::auxiliaryGCThreadBody() { + RuntimeAssert(!compiler::gcMarkSingleThreaded(), "Should not reach here during single threaded mark"); + while (!markDispatcher_.shutdownRequested()) { + markDispatcher_.runAuxiliary(); + } } void gc::ConcurrentMarkAndSweep::PerformFullGC(int64_t epoch) noexcept { + std::unique_lock mainGCLock(gcMutex); auto gcHandle = GCHandle::create(epoch); - SetMarkingRequested(epoch); + + markDispatcher_.beginMarkingEpoch(gcHandle); + GCLogDebug(epoch, "Main GC requested marking in mutators"); + + // Request STW bool didSuspend = mm::RequestThreadsSuspension(); RuntimeAssert(didSuspend, "Only GC thread can request suspension"); gcHandle.suspensionRequested(); - WaitForThreadsReadyToMark(); + // TODO (WaitForThreadsReadyToMark()) + RuntimeAssert(!kotlin::mm::IsCurrentThreadRegistered(), "GC must run on unregistered thread"); + + markDispatcher_.waitForThreadsPauseMutation(); + GCLogDebug(epoch, "All threads have paused mutation"); gcHandle.threadsAreSuspended(); #ifdef CUSTOM_ALLOCATOR @@ -158,10 +228,9 @@ void gc::ConcurrentMarkAndSweep::PerformFullGC(int64_t epoch) noexcept { state_.start(epoch); - CollectRootSetAndStartMarking(gcHandle); + markDispatcher_.runMainInSTW(); - // Can be unsafe, because we've stopped the world. - gc::Mark(gcHandle, markQueue_); + markDispatcher_.endMarkingEpoch(); mm::WaitForThreadsSuspension(); @@ -170,6 +239,7 @@ void gc::ConcurrentMarkAndSweep::PerformFullGC(int64_t epoch) noexcept { // would not publish into the global state at an unexpected time. std::optional extraObjectFactoryIterable = extraObjectDataFactory_.LockForIter(); std::optional objectFactoryIterable = objectFactory_.LockForIter(); + checkMarkCorrectness(*objectFactoryIterable); #endif if (compiler::concurrentWeakSweep()) { @@ -214,28 +284,14 @@ void gc::ConcurrentMarkAndSweep::PerformFullGC(int64_t epoch) noexcept { finalizerProcessor_.ScheduleTasks(std::move(finalizerQueue), epoch); } -void gc::ConcurrentMarkAndSweep::SetMarkingRequested(uint64_t epoch) noexcept { - markingRequested = markingBehavior_ == MarkingBehavior::kMarkOwnStack; - markingEpoch = epoch; -} - -void gc::ConcurrentMarkAndSweep::WaitForThreadsReadyToMark() noexcept { - RuntimeAssert(!kotlin::mm::IsCurrentThreadRegistered(), "GC must run on unregistered thread"); - mm::ThreadRegistry::Instance().waitAllThreads([](mm::ThreadData& thread) noexcept { - return thread.suspensionData().suspendedOrNative() || thread.gc().impl().gc().marking_.load(); - }); -} - -void gc::ConcurrentMarkAndSweep::CollectRootSetAndStartMarking(GCHandle gcHandle) noexcept { - std::unique_lock lock(markingMutex); - markingRequested = false; - gc::collectRootSet( - gcHandle, - markQueue_, - [](mm::ThreadData& thread) { - return !thread.gc().impl().gc().marking_.load(); - } - ); - RuntimeLogDebug({kTagGC}, "Requesting marking in threads"); - markingCondVar.notify_all(); -} +void gc::ConcurrentMarkAndSweep::reconfigure(std::size_t maxParallelism, bool mutatorsCooperate, std::size_t auxGCThreads) noexcept { + if (compiler::gcMarkSingleThreaded()) { + RuntimeCheck(auxGCThreads == 0, "Auxiliary GC threads must not be created with gcMarkSingleThread"); + return; + } + std::unique_lock mainGCLock(gcMutex); + markDispatcher_.reset(maxParallelism, mutatorsCooperate, [this] { auxThreads_.clear(); }); + for (std::size_t i = 0; i < auxGCThreads; ++i) { + auxThreads_.emplace_back(createGCThread("Auxiliary GC thread", [this] { auxiliaryGCThreadBody(); })); + } +} \ No newline at end of file diff --git a/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.hpp b/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.hpp index b64e5e99039..854e48871d4 100644 --- a/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.hpp +++ b/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.hpp @@ -23,6 +23,8 @@ #include "Types.h" #include "Utils.hpp" #include "std_support/Memory.hpp" +#include "MarkStack.hpp" +#include "ParallelMark.hpp" #ifdef CUSTOM_ALLOCATOR #include "CustomAllocator.hpp" @@ -37,48 +39,15 @@ namespace gc { // TODO: Also make marking run concurrently with Kotlin threads. class ConcurrentMarkAndSweep : private Pinned { public: - class ObjectData { - public: - bool tryMark() noexcept { - return trySetNext(reinterpret_cast(1)); - } - - bool marked() const noexcept { return next() != nullptr; } - - bool tryResetMark() noexcept { - if (next() == nullptr) return false; - next_.store(nullptr, std::memory_order_relaxed); - return true; - } - - private: - friend struct DefaultIntrusiveForwardListTraits; - - ObjectData* next() const noexcept { return next_.load(std::memory_order_relaxed); } - void setNext(ObjectData* next) noexcept { - RuntimeAssert(next, "next cannot be nullptr"); - next_.store(next, std::memory_order_relaxed); - } - bool trySetNext(ObjectData* next) noexcept { - RuntimeAssert(next, "next cannot be nullptr"); - ObjectData* expected = nullptr; - return next_.compare_exchange_strong(expected, next, std::memory_order_relaxed); - } - - std::atomic next_ = nullptr; - }; - - enum MarkingBehavior { kMarkOwnStack, kDoNotMark }; - - using MarkQueue = intrusive_forward_list; class ThreadData : private Pinned { public: - using ObjectData = ConcurrentMarkAndSweep::ObjectData; - + using ObjectData = mark::ObjectData; using Allocator = AllocatorWithGC; - explicit ThreadData(ConcurrentMarkAndSweep& gc, mm::ThreadData& threadData) noexcept : threadData_(threadData) {} + explicit ThreadData(ConcurrentMarkAndSweep& gc, mm::ThreadData& threadData) noexcept + : gc_(gc), threadData_(threadData) {} + ~ThreadData() = default; void OnOOM(size_t size) noexcept; @@ -91,13 +60,27 @@ public: BarriersThreadData& barriers() noexcept { return barriers_; } + bool tryLockRootSet(); + void beginCooperation(); + bool cooperative() const; + void publish(); + bool published() const; + void clearMarkFlags(); + + mm::ThreadData& commonThreadData() const; + private: friend ConcurrentMarkAndSweep; + ConcurrentMarkAndSweep& gc_; mm::ThreadData& threadData_; - std::atomic marking_; BarriersThreadData barriers_; + + std::atomic rootSetLocked_ = false; + std::atomic published_ = false; + std::atomic cooperative_ = false; }; + using ObjectData = ThreadData::ObjectData; using Allocator = ThreadData::Allocator; #ifndef CUSTOM_ALLOCATOR @@ -109,22 +92,22 @@ public: #endif #ifdef CUSTOM_ALLOCATOR - explicit ConcurrentMarkAndSweep(gcScheduler::GCScheduler& scheduler) noexcept; + explicit ConcurrentMarkAndSweep(gcScheduler::GCScheduler& scheduler, + bool mutatorsCooperate, std::size_t auxGCThreads) noexcept; #else ConcurrentMarkAndSweep( mm::ObjectFactory& objectFactory, mm::ExtraObjectDataFactory& extraObjectDataFactory, - gcScheduler::GCScheduler& scheduler) noexcept; + gcScheduler::GCScheduler& scheduler, + bool mutatorsCooperate, std::size_t auxGCThreads) noexcept; #endif ~ConcurrentMarkAndSweep(); void StartFinalizerThreadIfNeeded() noexcept; void StopFinalizerThreadIfRunning() noexcept; bool FinalizersThreadIsRunning() noexcept; - void SetMarkingBehaviorForTests(MarkingBehavior markingBehavior) noexcept; - void SetMarkingRequested(uint64_t epoch) noexcept; - void WaitForThreadsReadyToMark() noexcept; - void CollectRootSetAndStartMarking(GCHandle gcHandle) noexcept; + + void reconfigure(std::size_t maxParallelism, bool mutatorsCooperate, size_t auxGCThreads) noexcept; #ifdef CUSTOM_ALLOCATOR alloc::Heap& heap() noexcept { return heap_; } @@ -133,6 +116,8 @@ public: GCStateHolder& state() noexcept { return state_; } private: + void mainGCThreadBody(); + void auxiliaryGCThreadBody(); void PerformFullGC(int64_t epoch) noexcept; #ifndef CUSTOM_ALLOCATOR @@ -144,44 +129,12 @@ private: gcScheduler::GCScheduler& gcScheduler_; GCStateHolder state_; - ScopedThread gcThread_; FinalizerProcessor finalizerProcessor_; - MarkQueue markQueue_; - MarkingBehavior markingBehavior_; + mark::ParallelMark markDispatcher_; + ScopedThread mainThread_; + std_support::vector auxThreads_; }; -namespace internal { -struct MarkTraits { - using MarkQueue = gc::ConcurrentMarkAndSweep::MarkQueue; - - static void clear(MarkQueue& queue) noexcept { queue.clear(); } - - static ObjHeader* tryDequeue(MarkQueue& queue) noexcept { - if (auto* top = queue.try_pop_front()) { - auto node = mm::ObjectFactory::NodeRef::From(*top); - return node->GetObjHeader(); - } - return nullptr; - } - - static bool tryEnqueue(MarkQueue& queue, ObjHeader* object) noexcept { - auto& objectData = mm::ObjectFactory::NodeRef::From(object).ObjectData(); - return queue.try_push_front(objectData); - } - - static bool tryMark(ObjHeader* object) noexcept { - auto& objectData = mm::ObjectFactory::NodeRef::From(object).ObjectData(); - return objectData.tryMark(); - } - - static void processInMark(MarkQueue& markQueue, ObjHeader* object) noexcept { - auto process = object->type_info()->processObjectInMark; - RuntimeAssert(process != nullptr, "Got null processObjectInMark for object %p", object); - process(static_cast(&markQueue), object); - } -}; -} // namespace internal - } // namespace gc } // namespace kotlin diff --git a/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweepTest.cpp b/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweepTest.cpp index 03731c72382..379f6d783e3 100644 --- a/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweepTest.cpp +++ b/kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweepTest.cpp @@ -209,11 +209,27 @@ test_support::RegularWeakReferenceImpl& InstallWeakReference(mm::ThreadData& thr return weakReference; } -class ConcurrentMarkAndSweepTest : public testing::TestWithParam { +struct ParallelismOptions { + std::size_t maxParallelism; + bool cooperativeMutators; + std::size_t auxGCThreads; +}; + +class ConcurrentMarkAndSweepTest : public testing::TestWithParam { public: ConcurrentMarkAndSweepTest() { - mm::GlobalData::Instance().gc().impl().gc().SetMarkingBehaviorForTests(GetParam()); + if (supportedConfiguration()) { + mm::GlobalData::Instance().gc().impl().gc().reconfigure(GetParam().maxParallelism, + GetParam().cooperativeMutators, + GetParam().auxGCThreads); + } + } + + void SetUp() override { + if (!supportedConfiguration()) { + GTEST_SKIP() << "Unsupported parallelism configuration"; + } } ~ConcurrentMarkAndSweepTest() { @@ -225,6 +241,10 @@ public: testing::MockFunction& finalizerHook() { return finalizerHooks_.finalizerHook(); } private: + bool supportedConfiguration() const { + return !compiler::gcMarkSingleThreaded() || (!GetParam().cooperativeMutators && GetParam().auxGCThreads == 0); + } + FinalizerHooksTestSupport finalizerHooks_; }; @@ -1172,60 +1192,23 @@ TEST_P(ConcurrentMarkAndSweepTest, FreeObjectWithFreeWeakReversedOrder) { f1.wait(); } -TEST_P(ConcurrentMarkAndSweepTest, MutatorsCanMarkOwnLocals) { - std_support::vector mutators(kDefaultThreadCount); - std_support::vector globals(kDefaultThreadCount); - std_support::vector locals(kDefaultThreadCount); - std_support::vector reachablesLocals(kDefaultThreadCount); - std_support::vector reachablesGlobals(kDefaultThreadCount); - - auto expandRootSet = [&globals, &locals, &reachablesLocals, &reachablesGlobals](mm::ThreadData& threadData, Mutator& mutator, int i) { - auto& global = mutator.AddGlobalRoot(); - auto& local = mutator.AddStackRoot(); - auto& reachableLocal = AllocateObject(threadData); - auto& reachableGlobal = AllocateObject(threadData); - local->field1 = reachableLocal.header(); - global->field1 = reachableGlobal.header(); - globals[i] = global.header(); - locals[i] = local.header(); - reachablesLocals[i] = reachableLocal.header(); - reachablesGlobals[i] = reachableGlobal.header(); - }; - - for (int i = 0; i < kDefaultThreadCount; ++i) { - mutators[i] - .Execute([i, expandRootSet](mm::ThreadData& threadData, Mutator& mutator) { expandRootSet(threadData, mutator, i); }) - .wait(); - } - - std_support::vector> gcFutures(kDefaultThreadCount); - - mm::GlobalData::Instance().gc().impl().gc().SetMarkingRequested(0); - - for (int i = 0; i < kDefaultThreadCount; ++i) { - gcFutures[i] = mutators[i] - .Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().impl().gc().OnSuspendForGC(); }); - } - - if (GetParam() == gc::ConcurrentMarkAndSweep::kMarkOwnStack) { - mm::GlobalData::Instance().gc().impl().gc().WaitForThreadsReadyToMark(); - mm::GlobalData::Instance().gc().impl().gc().CollectRootSetAndStartMarking(gc::GCHandle::createFakeForTests()); - } - - for (int i = 0; i < kDefaultThreadCount; ++i) { - gcFutures[i].wait(); - // Verify that threads marked their own locals (but not their globals) according to the configured marking behavior: - ASSERT_THAT(IsMarked(reachablesLocals[i]), GetParam() == kotlin::gc::ConcurrentMarkAndSweep::kMarkOwnStack); - ASSERT_THAT(IsMarked(reachablesGlobals[i]), false); - } - - for (auto& future : gcFutures) { - future.wait(); - } -} - INSTANTIATE_TEST_SUITE_P(, - ConcurrentMarkAndSweepTest, - testing::Values(gc::ConcurrentMarkAndSweep::MarkingBehavior::kDoNotMark, gc::ConcurrentMarkAndSweep::MarkingBehavior::kMarkOwnStack), - [] (const testing::TestParamInfo& behavior) { return (behavior.param == gc::ConcurrentMarkAndSweep::MarkingBehavior::kDoNotMark) ? "SingleThreadedMarking" : "MutatorsMarkOwnStack"; }); + ConcurrentMarkAndSweepTest, + testing::Values( + ParallelismOptions{kDefaultThreadCount * 3, false, 0}, + ParallelismOptions{kDefaultThreadCount * 3, true, 0}, + ParallelismOptions{kDefaultThreadCount * 3, false, kDefaultThreadCount}, + ParallelismOptions{kDefaultThreadCount * 3, true, kDefaultThreadCount}, + + ParallelismOptions{kDefaultThreadCount / 2, true, kDefaultThreadCount}, + ParallelismOptions{kDefaultThreadCount / 2 * 3, true, kDefaultThreadCount} + ), + [] (const testing::TestParamInfo& paramInfo) { + using namespace std::string_literals; + auto base = "Mark"s; + auto parallelism = std::to_string(paramInfo.param.maxParallelism) + "Parallel"; + auto withMutators = paramInfo.param.cooperativeMutators ? "WithMutators" : ""; + auto withAux = paramInfo.param.auxGCThreads > 0 ? "WithGCThreads" : ""; + return base + parallelism + withMutators + withAux; + }); diff --git a/kotlin-native/runtime/src/gc/cms/cpp/GCImpl.cpp b/kotlin-native/runtime/src/gc/cms/cpp/GCImpl.cpp index 887309c7384..bef983babee 100644 --- a/kotlin-native/runtime/src/gc/cms/cpp/GCImpl.cpp +++ b/kotlin-native/runtime/src/gc/cms/cpp/GCImpl.cpp @@ -11,6 +11,7 @@ #include "MarkAndSweepUtils.hpp" #include "ObjectOps.hpp" #include "ThreadSuspension.hpp" +#include "MarkStack.hpp" #include "std_support/Memory.hpp" using namespace kotlin; @@ -118,17 +119,17 @@ bool gc::GC::FinalizersThreadIsRunning() noexcept { // static ALWAYS_INLINE void gc::GC::processObjectInMark(void* state, ObjHeader* object) noexcept { - gc::internal::processObjectInMark(state, object); + gc::internal::processObjectInMark(state, object); } // static ALWAYS_INLINE void gc::GC::processArrayInMark(void* state, ArrayHeader* array) noexcept { - gc::internal::processArrayInMark(state, array); + gc::internal::processArrayInMark(state, array); } // static ALWAYS_INLINE void gc::GC::processFieldInMark(void* state, ObjHeader* field) noexcept { - gc::internal::processFieldInMark(state, field); + gc::internal::processFieldInMark(state, field); } int64_t gc::GC::Schedule() noexcept { diff --git a/kotlin-native/runtime/src/gc/cms/cpp/GCImpl.hpp b/kotlin-native/runtime/src/gc/cms/cpp/GCImpl.hpp index 10f0657b98e..375fff5f617 100644 --- a/kotlin-native/runtime/src/gc/cms/cpp/GCImpl.hpp +++ b/kotlin-native/runtime/src/gc/cms/cpp/GCImpl.hpp @@ -23,9 +23,9 @@ using GCImpl = ConcurrentMarkAndSweep; class GC::Impl : private Pinned { public: #ifdef CUSTOM_ALLOCATOR - explicit Impl(gcScheduler::GCScheduler& gcScheduler) noexcept : gc_(gcScheduler) {} + explicit Impl(gcScheduler::GCScheduler& gcScheduler) noexcept : gc_(gcScheduler, compiler::gcMutatorsCooperate(), compiler::auxGCThreads()) {} #else - explicit Impl(gcScheduler::GCScheduler& gcScheduler) noexcept : gc_(objectFactory_, extraObjectDataFactory_, gcScheduler) {} + explicit Impl(gcScheduler::GCScheduler& gcScheduler) noexcept : gc_(objectFactory_, extraObjectDataFactory_, gcScheduler, compiler::gcMutatorsCooperate(), compiler::auxGCThreads()) {} #endif #ifndef CUSTOM_ALLOCATOR diff --git a/kotlin-native/runtime/src/gc/cms/cpp/MarkStack.hpp b/kotlin-native/runtime/src/gc/cms/cpp/MarkStack.hpp new file mode 100644 index 00000000000..e73fe15c203 --- /dev/null +++ b/kotlin-native/runtime/src/gc/cms/cpp/MarkStack.hpp @@ -0,0 +1,57 @@ +#pragma once + +#include +#include + +#include "Allocator.hpp" +#include "IntrusiveList.hpp" +#include "ObjectFactory.hpp" +#include "Types.h" +#include "Utils.hpp" +#include "std_support/Memory.hpp" + +namespace kotlin::gc::mark { + +class ObjectData { + struct ObjectFactoryTraits { + using ObjectData = ObjectData; + class Allocator; + }; +public: + using ObjectFactory = mm::ObjectFactory; + + bool tryMark() noexcept { + return trySetNext(reinterpret_cast(1)); + } + + bool marked() const noexcept { return next() != nullptr; } + + bool tryResetMark() noexcept { + if (next() == nullptr) return false; + next_.store(nullptr, std::memory_order_relaxed); + return true; + } + + ObjHeader* objHeader() noexcept { // FIXME const + return ObjectFactory::NodeRef::From(*this).GetObjHeader(); + } + +private: + friend struct DefaultIntrusiveForwardListTraits; + + ObjectData* next() const noexcept { return next_.load(std::memory_order_relaxed); } + void setNext(ObjectData* next) noexcept { + RuntimeAssert(next, "next cannot be nullptr"); + next_.store(next, std::memory_order_relaxed); + } + bool trySetNext(ObjectData* next) noexcept { + RuntimeAssert(next, "next cannot be nullptr"); + ObjectData* expected = nullptr; + return next_.compare_exchange_strong(expected, next, std::memory_order_relaxed); + } + + std::atomic next_ = nullptr; +}; + + +} // namespace kotlin::gc::mark \ No newline at end of file diff --git a/kotlin-native/runtime/src/gc/cms/cpp/ParallelMark.cpp b/kotlin-native/runtime/src/gc/cms/cpp/ParallelMark.cpp new file mode 100644 index 00000000000..f302b851095 --- /dev/null +++ b/kotlin-native/runtime/src/gc/cms/cpp/ParallelMark.cpp @@ -0,0 +1,243 @@ +#include "ParallelMark.hpp" + +#include "MarkStack.hpp" +#include "MarkAndSweepUtils.hpp" +#include "GCStatistics.hpp" +#include "Utils.hpp" +#include "std_support/Memory.hpp" + +// required to access gc thread data +#include "GCImpl.hpp" + +using namespace kotlin; + +namespace { + +template +void spinWait(Cond&& until) { + while (!until()) { + std::this_thread::yield(); + } +} + +} // namespace + +bool gc::mark::MarkPacer::is(gc::mark::MarkPacer::Phase phase) const { + return phase_.load(std::memory_order_relaxed) == phase; +} + +void gc::mark::MarkPacer::begin(gc::mark::MarkPacer::Phase phase) { + { + std::unique_lock lock(mutex_); + phase_.store(phase, std::memory_order_relaxed); + } + cond_.notify_all(); +} + +void gc::mark::MarkPacer::wait(gc::mark::MarkPacer::Phase phase) { + if (phase_.load(std::memory_order_relaxed) >= phase) return; + std::unique_lock lock(mutex_); + cond_.wait(lock, [=]() { return phase_ >= phase; }); +} + +void gc::mark::MarkPacer::beginEpoch(uint64_t epoch) { + epoch_ = epoch; + begin(Phase::kReady); + GCLogDebug(epoch_.load(), "Mark is ready to recruit workers in a new epoch."); +} + +void gc::mark::MarkPacer::waitNewEpochReadyOrShutdown() const { + std::unique_lock lock(mutex_); + cond_.wait(lock, [this]() { return (phase_.load(std::memory_order_relaxed) >= Phase::kReady); }); +} + +void gc::mark::MarkPacer::waitEpochFinished(uint64_t currentEpoch) const { + std::unique_lock lock(mutex_); + cond_.wait(lock, [this, currentEpoch]() { + return is(Phase::kIdle) || is(Phase::kShutdown) || epoch_.load(std::memory_order_relaxed) > currentEpoch; + }); +} + +bool gc::mark::MarkPacer::acceptingNewWorkers() const { + return Phase::kReady <= phase_ && phase_ <= Phase::kParallelMark; +} + + +gc::mark::ParallelMark::ParallelMark(bool mutatorsCooperate) { + std::size_t maxParallelism = std::thread::hardware_concurrency(); + if (maxParallelism == 0) { + maxParallelism = std::numeric_limits::max(); + } + setParallelismLevel(maxParallelism, mutatorsCooperate); +} + +void gc::mark::ParallelMark::beginMarkingEpoch(gc::GCHandle gcHandle) { + gcHandle_ = gcHandle; + + lockedMutatorsList_ = mm::ThreadRegistry::Instance().LockForIter(); + + parallelProcessor_.construct(); + + if (!compiler::gcMarkSingleThreaded()) { + std::unique_lock guard(workerCreationMutex_); + pacer_.beginEpoch(gcHandle.getEpoch()); + // main worker is always accounted, so others would not be able to exhaust all the parallelism before main is instantiated + activeWorkersCount_ = 1; + } +} + +void gc::mark::ParallelMark::waitForThreadsPauseMutation() noexcept { + RuntimeAssert(!kotlin::mm::IsCurrentThreadRegistered(), "Dispatcher thread must not be registered"); + spinWait([this] { + return allMutators([](mm::ThreadData& mut) { + return mm::isSuspendedOrNative(mut) || mut.gc().impl().gc().cooperative(); + }); + }); +} + +void gc::mark::ParallelMark::endMarkingEpoch() { + if (!compiler::gcMarkSingleThreaded()) { + // We must now wait for every worker to finish the Mark procedure: + // wake up from possible waiting, publish statistics, etc. + // Only then it's safe to destroy the parallelProcessor and proceed to other GC tasks such as sweep. + spinWait([=]() { return activeWorkersCount_.load(std::memory_order_relaxed) == 0; }); + + std::unique_lock guard(workerCreationMutex_); + RuntimeAssert(activeWorkersCount_ == 0, "All the workers must already finish"); + pacer_.begin(MarkPacer::Phase::kIdle); + } + parallelProcessor_.destroy(); + resetMutatorFlags(); + lockedMutatorsList_ = std::nullopt; +} + +void gc::mark::ParallelMark::runMainInSTW() { + if (compiler::gcMarkSingleThreaded()) { + ParallelProcessor::Worker worker(*parallelProcessor_); + gc::collectRootSet(gcHandle(), worker, [] (mm::ThreadData&) { return true; }); + gc::Mark(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(gcHandle(), mainWorker); + + pacer_.begin(MarkPacer::Phase::kParallelMark); + parallelMark(mainWorker); + } +} + +void gc::mark::ParallelMark::runOnMutator(mm::ThreadData& mutatorThread) { + if (compiler::gcMarkSingleThreaded() || !mutatorsCooperate_) return; + + auto epoch = gcHandle().getEpoch(); + auto parallelWorker = createWorker(); + if (parallelWorker) { + auto& gcData = mutatorThread.gc().impl().gc(); + gcData.beginCooperation(); + GCLogDebug(epoch, "Mutator thread cooperates in marking"); + + tryCollectRootSet(mutatorThread, *parallelWorker); + + completeRootSetAndMark(*parallelWorker); + } +} + +void gc::mark::ParallelMark::runAuxiliary() { + RuntimeAssert(!compiler::gcMarkSingleThreaded(), "Should not reach here during single threaded mark"); + + pacer_.waitNewEpochReadyOrShutdown(); + if (pacer_.is(MarkPacer::Phase::kShutdown)) return; + + auto curEpoch = gcHandle().getEpoch(); + auto parallelWorker = createWorker(); + if (parallelWorker) { + completeRootSetAndMark(*parallelWorker); + } + + pacer_.waitEpochFinished(curEpoch); +} + +void gc::mark::ParallelMark::requestShutdown() { + pacer_.begin(MarkPacer::Phase::kShutdown); +} + +bool gc::mark::ParallelMark::shutdownRequested() const { + return pacer_.is(MarkPacer::Phase::kShutdown); +} + + +gc::GCHandle& gc::mark::ParallelMark::gcHandle() { + RuntimeAssert(gcHandle_.isValid(), "GCHandle must be initialized"); + return gcHandle_; +} + +void gc::mark::ParallelMark::setParallelismLevel(size_t maxParallelism, bool mutatorsCooperate) { + RuntimeCheck(maxParallelism > 0, "Parallelism level can't be 0"); + maxParallelism_ = maxParallelism; + mutatorsCooperate_ = mutatorsCooperate; + RuntimeLogInfo({kTagGC}, + "Set up parallel mark with maxParallelism = %zu and %s" "cooperative mutators", + maxParallelism_, (mutatorsCooperate_ ? "" : "non-")); +} + +void gc::mark::ParallelMark::completeRootSetAndMark(ParallelProcessor::Worker& parallelWorker) { + pacer_.wait(MarkPacer::Phase::kRootSet); + completeMutatorsRootSet(parallelWorker); + pacer_.wait(MarkPacer::Phase::kParallelMark); + parallelMark(parallelWorker); +} + +void gc::mark::ParallelMark::completeMutatorsRootSet(MarkTraits::MarkQueue& markQueue) { + // workers compete for mutators to collect their root set + for (auto& thread: *lockedMutatorsList_) { + tryCollectRootSet(thread, markQueue); + } +} + +void gc::mark::ParallelMark::tryCollectRootSet(mm::ThreadData& thread, MarkTraits::MarkQueue& markQueue) { + auto& gcData = thread.gc().impl().gc(); + if (!gcData.tryLockRootSet()) return; + + GCLogDebug(gcHandle().getEpoch(), "Root set collection on thread %d for thread %d", + konan::currentThreadId(), thread.threadId()); + gcData.publish(); + collectRootSetForThread(gcHandle(), markQueue, thread); +} + +void gc::mark::ParallelMark::parallelMark(ParallelProcessor::Worker& worker) { + GCLogDebug(gcHandle().getEpoch(), "Mark loop has begun"); + Mark(gcHandle(), worker); + + std::unique_lock guard(workerCreationMutex_); + activeWorkersCount_.fetch_sub(1, std::memory_order_relaxed); +} + +std::optional 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_); +} + +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(); + } +} diff --git a/kotlin-native/runtime/src/gc/cms/cpp/ParallelMark.hpp b/kotlin-native/runtime/src/gc/cms/cpp/ParallelMark.hpp new file mode 100644 index 00000000000..8568ad5f8d0 --- /dev/null +++ b/kotlin-native/runtime/src/gc/cms/cpp/ParallelMark.hpp @@ -0,0 +1,181 @@ +#pragma once + +#include +#include + +#include "GCStatistics.hpp" +#include "MarkStack.hpp" +#include "std_support/Vector.hpp" +#include "ThreadRegistry.hpp" +#include "Utils.hpp" +#include "ParallelProcessor.hpp" +#include "ManuallyScoped.hpp" + +namespace kotlin::gc::mark { + +class MarkPacer : private Pinned { +public: + enum class Phase { + /** Mark is not in progress. */ + kIdle, + /** + * MarkDispatcher is ready to recruit new workers. + * + * In case of cooperative mark mutator threads are welcome to mark their own root sets. + * Each thread is free to start as soon as it reaches a safe point. + * No need to wait for others. + */ + kReady, + /** + * All mutator threads must be in a safe state at this point: + * 1) Suspended on a safe point; + * 2) In the native code; + * 3) Registered as cooperative markers during previous phase. + * + * Now all the GC workers are summoned to participate in a root set collection. + */ + kRootSet, + /** + * Root set is collected. No more workers can be instantiated, time to begin parallel mark. + * Parallel mark can't stop before all the created workers begin the marking. + */ + kParallelMark, + /** A shutdown was requested. There is nothing more to wait for. */ + kShutdown, + }; + + bool is(Phase phase) const; + void begin(Phase phase); + void wait(Phase phase); + + void beginEpoch(uint64_t epoch); + void waitNewEpochReadyOrShutdown() const; + void waitEpochFinished(uint64_t epoch) const; + + bool acceptingNewWorkers() const; + +private: + std::atomic epoch_ = 0; + std::atomic 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; + // work balancing parameters were chosen pretty arbitrary + using ParallelProcessor = ParallelProcessor; +public: + class MarkTraits { + public: + using MarkQueue = ParallelProcessor::Worker; + using ObjectFactory = ObjectData::ObjectFactory; + + static void clear(MarkQueue& queue) noexcept { + RuntimeAssert(queue.localEmpty(), "Mark queue must be empty"); + } + + static ALWAYS_INLINE ObjHeader* tryDequeue(MarkQueue& queue) noexcept { + auto* obj = compiler::gcMarkSingleThreaded() ? queue.tryPopLocal() : queue.tryPop(); + if (obj) { + auto node = ObjectFactory::NodeRef::From(*obj); + return node->GetObjHeader(); + } + return nullptr; + } + + static ALWAYS_INLINE bool tryEnqueue(MarkQueue& queue, ObjHeader* object) noexcept { + auto& objectData = ObjectFactory::NodeRef::From(object).ObjectData(); + return compiler::gcMarkSingleThreaded() ? queue.tryPushLocal(objectData) : queue.tryPush(objectData); + } + + static ALWAYS_INLINE bool tryMark(ObjHeader* object) noexcept { + auto& objectData = ObjectFactory::NodeRef::From(object).ObjectData(); + return objectData.tryMark(); + } + + static ALWAYS_INLINE void processInMark(MarkQueue& markQueue, ObjHeader* object) noexcept { + auto process = object->type_info()->processObjectInMark; + RuntimeAssert(process != nullptr, "Got null processObjectInMark for object %p", object); + process(static_cast(&markQueue), object); + } + }; + + ParallelMark(bool mutatorsCooperate); + + void beginMarkingEpoch(gc::GCHandle gcHandle); + void waitForThreadsPauseMutation() noexcept; + void endMarkingEpoch(); + + /** To be run by a single "main" GC thread during STW. */ + void runMainInSTW(); + + /** + * To be run by mutator threads that would like to participate in mark. + * Will wait for STW detection by a "main" routine. + */ + void runOnMutator(mm::ThreadData& mutatorThread); + + /** + * To be run by auxiliary GC threads. + * Will wait for STW detection by a "main" routine. + */ + void runAuxiliary(); + + void requestShutdown(); + bool shutdownRequested() const; + + template + 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 + 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 createWorker(); + + void resetMutatorFlags(); + + std::size_t maxParallelism_ = 1; + bool mutatorsCooperate_ = false; + + GCHandle gcHandle_ = GCHandle::invalid(); + MarkPacer pacer_; + std::optional lockedMutatorsList_; + ManuallyScoped parallelProcessor_; + + std::mutex workerCreationMutex_; + std::atomic activeWorkersCount_ = 0; +}; + +} // namespace kotlin::gc::mark + diff --git a/kotlin-native/runtime/src/gc/common/cpp/GCStatistics.cpp b/kotlin-native/runtime/src/gc/common/cpp/GCStatistics.cpp index e28f41b5870..cad3841a186 100644 --- a/kotlin-native/runtime/src/gc/common/cpp/GCStatistics.cpp +++ b/kotlin-native/runtime/src/gc/common/cpp/GCStatistics.cpp @@ -12,7 +12,7 @@ #include "ThreadData.hpp" #include "std_support/Optional.hpp" #include -#include +#include using namespace kotlin; @@ -153,15 +153,18 @@ GCHandle GCHandle::create(uint64_t epoch) { current.epoch = static_cast(epoch); current.startTime = static_cast(konan::getTimeNanos()); if (last.endTime) { - GCLogInfo(epoch, "Started. Time since last GC %" PRIu64 " microseconds.", *current.startTime - *last.endTime); + auto time = (*current.startTime - *last.endTime) / 1000; + GCLogInfo(epoch, "Started. Time since last GC %" PRIu64 " microseconds.", time); } else { GCLogInfo(epoch, "Started."); } current.memoryUsageBefore.heap = currentHeapUsage(); return getByEpoch(epoch); } -GCHandle GCHandle::createFakeForTests() { return getByEpoch(std::numeric_limits::max()); } +GCHandle GCHandle::createFakeForTests() { return getByEpoch(invalid().getEpoch() - 1); } GCHandle GCHandle::getByEpoch(uint64_t epoch) { + GCHandle handle{epoch}; + RuntimeAssert(handle.isValid(), "Must be valid"); return GCHandle{epoch}; } @@ -174,11 +177,17 @@ std::optional gc::GCHandle::currentEpoch() noexcept { return std::nullopt; } +GCHandle GCHandle::invalid() { + return GCHandle{std::numeric_limits::max()}; +} void GCHandle::ClearForTests() { std::lock_guard guard(lock); current = {}; last = {}; } +bool GCHandle::isValid() const { + return epoch_ != GCHandle::invalid().epoch_; +} void GCHandle::finished() { std::lock_guard guard(lock); if (auto* stat = statByEpoch(epoch_)) { diff --git a/kotlin-native/runtime/src/gc/common/cpp/GCStatistics.hpp b/kotlin-native/runtime/src/gc/common/cpp/GCStatistics.hpp index 1df9fc169e7..e4447086b6c 100644 --- a/kotlin-native/runtime/src/gc/common/cpp/GCStatistics.hpp +++ b/kotlin-native/runtime/src/gc/common/cpp/GCStatistics.hpp @@ -9,6 +9,7 @@ #include #include "Common.h" +#include "Logging.hpp" #include "Porting.h" #include "Utils.hpp" #include "std_support/Optional.hpp" @@ -135,9 +136,11 @@ public: static GCHandle createFakeForTests(); static GCHandle getByEpoch(uint64_t epoch); static std::optional currentEpoch() noexcept; + static GCHandle invalid(); static void ClearForTests(); uint64_t getEpoch() { return epoch_; } + bool isValid() const; void finished(); void finalizersDone(); void finalizersScheduled(uint64_t finalizersCount); diff --git a/kotlin-native/runtime/src/gc/common/cpp/MarkAndSweepUtils.hpp b/kotlin-native/runtime/src/gc/common/cpp/MarkAndSweepUtils.hpp index f23ad879cb0..953f81fbf91 100644 --- a/kotlin-native/runtime/src/gc/common/cpp/MarkAndSweepUtils.hpp +++ b/kotlin-native/runtime/src/gc/common/cpp/MarkAndSweepUtils.hpp @@ -35,25 +35,20 @@ void processFieldInMark(void* state, ObjHeader* field) noexcept { template void processObjectInMark(void* state, ObjHeader* object) noexcept { - auto* typeInfo = object->type_info(); - RuntimeAssert(typeInfo != theArrayTypeInfo, "Must not be an array of objects"); - for (int i = 0; i < typeInfo->objOffsetsCount_; ++i) { - auto* field = *reinterpret_cast(reinterpret_cast(object) + typeInfo->objOffsets_[i]); - if (!field) continue; - processFieldInMark(state, field); - } + traverseClassObjectFields(object, [state] (ObjHeader** fieldLocation) noexcept { + if (auto field = *fieldLocation) { + processFieldInMark(state, field); + } + }); } template void processArrayInMark(void* state, ArrayHeader* array) noexcept { - RuntimeAssert(array->type_info() == theArrayTypeInfo, "Must be an array of objects"); - auto* begin = ArrayAddressOfElementAt(array, 0); - auto* end = ArrayAddressOfElementAt(array, array->count_); - for (auto* it = begin; it != end; ++it) { - auto* field = *it; - if (!field) continue; - processFieldInMark(state, field); - } + traverseArrayOfObjectsElements(array, [state] (ObjHeader** elemLocation) noexcept { + if (auto elem = *elemLocation) { + processFieldInMark(state, elem); + } + }); } template @@ -93,6 +88,11 @@ void processExtraObjectData(GCHandle::GCMarkScope& markHandle, typename Traits:: template void Mark(GCHandle handle, typename Traits::MarkQueue& markQueue) noexcept { auto markHandle = handle.mark(); + Mark(markHandle, markQueue); +} + +template +void Mark(GCHandle::GCMarkScope& markHandle, typename Traits::MarkQueue& markQueue) noexcept { while (ObjHeader* top = Traits::tryDequeue(markQueue)) { markHandle.addObject(); diff --git a/kotlin-native/runtime/src/main/cpp/BoundedQueue.hpp b/kotlin-native/runtime/src/main/cpp/BoundedQueue.hpp new file mode 100644 index 00000000000..52f5bd27847 --- /dev/null +++ b/kotlin-native/runtime/src/main/cpp/BoundedQueue.hpp @@ -0,0 +1,114 @@ +/* + * Copyright 2010-2023 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license + * that can be found in the LICENSE file. + */ + + +/* + * An implementation of Dmitry Vyukov's Bounded Multi-producer/multi-consumer bounded queue. + * + * Copyright (c) 2010-2011, Dmitry Vyukov. All rights reserved. + * Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: + * 1. Redistributions of source code must retain the above copyright notice, this list of + * conditions and the following disclaimer. + * 2. Redistributions in binary form must reproduce the above copyright notice, this list + * of conditions and the following disclaimer in the documentation and/or other materials + * provided with the distribution. + * THIS SOFTWARE IS PROVIDED BY DMITRY VYUKOV "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, + * THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL + * DMITRY VYUKOV OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES + * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) + * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) + * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * The views and conclusions contained in the software and documentation are those of the authors and should not be interpreted + * as representing official policies, either expressed or implied, of Dmitry Vyukov. + * + * TODO what about a binary distribution? + */ + +#pragma once + +#include + +#include "Utils.hpp" +#include "ManuallyScoped.hpp" + +namespace kotlin { + +/** + * A fixed-size concurrent multi-producer/multi-consumer queue. + * @tparam kCapacity must be a power of 2. + */ +template +class BoundedQueue : private Pinned { +public: + BoundedQueue() { + static_assert((kCapacity >= 2) && ((kCapacity & (kCapacity - 1)) == 0), "Queue capacity must be a power of 2"); + + for (size_t i = 0; i < kCapacity; ++i) { + buffer_[i].sequence_.store(i, std::memory_order_relaxed); + } + enqueuePos_.store(0, std::memory_order_relaxed); + dequeuePos_.store(0, std::memory_order_relaxed); + } + + bool enqueue(T&& value) { + Cell* cell; + std::size_t pos = enqueuePos_.load(std::memory_order_relaxed); + while (true) { + cell = &buffer_[pos & (kCapacity - 1)]; + std::size_t seq = cell->sequence_.load(std::memory_order_acquire); + std::intptr_t dif = static_cast(seq) - static_cast(pos); + if (dif == 0) { + if (enqueuePos_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed)) { + break; + } + } else if (dif < 0) { + return false; + } else { + pos = enqueuePos_.load(std::memory_order_relaxed); + } + } + cell->data_.construct(std::move(value)); + cell->sequence_.store(pos + 1, std::memory_order_release); + return true; + } + + std::optional dequeue() { + Cell* cell; + std::size_t pos = dequeuePos_.load(std::memory_order_relaxed); + while (true) { + cell = &buffer_[pos & (kCapacity - 1)]; + std::size_t seq = cell->sequence_.load(std::memory_order_acquire); + std::intptr_t dif = static_cast(seq) - static_cast(pos + 1); + if (dif == 0) { + if (dequeuePos_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed)) { + break; + } + } else if (dif < 0) { + return std::nullopt; + } else { + pos = dequeuePos_.load(std::memory_order_relaxed); + } + } + auto result = std::move(*cell->data_); + cell->data_.destroy(); + cell->sequence_.store(pos + kCapacity, std::memory_order_release); + return std::move(result); + } + +private: + struct Cell { + // TODO describe + std::atomic sequence_; + ManuallyScoped data_; + }; + + constexpr static auto kCacheLineSize = 128; + + alignas(kCacheLineSize) Cell buffer_[kCapacity]; + alignas(kCacheLineSize) std::atomic enqueuePos_; + alignas(kCacheLineSize) std::atomic dequeuePos_; +}; + +} // namespace kotlin diff --git a/kotlin-native/runtime/src/main/cpp/BoundedQueueTest.cpp b/kotlin-native/runtime/src/main/cpp/BoundedQueueTest.cpp new file mode 100644 index 00000000000..30bb7178f99 --- /dev/null +++ b/kotlin-native/runtime/src/main/cpp/BoundedQueueTest.cpp @@ -0,0 +1,130 @@ +/* + * Copyright 2010-2023 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license + * that can be found in the LICENSE file. + */ + +#include "gmock/gmock.h" +#include "gtest/gtest.h" + +#include + +#include "IntrusiveList.hpp" + +#include "ParallelProcessor.hpp" + +#include "std_support/Vector.hpp" +#include "SingleThreadExecutor.hpp" +#include "TestSupport.hpp" + +using ::testing::_; +using namespace kotlin; + +namespace { + +class Element { +public: + Element() : Element(0) {} + explicit Element(int value) : a(value), b(value), c(value), d(value) {} + + bool isValid() const { + return a == b && b == c && c == d; + } + +private: + std::size_t a; + std::size_t b; + std::size_t c; + std::size_t d; +}; + +} // namespace + + +TEST(BoundedQueueTest, ConcurrentEnqueue) { + constexpr auto kThreadCount = 16; + constexpr auto kElemsPerThread = 1024; + BoundedQueue queue; + + std::atomic start = false; + std::list threads; + for (int t = 0; t < kThreadCount; ++t) { + threads.emplace_back([&, t]() { + while (!start) { + std::this_thread::yield(); + } + for (int e = 0; e < kElemsPerThread; ++e) { + queue.enqueue(Element(t + e)); + } + }); + } + start = true; + + threads.clear(); + + while (auto elem = queue.dequeue()) { + EXPECT_TRUE(elem->isValid()); + } +} + +TEST(BoundedQueueTest, ConcurrentDequeue) { + constexpr auto kThreadCount = 16; + constexpr auto kElemsPerThread = 1024; + BoundedQueue queue; + + for (int i = 0; i < kThreadCount * kElemsPerThread; ++i) { + queue.enqueue(Element(i)); + } + + std::atomic start = false; + std::list threads; + for (int t = 0; t < kThreadCount; ++t) { + threads.emplace_back([&]() { + while (!start) { + std::this_thread::yield(); + } + while (auto elem = queue.dequeue()) { + EXPECT_TRUE(elem->isValid()); + } + }); + } + start = true; +} + +TEST(BoundedQueueTest, PingPongWithOverflow) { + constexpr auto kElemsPerThread = 1024; + BoundedQueue queue; + + std::atomic start = false; + std::list writers; + for (std::size_t t = 0; t < kDefaultThreadCount; ++t) { + writers.emplace_back([&]() { + while (!start) { + std::this_thread::yield(); + } + for (int i = 0; i < kElemsPerThread; ++i) { + while (!queue.enqueue(Element(i))) { + std::this_thread::yield(); + } + } + }); + } + + std::atomic allWritten = false; + std::list readers; + for (std::size_t t = 0; t < kDefaultThreadCount; ++t) { + readers.emplace_back([&]() { + while (!start) { + std::this_thread::yield(); + } + while (!allWritten) { + while (auto elem = queue.dequeue()) { + EXPECT_TRUE(elem->isValid()); + } + } + }); + } + + start = true; + writers.clear(); + allWritten = true; +} diff --git a/kotlin-native/runtime/src/main/cpp/CompilerConstants.cpp b/kotlin-native/runtime/src/main/cpp/CompilerConstants.cpp index e95cc2347cf..dbe578da169 100644 --- a/kotlin-native/runtime/src/main/cpp/CompilerConstants.cpp +++ b/kotlin-native/runtime/src/main/cpp/CompilerConstants.cpp @@ -19,7 +19,8 @@ using Kotlin_getSourceInfo_FunctionType = int(*)(void * /*addr*/, SourceInfo* /* * but can be changed after compiling caches. So use this way for variables, which will be rarely accessed. */ RUNTIME_WEAK int32_t Kotlin_destroyRuntimeMode = 1; -RUNTIME_WEAK int32_t Kotlin_gcMarkSingleThreaded = 1; +RUNTIME_WEAK int32_t Kotlin_gcMutatorsCooperate = 0; +RUNTIME_WEAK uint32_t Kotlin_auxGCThreads = 0; RUNTIME_WEAK int32_t Kotlin_workerExceptionHandling = 0; RUNTIME_WEAK int32_t Kotlin_suspendFunctionsFromAnyThreadFromObjC = 0; RUNTIME_WEAK Kotlin_getSourceInfo_FunctionType Kotlin_getSourceInfo_Function = nullptr; @@ -36,8 +37,12 @@ ALWAYS_INLINE compiler::DestroyRuntimeMode compiler::destroyRuntimeMode() noexce return static_cast(Kotlin_destroyRuntimeMode); } -ALWAYS_INLINE bool compiler::gcMarkSingleThreaded() noexcept { - return Kotlin_gcMarkSingleThreaded != 0; +ALWAYS_INLINE bool compiler::gcMutatorsCooperate() noexcept { + return Kotlin_gcMutatorsCooperate != 0; +} + +ALWAYS_INLINE uint32_t compiler::auxGCThreads() noexcept { + return Kotlin_auxGCThreads; } ALWAYS_INLINE compiler::WorkerExceptionHandling compiler::workerExceptionHandling() noexcept { diff --git a/kotlin-native/runtime/src/main/cpp/CompilerConstants.hpp b/kotlin-native/runtime/src/main/cpp/CompilerConstants.hpp index 27db9fdc522..d19006d5ea9 100644 --- a/kotlin-native/runtime/src/main/cpp/CompilerConstants.hpp +++ b/kotlin-native/runtime/src/main/cpp/CompilerConstants.hpp @@ -38,6 +38,7 @@ extern "C" const int32_t Kotlin_runtimeAssertsMode; extern "C" const int32_t Kotlin_disableMmap; extern "C" const char* const Kotlin_runtimeLogs; extern "C" const int32_t Kotlin_concurrentWeakSweep; +extern "C" const int32_t Kotlin_gcMarkSingleThreaded; extern "C" const int32_t Kotlin_freezingEnabled; extern "C" const int32_t Kotlin_freezingChecksEnabled; @@ -103,9 +104,15 @@ ALWAYS_INLINE inline bool concurrentWeakSweep() noexcept { return Kotlin_concurrentWeakSweep != 0; } +ALWAYS_INLINE inline bool gcMarkSingleThreaded() noexcept { + return Kotlin_gcMarkSingleThreaded != 0; +} + + WorkerExceptionHandling workerExceptionHandling() noexcept; DestroyRuntimeMode destroyRuntimeMode() noexcept; -bool gcMarkSingleThreaded() noexcept; +bool gcMutatorsCooperate() noexcept; +uint32_t auxGCThreads() noexcept; bool suspendFunctionsFromAnyThreadFromObjCEnabled() noexcept; AppStateTracking appStateTracking() noexcept; int getSourceInfo(void* addr, SourceInfo *result, int result_size) noexcept; diff --git a/kotlin-native/runtime/src/main/cpp/CooperativeIntrusiveList.hpp b/kotlin-native/runtime/src/main/cpp/CooperativeIntrusiveList.hpp deleted file mode 100644 index 95fe16a65ae..00000000000 --- a/kotlin-native/runtime/src/main/cpp/CooperativeIntrusiveList.hpp +++ /dev/null @@ -1,116 +0,0 @@ -/* - * Copyright 2010-2023 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license - * that can be found in the LICENSE file. - */ - -#pragma once - -#include -#include "IntrusiveList.hpp" -#include "Porting.h" -#include "Mutex.hpp" - -namespace kotlin { - -/** - * A list for thread-local use with a separate shared part that can be accessed by other threads. - */ -template> -class CooperativeIntrusiveList : Pinned { - using ListImpl = intrusive_forward_list; -public: - using value_type = typename ListImpl::value_type; - using size_type = typename ListImpl::size_type; - using reference = typename ListImpl::reference; - using pointer = typename ListImpl::pointer; - - CooperativeIntrusiveList() = default; - - bool localEmpty() const { - return local_.empty(); - } - - size_type localSize() const { - return localSize_; - } - - /** - * Tries to add `value` to the local list. - * See `intrusive_forward_list.try_push_front`. - */ - bool tryPushLocal(reference value) { - auto pushed = local_.try_push_front(value); - if (pushed) ++localSize_; - return pushed; - } - - /** - * Tries to pop a value from the local list. - * See `intrusive_forward_list.try_pop_front`. - */ - pointer tryPopLocal() { - auto popped = local_.try_pop_front(); - if (popped) { - --localSize_; - } else { - RuntimeAssert(localEmpty(), "Pop can only fail if the list is empty"); - } - return popped; - } - - void clearLocal() { - local_.clear(); - localSize_ = 0; - } - - bool sharedEmpty() const { - return shared_.empty(); - } - - /** - * Tries to move at most `maxAmount` elements from a from's shared list into `this`'s local list. - * In case some other thread is currently operating with the from's shared list, returns `0`. - * @return the number of elements stolen - */ - size_type tryTransferFrom(CooperativeIntrusiveList& from, size_type maxAmount) noexcept { - std::unique_lock guard(from.sharedLocked_, std::try_to_lock); - if (!guard || from.sharedEmpty()) { - return 0; - } - - auto amount = local_.splice_after(local_.before_begin(), - from.shared_.before_begin(), - from.shared_.end(), - maxAmount); - from.sharedSize_ -= amount; - localSize_ += amount; - - return amount; - } - - /** - * Moves all of the local items into own shared list. - * @return `0` if the shared list is busy, amount of newly shared items otherwise. - */ - size_type shareAll() noexcept { - RuntimeAssert(!localEmpty(), "Nothing to share"); - std::unique_lock guard(sharedLocked_, std::try_to_lock); - if (!guard) return 0; - - auto amount = shared_.splice_after(shared_.before_begin(), local_.before_begin(), local_.end(), localSize_); - sharedSize_ += amount; - localSize_ -= amount; - - return amount; - } - -private: - ListImpl local_; - size_type localSize_ = 0; - - ListImpl shared_; - size_type sharedSize_ = 0; - SpinLock sharedLocked_; -}; - -} \ No newline at end of file diff --git a/kotlin-native/runtime/src/main/cpp/CooperativeIntrusiveListTest.cpp b/kotlin-native/runtime/src/main/cpp/CooperativeIntrusiveListTest.cpp deleted file mode 100644 index 7643765d33d..00000000000 --- a/kotlin-native/runtime/src/main/cpp/CooperativeIntrusiveListTest.cpp +++ /dev/null @@ -1,236 +0,0 @@ -/* - * Copyright 2010-2023 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license - * that can be found in the LICENSE file. - */ - -#include "CooperativeIntrusiveList.hpp" - -#include "gmock/gmock.h" -#include "gtest/gtest.h" - -#include "ScopedThread.hpp" -#include "TestSupport.hpp" - -#include "std_support/Deque.hpp" -#include "std_support/Vector.hpp" -#include "std_support/List.hpp" - -using namespace kotlin; - -using ::testing::_; - -namespace { - -class Node : private Pinned { -public: - explicit Node(int value) : value_(value) {} - - int& operator*() { return value_; } - const int& operator*() const { return value_; } - - void clearNext() noexcept { next_ = nullptr; } - - int value() const { - return value_; - } - -private: - friend struct DefaultIntrusiveForwardListTraits; - - Node* next() const noexcept { return next_; } - void setNext(Node* next) noexcept { - RuntimeAssert(next, "next cannot be nullptr"); - next_ = next; - } - bool trySetNext(Node* next) noexcept { - RuntimeAssert(next, "next cannot be nullptr"); - if (next_) return false; - next_ = next; - return true; - } - - int value_; - Node* next_ = nullptr; -}; - -using TestSubject = CooperativeIntrusiveList; - -std_support::vector range(int first, int lastExclusive) { - std_support::vector values; - for (int i = first; i < lastExclusive; ++i) { - values.push_back(i); - } - return values; -} - -template -[[nodiscard]] std_support::list fill(TestSubject& list, Values&& values) { - std_support::list nodesHandle; - for (int value: values) { - auto& elem = nodesHandle.emplace_back(value); - list.tryPushLocal(elem); - } - return nodesHandle; -} - -void drainLocalInto(TestSubject& list, std_support::vector& dest) { - while (auto elem = list.tryPopLocal()) { - dest.push_back(elem->value()); - } -} - -} // namespace - -TEST(CooperativeIntrusiveListTest, Init) { - TestSubject list; - EXPECT_THAT(list.localEmpty(), true); - EXPECT_THAT(list.localSize(), 0); - EXPECT_THAT(list.sharedEmpty(), true); -} - -TEST(CooperativeIntrusiveListTest, TryPopLocalEmpty) { - TestSubject list; - auto res = list.tryPopLocal(); - EXPECT_THAT(res, nullptr); -} - -TEST(CooperativeIntrusiveListTest, TryPushLocalPopLocal) { - TestSubject list; - typename TestSubject::value_type value1(1); - typename TestSubject::value_type value2(2); - bool pushed1 = list.tryPushLocal(value1); - bool pushed2 = list.tryPushLocal(value2); - EXPECT_THAT(pushed1, true); - EXPECT_THAT(pushed2, true); - EXPECT_THAT(list.localEmpty(), false); - EXPECT_THAT(list.localSize(), 2); - EXPECT_THAT(list.sharedEmpty(), true); - std_support::vector popped; - drainLocalInto(list, popped); - EXPECT_THAT(list.localEmpty(), true); - EXPECT_THAT(list.localSize(), 0); - EXPECT_THAT(list.sharedEmpty(), true); - EXPECT_THAT(popped, testing::UnorderedElementsAre(1, 2)); -} - -TEST(CooperativeIntrusiveListTest, TryPushLocalTwice) { - TestSubject list; - typename TestSubject::value_type value(1); - bool pushed1 = list.tryPushLocal(value); - EXPECT_THAT(pushed1, true); - bool pushed2 = list.tryPushLocal(value); - EXPECT_THAT(pushed2, false); - EXPECT_THAT(list.localEmpty(), false); - EXPECT_THAT(list.localSize(), 1); - EXPECT_THAT(list.sharedEmpty(), true); -} - -TEST(CooperativeIntrusiveListTest, ShareSome) { - TestSubject list; - auto values = range(0, 10); - auto nodeHandle = fill(list, values); - EXPECT_THAT(list.localEmpty(), false); - EXPECT_THAT(list.localSize(), values.size()); - EXPECT_THAT(list.sharedEmpty(), true); - auto sharedAmount = list.shareAll(); - EXPECT_THAT(sharedAmount, values.size()); - EXPECT_THAT(list.localEmpty(), true); - EXPECT_THAT(list.sharedEmpty(), false); -} - -TEST(CooperativeIntrusiveListTest, TryTransferFromEmpty) { - TestSubject from; - TestSubject thief; - auto stolenAmount = thief.tryTransferFrom(from, 1); - EXPECT_THAT(stolenAmount, 0); -} - -TEST(CooperativeIntrusiveListTest, TryTransferHalf) { - TestSubject from; - auto values = range(0, 10); - auto nodeHandle = fill(from, values); - from.shareAll(); - - TestSubject thief; - auto toTransfer = values.size() / 2; - auto stolenAmount = thief.tryTransferFrom(from, toTransfer); - EXPECT_THAT(stolenAmount, toTransfer); - EXPECT_THAT(thief.localSize(), stolenAmount); - - from.tryTransferFrom(from, values.size()); - EXPECT_THAT(from.sharedEmpty(), true); - - std_support::vector allTheElements; - drainLocalInto(from, allTheElements); - drainLocalInto(thief, allTheElements); - EXPECT_THAT(allTheElements, testing::UnorderedElementsAreArray(values)); -} - -TEST(CooperativeIntrusiveListTest, TryTransferAllEventually) { - TestSubject from; - auto values = range(0, 10); - auto nodeHandle = fill(from, values); - from.shareAll(); - - TestSubject thief; - for (std::size_t i = 0; i < values.size(); ++i) { - auto stolenAmount = thief.tryTransferFrom(from, 1); - EXPECT_THAT(stolenAmount, 1); - } - EXPECT_THAT(from.sharedEmpty(), true); - EXPECT_THAT(thief.localSize(), values.size()); - - std_support::vector allTheElements; - drainLocalInto(from, allTheElements); - drainLocalInto(thief, allTheElements); - EXPECT_THAT(allTheElements, testing::UnorderedElementsAreArray(values)); -} - -TEST(CooperativeIntrusiveListTest, TransferingPingPong) { - TestSubject list1; - TestSubject list2; - const auto size = 100; - auto values = range(0, size); - auto nodesHandle1 = fill(list1, values); - auto nodesHandle2 = fill(list2, values); - - std::atomic ready = false; - auto kIters = 10000; - std_support::vector threads; - for (int tIdx = 0; tIdx < 2; ++tIdx) { - threads.emplace_back([&ready, kIters, tIdx, &list1, &list2] { - TestSubject& self = tIdx % 2 == 0 ? list1 : list2; - TestSubject& from = tIdx % 2 == 0 ? list2 : list1; - while (!ready.load()) { - std::this_thread::yield(); - } - for (int iter = 0; iter < kIters; ++iter) { - if (!self.localEmpty()) self.shareAll(); - self.tryTransferFrom(from, size / 2); - if (!self.localEmpty()) self.shareAll(); - self.tryTransferFrom(from, size); - if (auto popped = self.tryPopLocal()) { - popped->clearNext(); - self.tryPushLocal(*popped); - } - } - }); - } - ready = true; - - for (auto& thr: threads) { - thr.join(); - } - - // check nothing is lost - list1.tryTransferFrom(list1, size * 2); - list2.tryTransferFrom(list2, size * 2); - std_support::vector allTheElements; - drainLocalInto(list1, allTheElements); - drainLocalInto(list2, allTheElements); - - std_support::vector expected; - expected.insert(expected.end(), values.begin(), values.end()); - expected.insert(expected.end(), values.begin(), values.end()); - EXPECT_THAT(allTheElements, testing::UnorderedElementsAreArray(expected)); -} diff --git a/kotlin-native/runtime/src/main/cpp/ManuallyScoped.hpp b/kotlin-native/runtime/src/main/cpp/ManuallyScoped.hpp index 8e49bc05e6d..35b8b4d3a62 100644 --- a/kotlin-native/runtime/src/main/cpp/ManuallyScoped.hpp +++ b/kotlin-native/runtime/src/main/cpp/ManuallyScoped.hpp @@ -31,7 +31,7 @@ public: const T* operator->() const noexcept { return impl(); } private: - T* impl() noexcept { return reinterpret_cast(implStorage_); } + __attribute__((used)) T* impl() noexcept { return reinterpret_cast(implStorage_); } const T* impl() const noexcept { return reinterpret_cast(implStorage_); } alignas(T) char implStorage_[sizeof(T)]; diff --git a/kotlin-native/runtime/src/main/cpp/ObjectTraversal.hpp b/kotlin-native/runtime/src/main/cpp/ObjectTraversal.hpp index 9b99a67574f..5fb85c11b67 100644 --- a/kotlin-native/runtime/src/main/cpp/ObjectTraversal.hpp +++ b/kotlin-native/runtime/src/main/cpp/ObjectTraversal.hpp @@ -16,6 +16,23 @@ namespace kotlin { // TODO: Consider an iterator/ranges based approaches for traversals. +template +ALWAYS_INLINE void traverseClassObjectFields(ObjHeader* object, F process) noexcept(noexcept(process(std::declval()))) { + const TypeInfo* typeInfo = object->type_info(); + RuntimeAssert(typeInfo != theArrayTypeInfo, "Must not be an array of objects"); + for (int index = 0; index < typeInfo->objOffsetsCount_; index++) { + process(reinterpret_cast(reinterpret_cast(object) + typeInfo->objOffsets_[index])); + } +} + +template +ALWAYS_INLINE void traverseArrayOfObjectsElements(ArrayHeader* array, F process) noexcept(noexcept(process(std::declval()))) { + RuntimeAssert(array->type_info() == theArrayTypeInfo, "Must be an array of objects"); + for (uint32_t index = 0; index < array->count_; index++) { + process(ArrayAddressOfElementAt(array, index)); + } +} + template void traverseObjectFields(ObjHeader* object, F process) noexcept(noexcept(process(std::declval()))) { const TypeInfo* typeInfo = object->type_info(); diff --git a/kotlin-native/runtime/src/main/cpp/ParallelProcessor.hpp b/kotlin-native/runtime/src/main/cpp/ParallelProcessor.hpp new file mode 100644 index 00000000000..7d132ac2aa3 --- /dev/null +++ b/kotlin-native/runtime/src/main/cpp/ParallelProcessor.hpp @@ -0,0 +1,215 @@ +/* + * Copyright 2010-2023 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license + * that can be found in the LICENSE file. + */ + +#pragma once + +#include "CompilerConstants.hpp" +#include "KAssert.h" +#include "Logging.hpp" +#include "Utils.hpp" +#include "Porting.h" +#include "BoundedQueue.hpp" + +namespace kotlin { + +/** + * Coordinates a group of workers working in parallel on a large amounts of identical tasks. + * The dispatcher will try to balance the work among the workers. + * + * Requirements: + * - Every instantiated worker must execute `tryPop` sooner or later; + * - Every instantiated worker must finish execution before the destruction of the processor; + */ +template +class ParallelProcessor : private Pinned { + class Batch { + public: + ALWAYS_INLINE bool empty() const noexcept { + return elems_.empty(); + } + + ALWAYS_INLINE bool full() const noexcept { + return elemsCount_ == kBatchSize; + } + + ALWAYS_INLINE std::size_t elementsCount() const noexcept { + return elemsCount_; + } + + ALWAYS_INLINE bool tryPush(typename ListImpl::reference value) noexcept { + RuntimeAssert(!full(), "Batch overflow"); + bool pushed = elems_.try_push_front(value); + if (pushed) { + ++elemsCount_; + } + return pushed; + } + + ALWAYS_INLINE typename ListImpl::pointer tryPop() noexcept { + auto popped = elems_.try_pop_front(); + if (popped) { + --elemsCount_; + } + return popped; + } + + void transferAllInto(ListImpl& dst) noexcept { + dst.splice_after(dst.before_begin(), elems_.before_begin(), elems_.end(), std::numeric_limits::max()); + RuntimeAssert(empty(), "All the elements must be transferred"); + elemsCount_ = 0; + } + + void fillFrom(ListImpl& src) noexcept { + auto spliced = elems_.splice_after(elems_.before_begin(), src.before_begin(), src.end(), kBatchSize); + elemsCount_ = spliced; + } + + private: + ListImpl elems_; + std::size_t elemsCount_ = 0; + }; + +public: + class Worker : private Pinned { + friend ParallelProcessor; + public: + explicit Worker(ParallelProcessor& dispatcher) : dispatcher_(dispatcher) { + dispatcher_.registeredWorkers_.fetch_add(1, std::memory_order_relaxed); + RuntimeLogDebug({ "balancing" }, "Worker registered"); + } + + ALWAYS_INLINE bool localEmpty() const noexcept { + return batch_.empty() && overflowList_.empty(); + } + + ALWAYS_INLINE bool tryPushLocal(typename ListImpl::reference value) noexcept { + return overflowList_.try_push_front(value); + } + + ALWAYS_INLINE typename ListImpl::pointer tryPopLocal() noexcept { + return overflowList_.try_pop_front(); + } + + ALWAYS_INLINE bool tryPush(typename ListImpl::reference value) noexcept { + if (batch_.full()) { + bool released = dispatcher_.releaseBatch(std::move(batch_)); + if (!released) { + RuntimeLogDebug({ "balancing" }, "Batches pool overflow"); + batch_.transferAllInto(overflowList_); + } + batch_ = Batch{}; + } + return batch_.tryPush(value); + } + + ALWAYS_INLINE typename ListImpl::pointer tryPop() noexcept { + if (batch_.empty()) { + while (true) { + bool acquired = dispatcher_.acquireBatch(batch_); + if (!acquired) { + if (!overflowList_.empty()) { + batch_.fillFrom(overflowList_); + RuntimeLogDebug({ "balancing" }, "Acquired %zu elements from the overflow list", batch_.elementsCount()); + } else { + bool newWorkAvailable = waitForMoreWork(); + if (newWorkAvailable) continue; + return nullptr; + } + } + RuntimeAssert(!batch_.empty(), "Must have acquired some elements"); + break; + } + } + + return batch_.tryPop(); + } + + private: + bool waitForMoreWork() noexcept { + RuntimeAssert(batch_.empty(), "Local batch must be depleted before waiting for shared work"); + RuntimeAssert(overflowList_.empty(), "Local overflow list must be depleted before waiting for shared work"); + + std::unique_lock lock(dispatcher_.waitMutex_); + + auto nowWaiting = dispatcher_.waitingWorkers_.fetch_add(1, std::memory_order_relaxed) + 1; + RuntimeLogDebug({ "balancing" }, "Worker goes to sleep (now sleeping %zu of %zu)", + nowWaiting, dispatcher_.registeredWorkers_.load(std::memory_order_relaxed)); + + if (dispatcher_.allDone_) { + dispatcher_.waitingWorkers_.fetch_sub(1, std::memory_order_relaxed); + return false; + } + + if (nowWaiting == dispatcher_.registeredWorkers_.load(std::memory_order_relaxed)) { + // we are the last ones awake + RuntimeLogDebug({ "balancing" }, "Worker has detected termination"); + dispatcher_.allDone_ = true; + lock.unlock(); + dispatcher_.waitCV_.notify_all(); + dispatcher_.waitingWorkers_.fetch_sub(1, std::memory_order_relaxed); + return false; + } + + dispatcher_.waitCV_.wait(lock); + dispatcher_.waitingWorkers_.fetch_sub(1, std::memory_order_relaxed); + if (dispatcher_.allDone_) { + return false; + } + RuntimeLogDebug({ "balancing" }, "Worker woke up"); + + return true; + } + + ParallelProcessor& dispatcher_; + + Batch batch_; + ListImpl overflowList_; + }; + + ParallelProcessor() = default; + + ~ParallelProcessor() { + RuntimeAssert(waitingWorkers_.load() == 0, "All the workers must terminate before dispatcher destruction"); + } + + size_t registeredWorkers() { + return registeredWorkers_.load(std::memory_order_relaxed); + } + +private: + bool releaseBatch(Batch&& batch) { + RuntimeAssert(!batch.empty(), "A batch to release into shared pool must be non-empty"); + RuntimeLogDebug({ "balancing" }, "Releasing batch of %zu elements", batch.elementsCount()); + bool shared = sharedBatches_.enqueue(std::move(batch)); + if (shared) { + if (waitingWorkers_.load(std::memory_order_relaxed) > 0) { + waitCV_.notify_one(); + } + } + return shared; + } + + bool acquireBatch(Batch& dst) { + RuntimeAssert(dst.empty(), "Destination batch must be already depleted"); + auto acquired = sharedBatches_.dequeue(); + if (acquired) { + dst = std::move(*acquired); + RuntimeLogDebug({ "balancing" }, "Acquired a batch of %zu elements", dst.elementsCount()); + return true; + } + return false; + } + + BoundedQueue sharedBatches_; + + std::atomic registeredWorkers_ = 0; + std::atomic waitingWorkers_ = 0; + + std::atomic allDone_ = false; + mutable std::mutex waitMutex_; + mutable std::condition_variable waitCV_; +}; + +} diff --git a/kotlin-native/runtime/src/main/cpp/ParallelProcessorTest.cpp b/kotlin-native/runtime/src/main/cpp/ParallelProcessorTest.cpp new file mode 100644 index 00000000000..750eeea474b --- /dev/null +++ b/kotlin-native/runtime/src/main/cpp/ParallelProcessorTest.cpp @@ -0,0 +1,115 @@ +/* + * Copyright 2010-2023 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license + * that can be found in the LICENSE file. + */ + +#include "gmock/gmock.h" +#include "gtest/gtest.h" + +#include + +#include "IntrusiveList.hpp" + +#include "ParallelProcessor.hpp" + +#include "std_support/Vector.hpp" +#include "SingleThreadExecutor.hpp" +#include "TestSupport.hpp" + +using ::testing::_; +using namespace kotlin; + +namespace { + +struct Task { + template + static void workLoop(WorkList& workList) { + while (Task* task = workList.tryPop()) { + RuntimeAssert(!task->done_.load(), "Tasks are not idempotent"); + task->done_ = true; + } + } + + Task* next() const noexcept { return next_; } + + void setNext(Task* next) noexcept { + RuntimeAssert(next, "next cannot be nullptr"); + next_ = next; + } + + bool trySetNext(Task* next) noexcept { + RuntimeAssert(next, "next cannot be nullptr"); + if (next_ == nullptr) { + next_ = next; + return true; + } + return false; + } + + std::atomic done_ = false; + Task* next_ = nullptr; +}; + +auto workBatch(std::size_t size) { + std::list batch; + for (size_t i = 0; i < size; ++i) { + batch.emplace_back(); + } + return batch; +} + +template +void offerWork(WorkList& wl, Iterable& batch) { + for (auto& task: batch) { + bool accepted = wl.tryPush(task); + RuntimeAssert(accepted, "Must be accepted"); + } +} + +using ListImpl = intrusive_forward_list; +static constexpr auto kBatchSize = 256; +using Processor = ParallelProcessor; +using Worker = typename Processor::Worker; + +} // namespace + + +TEST(ParallelProcessorTest, ContededRegistration) { + Processor processor; + std::vector> workers(kDefaultThreadCount); + + std::atomic start = false; + std::list threads; + for (int i = 0; i < kDefaultThreadCount; ++i) { + threads.emplace_back([i, &start, &workers, &processor] { + while (!start.load()) {} + workers[i] = std::make_unique(processor); + }); + } + + start = true; + + for (auto& t : threads) { + t.join(); + } + + EXPECT_THAT(processor.registeredWorkers(), kDefaultThreadCount); + + workers.clear(); +} + +TEST(ParallelProcessorTest, Sharing) { + Processor processor; + Worker giver(processor); + Worker taker(processor); + + auto work = workBatch(kBatchSize * 2); + offerWork(giver, work); + + EXPECT_TRUE(taker.localEmpty()); + + // have to steal from giver + EXPECT_NE(taker.tryPop(), nullptr); + + EXPECT_FALSE(taker.localEmpty()); +} diff --git a/kotlin-native/runtime/src/main/cpp/Utils.hpp b/kotlin-native/runtime/src/main/cpp/Utils.hpp index 1f5a4b24ed5..ac47b6e08df 100644 --- a/kotlin-native/runtime/src/main/cpp/Utils.hpp +++ b/kotlin-native/runtime/src/main/cpp/Utils.hpp @@ -56,6 +56,22 @@ protected: ~Pinned() = default; }; +// A helper that executes the action provided upon destruction of the ScopeGuard instance. +template +class ScopeGuard final : private Pinned { +public: + template + ScopeGuard(InitAction initAction, FinalAction finalAction) noexcept : finalAction_(finalAction) { + initAction(); + } + ScopeGuard(FinalAction finalAction) noexcept : finalAction_(finalAction) {} + ~ScopeGuard() noexcept { + finalAction_(); + } +private: + FinalAction finalAction_; +}; + // A helper that scopley assigns a value to a variable. The variable will // be set to its original value upon destruction of the AutoReset instance. // Note that an AutoReset instance must have a shorter lifetime than diff --git a/kotlin-native/runtime/src/mm/cpp/ThreadSuspension.cpp b/kotlin-native/runtime/src/mm/cpp/ThreadSuspension.cpp index a13e6919970..f1a79d3d99a 100644 --- a/kotlin-native/runtime/src/mm/cpp/ThreadSuspension.cpp +++ b/kotlin-native/runtime/src/mm/cpp/ThreadSuspension.cpp @@ -25,6 +25,11 @@ namespace { } // namespace +bool kotlin::mm::isSuspendedOrNative(kotlin::mm::ThreadData& thread) noexcept { + auto& suspensionData = thread.suspensionData(); + return suspensionData.suspended() || suspensionData.state() == kotlin::ThreadState::kNative; +} + std::atomic kotlin::mm::internal::gSuspensionRequested = false; kotlin::ThreadState kotlin::mm::ThreadSuspensionData::setState(kotlin::ThreadState newState) noexcept { @@ -45,10 +50,10 @@ kotlin::ThreadState kotlin::mm::ThreadSuspensionData::setState(kotlin::ThreadSta NO_EXTERNAL_CALLS_CHECK void kotlin::mm::ThreadSuspensionData::suspendIfRequested() noexcept { if (IsThreadSuspensionRequested()) { + auto suspendStartMs = konan::getTimeMicros(); threadData_.gc().OnSuspendForGC(); std::unique_lock lock(gSuspensionMutex); auto threadId = konan::currentThreadId(); - auto suspendStartMs = konan::getTimeMicros(); RuntimeLogDebug({kTagGC, kTagMM}, "Suspending thread %d", threadId); AutoReset scopedAssignSuspended(&suspended_, true); gSuspensionCondVar.wait(lock, []() { return !IsThreadSuspensionRequested(); }); diff --git a/kotlin-native/runtime/src/mm/cpp/ThreadSuspension.hpp b/kotlin-native/runtime/src/mm/cpp/ThreadSuspension.hpp index bfd2175a869..a8fd1dad31f 100644 --- a/kotlin-native/runtime/src/mm/cpp/ThreadSuspension.hpp +++ b/kotlin-native/runtime/src/mm/cpp/ThreadSuspension.hpp @@ -52,6 +52,8 @@ private: bool RequestThreadsSuspension() noexcept; void WaitForThreadsSuspension() noexcept; +bool isSuspendedOrNative(kotlin::mm::ThreadData& thread) noexcept; + /** * Suspends all threads registered in ThreadRegistry except threads that are in the Native state. * Blocks until all such threads are suspended. Threads that are in the Native state on the moment diff --git a/kotlin-native/runtime/src/test_support/cpp/CompilerGenerated.cpp b/kotlin-native/runtime/src/test_support/cpp/CompilerGenerated.cpp index 760f8ea53d3..7c34503b1b1 100644 --- a/kotlin-native/runtime/src/test_support/cpp/CompilerGenerated.cpp +++ b/kotlin-native/runtime/src/test_support/cpp/CompilerGenerated.cpp @@ -75,6 +75,12 @@ extern const int32_t Kotlin_disableMmap = 0; #endif extern const char* const Kotlin_runtimeLogs = nullptr; extern const int32_t Kotlin_concurrentWeakSweep = 1; +#if KONAN_WINDOWS +// parallel mark tests hang on mingw due to (presumably) a bug in winpthread +extern const int32_t Kotlin_gcMarkSingleThreaded = 1; +#else +extern const int32_t Kotlin_gcMarkSingleThreaded = 0; +#endif extern const int32_t Kotlin_freezingChecksEnabled = 1; extern const int32_t Kotlin_freezingEnabled = 1;