azalea/kotlin-fork
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[K/N] Implement non-blocking approach for sweep

Browse Source
This commit is contained in:
Pavel Kunyavskiy
2021-11-23 19:05:06 +03:00
committed by Space
parent df4b367296
commit ebca4d793f
17 changed files with 650 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.cpp
+101 -93
View File
@@ -17,6 +17,8 @@
#include "ThreadData.hpp"
#include "ThreadRegistry.hpp"
#include "ThreadSuspension.hpp"
#include "GCState.hpp"
#include "FinalizerProcessor.hpp"
using namespace kotlin;
@@ -55,13 +57,8 @@ struct SweepTraits {
}
};
struct FinalizeTraits {
using ObjectFactory = mm::ObjectFactory<gc::ConcurrentMarkAndSweep>;
};
// Global, because it's accessed on a hot path: avoid memory load from `this`.
std::atomic<gc::ConcurrentMarkAndSweep::SafepointFlag> gSafepointFlag = gc::ConcurrentMarkAndSweep::SafepointFlag::kNone;
std::atomic<bool> gNeedSafepointSlowpath = false;
} // namespace
ALWAYS_INLINE void gc::ConcurrentMarkAndSweep::ThreadData::SafePointFunctionPrologue() noexcept {
@@ -74,130 +71,141 @@ ALWAYS_INLINE void gc::ConcurrentMarkAndSweep::ThreadData::SafePointLoopBody() n
void gc::ConcurrentMarkAndSweep::ThreadData::SafePointAllocation(size_t size) noexcept {
threadData_.gcScheduler().OnSafePointAllocation(size);
SafepointFlag flag = gSafepointFlag.load();
if (flag != SafepointFlag::kNone) {
SafePointSlowPath(flag);
if (gNeedSafepointSlowpath.load()) {
SafePointSlowPath();
}
}
void gc::ConcurrentMarkAndSweep::ThreadData::ScheduleAndWaitFullGC() noexcept {
ThreadStateGuard guard(ThreadState::kNative);
auto scheduled_epoch = gc_.state_.schedule();
gc_.state_.waitEpochFinished(scheduled_epoch);
}
void gc::ConcurrentMarkAndSweep::ThreadData::PerformFullGC() noexcept {
auto didGC = gc_.PerformFullGC();
void gc::ConcurrentMarkAndSweep::ThreadData::ScheduleAndWaitFullGCWithFinalizers() noexcept {
ThreadStateGuard guard(ThreadState::kNative);
auto scheduled_epoch = gc_.state_.schedule();
gc_.state_.waitEpochFinalized(scheduled_epoch);
}
if (!didGC) {
// If we failed to suspend threads, someone else might be asking to suspend them.
threadData_.suspensionData().suspendIfRequested();
}
void gc::ConcurrentMarkAndSweep::ThreadData::StopFinalizerThreadForTests() noexcept {
gc_.finalizerProcessor_->StopFinalizerThread();
}
void gc::ConcurrentMarkAndSweep::ThreadData::OnOOM(size_t size) noexcept {
RuntimeLogDebug({kTagGC}, "Attempt to GC on OOM at size=%zu", size);
PerformFullGC();
ScheduleAndWaitFullGC();
}
ALWAYS_INLINE void gc::ConcurrentMarkAndSweep::ThreadData::SafePointRegular(size_t weight) noexcept {
threadData_.gcScheduler().OnSafePointRegular(weight);
SafepointFlag flag = gSafepointFlag.load();
if (flag != SafepointFlag::kNone) {
SafePointSlowPath(flag);
if (gNeedSafepointSlowpath.load()) {
SafePointSlowPath();
}
}
NO_INLINE void gc::ConcurrentMarkAndSweep::ThreadData::SafePointSlowPath(SafepointFlag flag) noexcept {
RuntimeAssert(flag != SafepointFlag::kNone, "Must've been handled by the caller");
// No need to check for kNeedsSuspend, because `suspendIfRequested` checks for its own atomic.
NO_EXTERNAL_CALLS_CHECK NO_INLINE void gc::ConcurrentMarkAndSweep::ThreadData::SafePointSlowPath() noexcept {
threadData_.suspensionData().suspendIfRequested();
if (flag == SafepointFlag::kNeedsGC) {
RuntimeLogDebug({kTagGC}, "Attempt to GC at SafePoint");
PerformFullGC();
}
}
gc::ConcurrentMarkAndSweep::ConcurrentMarkAndSweep() noexcept {
mm::GlobalData::Instance().gcScheduler().SetScheduleGC([]() {
gc::ConcurrentMarkAndSweep::ConcurrentMarkAndSweep() noexcept :
finalizerProcessor_(make_unique<FinalizerProcessor>([this](int64_t epoch) { state_.finalized(epoch);})) {
mm::GlobalData::Instance().gcScheduler().SetScheduleGC([this]() NO_EXTERNAL_CALLS_CHECK NO_INLINE {
RuntimeLogDebug({kTagGC}, "Scheduling GC by thread %d", konan::currentThreadId());
gSafepointFlag = SafepointFlag::kNeedsGC;
state_.schedule();
});
gcThread_ = std::thread([this] {
while (true) {
auto epoch = state_.waitScheduled();
if (epoch.has_value()) {
PerformFullGC(*epoch);
} else {
break;
}
}
});
}
bool gc::ConcurrentMarkAndSweep::PerformFullGC() noexcept {
gc::ConcurrentMarkAndSweep::~ConcurrentMarkAndSweep() {
state_.shutdown();
gcThread_.join();
}
void gc::ConcurrentMarkAndSweep::RequestThreadsSuspension() noexcept {
gNeedSafepointSlowpath = true;
bool didSuspend = mm::RequestThreadsSuspension();
RuntimeAssert(didSuspend, "Only GC thread can request suspension");
}
void gc::ConcurrentMarkAndSweep::ResumeThreads() noexcept {
mm::ResumeThreads();
gNeedSafepointSlowpath = false;
}
bool gc::ConcurrentMarkAndSweep::PerformFullGC(int64_t epoch) noexcept {
RuntimeLogDebug({kTagGC}, "Attempt to suspend threads by thread %d", konan::currentThreadId());
auto timeStartUs = konan::getTimeMicros();
bool didSuspend = mm::RequestThreadsSuspension();
if (!didSuspend) {
RuntimeLogDebug({kTagGC}, "Failed to suspend threads by thread %d", konan::currentThreadId());
// Somebody else suspended the threads, and so ran a GC.
// TODO: This breaks if suspension is used by something apart from GC.
return false;
}
RequestThreadsSuspension();
RuntimeLogDebug({kTagGC}, "Requested thread suspension by thread %d", konan::currentThreadId());
gSafepointFlag = SafepointFlag::kNeedsSuspend;
mm::ObjectFactory<gc::ConcurrentMarkAndSweep>::FinalizerQueue finalizerQueue;
{
// Switch state to native to simulate this thread being a GC thread.
ThreadStateGuard guard(ThreadState::kNative);
RuntimeAssert(!kotlin::mm::IsCurrentThreadRegistered(), "Concurrent GC must run on unregistered thread");
mm::WaitForThreadsSuspension();
auto timeSuspendUs = konan::getTimeMicros();
RuntimeLogDebug({kTagGC}, "Suspended all threads in %" PRIu64 " microseconds", timeSuspendUs - timeStartUs);
mm::WaitForThreadsSuspension();
auto timeSuspendUs = konan::getTimeMicros();
RuntimeLogDebug({kTagGC}, "Suspended all threads in %" PRIu64 " microseconds", timeSuspendUs - timeStartUs);
auto& scheduler = mm::GlobalData::Instance().gcScheduler();
scheduler.gcData().OnPerformFullGC();
auto& scheduler = mm::GlobalData::Instance().gcScheduler();
scheduler.gcData().OnPerformFullGC();
RuntimeLogInfo(
{kTagGC}, "Started GC epoch %zu. Time since last GC %" PRIu64 " microseconds", epoch_, timeStartUs - lastGCTimestampUs_);
auto graySet = collectRootSet();
auto timeRootSetUs = konan::getTimeMicros();
// Can be unsafe, because we've stopped the world.
auto objectsCountBefore = mm::GlobalData::Instance().objectFactory().GetSizeUnsafe();
state_.start(epoch);
RuntimeLogInfo(
{kTagGC}, "Started GC epoch %" PRId64 ". Time since last GC %" PRIu64 " microseconds", epoch, timeStartUs - lastGCTimestampUs_);
auto graySet = collectRootSet();
auto timeRootSetUs = konan::getTimeMicros();
// Can be unsafe, because we've stopped the world.
RuntimeLogInfo(
{kTagGC}, "Collected root set of size %zu in %" PRIu64 " microseconds", graySet.size(),
timeRootSetUs - timeSuspendUs);
auto markStats = gc::Mark<MarkTraits>(std::move(graySet));
auto timeMarkUs = konan::getTimeMicros();
RuntimeLogDebug({kTagGC}, "Marked %zu objects in %" PRIu64 " microseconds. Processed %zu duplicate entries in the gray set", markStats.aliveHeapSet, timeMarkUs - timeRootSetUs, markStats.duplicateEntries);
scheduler.gcData().UpdateAliveSetBytes(markStats.aliveHeapSetBytes);
gc::SweepExtraObjects<SweepTraits>(mm::GlobalData::Instance().extraObjectDataFactory());
auto timeSweepExtraObjectsUs = konan::getTimeMicros();
RuntimeLogDebug({kTagGC}, "Sweeped extra objects in %" PRIu64 " microseconds", timeSweepExtraObjectsUs - timeMarkUs);
finalizerQueue = gc::Sweep<SweepTraits>(mm::GlobalData::Instance().objectFactory());
auto timeSweepUs = konan::getTimeMicros();
RuntimeLogDebug({kTagGC}, "Sweeped in %" PRIu64 " microseconds", timeSweepUs - timeSweepExtraObjectsUs);
auto objectsCountBefore = mm::GlobalData::Instance().objectFactory().GetSizeUnsafe();
RuntimeLogInfo(
{kTagGC}, "Collected root set of size %zu in %" PRIu64 " microseconds", graySet.size(),
timeRootSetUs - timeSuspendUs);
auto markStats = gc::Mark<MarkTraits>(std::move(graySet));
auto timeMarkUs = konan::getTimeMicros();
RuntimeLogDebug({kTagGC}, "Marked %zu objects in %" PRIu64 " microseconds. Processed %zu duplicate entries in the gray set", markStats.aliveHeapSet, timeMarkUs - timeRootSetUs, markStats.duplicateEntries);
scheduler.gcData().UpdateAliveSetBytes(markStats.aliveHeapSetBytes);
gc::SweepExtraObjects<SweepTraits>(mm::GlobalData::Instance().extraObjectDataFactory());
auto timeSweepExtraObjectsUs = konan::getTimeMicros();
RuntimeLogDebug({kTagGC}, "Sweeped extra objects in %" PRIu64 " microseconds", timeSweepExtraObjectsUs - timeMarkUs);
// Can be unsafe, because we've stopped the world.
auto objectsCountAfter = mm::GlobalData::Instance().objectFactory().GetSizeUnsafe();
auto extraObjectsCountAfter = mm::GlobalData::Instance().extraObjectDataFactory().GetSizeUnsafe();
auto objectFactoryIterable = mm::GlobalData::Instance().objectFactory().LockForIter();
gSafepointFlag = SafepointFlag::kNone;
mm::ResumeThreads();
auto timeResumeUs = konan::getTimeMicros();
ResumeThreads();
auto timeResumeUs = konan::getTimeMicros();
RuntimeLogDebug({kTagGC}, "Resumed threads in %" PRIu64 " microseconds.", timeResumeUs - timeSweepUs);
RuntimeLogDebug({kTagGC},
"Resumed threads in %" PRIu64 " microseconds. Total pause for most threads is %" PRIu64" microseconds",
timeResumeUs - timeSweepExtraObjectsUs, timeResumeUs - timeStartUs);
auto finalizersCount = finalizerQueue.size();
auto collectedCount = objectsCountBefore - objectsCountAfter - finalizersCount;
auto finalizerQueue = gc::Sweep<SweepTraits>(objectFactoryIterable);
auto timeSweepUs = konan::getTimeMicros();
RuntimeLogDebug({kTagGC}, "Swept in %" PRIu64 " microseconds", timeSweepUs - timeResumeUs);
RuntimeLogInfo(
{kTagGC},
"Finished GC epoch %zu. Collected %zu objects, to be finalized %zu objects, %zu objects and %zd extra data objects remain. Total pause time %" PRIu64
" microseconds",
epoch_, collectedCount, finalizersCount, objectsCountAfter, extraObjectsCountAfter, timeResumeUs - timeStartUs);
++epoch_;
lastGCTimestampUs_ = timeResumeUs;
}
// Can be unsafe, because we have a lock in objectFactoryIterable
auto objectsCountAfter = mm::GlobalData::Instance().objectFactory().GetSizeUnsafe();
auto extraObjectsCountAfter = mm::GlobalData::Instance().extraObjectDataFactory().GetSizeUnsafe();
// Finalizers are run after threads are resumed, because finalizers may request GC themselves, which would
// try to suspend threads again. Also, we run finalizers in the runnable state, because they may be executing
// kotlin code.
auto finalizersCount = finalizerQueue.size();
auto collectedCount = objectsCountBefore - objectsCountAfter - finalizersCount;
// TODO: These will actually need to be run on a separate thread.
AssertThreadState(ThreadState::kRunnable);
RuntimeLogDebug({kTagGC}, "Starting to run finalizers");
auto timeBeforeUs = konan::getTimeMicros();
finalizerQueue.Finalize();
auto timeAfterUs = konan::getTimeMicros();
RuntimeLogInfo({kTagGC}, "Finished running finalizers in %" PRIu64 " microseconds", timeAfterUs - timeBeforeUs);
state_.finish(epoch);
finalizerProcessor_->ScheduleTasks(std::move(finalizerQueue), epoch);
RuntimeLogInfo(
{kTagGC},
"Finished GC epoch %" PRId64 ". Collected %zu objects, to be finalized %zu objects, %zu objects and %zd extra data objects remain. Total pause time %" PRIu64
" microseconds",
epoch, collectedCount, finalizersCount, objectsCountAfter, extraObjectsCountAfter, timeSweepUs - timeStartUs);
lastGCTimestampUs_ = timeResumeUs;
return true;
}
kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweep.hpp
+16 -11
View File
@@ -12,6 +12,7 @@
#include "ObjectFactory.hpp"
#include "Types.h"
#include "Utils.hpp"
#include "GCState.hpp"
namespace kotlin {
@@ -21,14 +22,12 @@ class ThreadData;
namespace gc {
// Stop-the-world Mark-and-Sweep that runs on mutator threads. Can support targets that do not have threads.
class FinalizerProcessor;
// Stop-the-world mark + concurrent sweep. The GC runs in a separate thread, finalizers run in another thread of their own.
// TODO: Also make mark concurrent.
class ConcurrentMarkAndSweep : private Pinned {
public:
enum class SafepointFlag {
kNone,
kNeedsSuspend,
kNeedsGC,
};
class ObjectData {
public:
@@ -58,7 +57,9 @@ public:
void SafePointExceptionUnwind() noexcept;
void SafePointAllocation(size_t size) noexcept;
void PerformFullGC() noexcept;
void ScheduleAndWaitFullGC() noexcept;
void ScheduleAndWaitFullGCWithFinalizers() noexcept;
void StopFinalizerThreadForTests() noexcept;
void OnOOM(size_t size) noexcept;
@@ -66,7 +67,7 @@ public:
private:
void SafePointRegular(size_t weight) noexcept;
void SafePointSlowPath(SafepointFlag flag) noexcept;
void SafePointSlowPath() noexcept;
ConcurrentMarkAndSweep& gc_;
mm::ThreadData& threadData_;
@@ -75,14 +76,18 @@ public:
using Allocator = ThreadData::Allocator;
ConcurrentMarkAndSweep() noexcept;
~ConcurrentMarkAndSweep() = default;
~ConcurrentMarkAndSweep();
private:
// Returns `true` if GC has happened, and `false` if not (because someone else has suspended the threads).
bool PerformFullGC() noexcept;
bool PerformFullGC(int64_t epoch) noexcept;
void RequestThreadsSuspension() noexcept;
void ResumeThreads() noexcept;
size_t epoch_ = 0;
uint64_t lastGCTimestampUs_ = 0;
GCStateHolder state_;
std::thread gcThread_;
KStdUniquePtr<FinalizerProcessor> finalizerProcessor_;
};
} // namespace gc
kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweepTest.cpp
+38 -24
View File
@@ -24,7 +24,6 @@
using namespace kotlin;
// These tests can only work if `GC` is `ConcurrentMarkAndSweep`.
// TODO: Extracting GC into a separate module will help with this.
namespace {
@@ -251,7 +250,7 @@ TEST_F(ConcurrentMarkAndSweepTest, RootSet) {
ASSERT_THAT(GetColor(stack2.header()), Color::kWhite);
ASSERT_THAT(GetColor(stack3.header()), Color::kWhite);
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(
Alive(threadData),
@@ -297,7 +296,7 @@ TEST_F(ConcurrentMarkAndSweepTest, InterconnectedRootSet) {
ASSERT_THAT(GetColor(stack2.header()), Color::kWhite);
ASSERT_THAT(GetColor(stack3.header()), Color::kWhite);
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(
Alive(threadData),
@@ -321,7 +320,7 @@ TEST_F(ConcurrentMarkAndSweepTest, FreeObjects) {
ASSERT_THAT(GetColor(object1.header()), Color::kWhite);
ASSERT_THAT(GetColor(object2.header()), Color::kWhite);
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(Alive(threadData), testing::UnorderedElementsAre());
});
@@ -338,7 +337,8 @@ TEST_F(ConcurrentMarkAndSweepTest, FreeObjectsWithFinalizers) {
EXPECT_CALL(finalizerHook(), Call(object1.header()));
EXPECT_CALL(finalizerHook(), Call(object2.header()));
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGCWithFinalizers();
threadData.gc().StopFinalizerThreadForTests();
EXPECT_THAT(Alive(threadData), testing::UnorderedElementsAre());
});
@@ -357,7 +357,8 @@ TEST_F(ConcurrentMarkAndSweepTest, FreeObjectWithFreeWeak) {
ASSERT_THAT(GetColor(weak1.header()), Color::kWhite);
ASSERT_THAT(weak1->referred, object1.header());
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGCWithFinalizers();
threadData.gc().StopFinalizerThreadForTests();
EXPECT_THAT(Alive(threadData), testing::UnorderedElementsAre());
});
@@ -374,7 +375,7 @@ TEST_F(ConcurrentMarkAndSweepTest, FreeObjectWithHoldedWeak) {
ASSERT_THAT(GetColor(weak1.header()), Color::kWhite);
ASSERT_THAT(weak1->referred, object1.header());
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(Alive(threadData), testing::UnorderedElementsAre(weak1.header(), stack.header()));
EXPECT_THAT(GetColor(weak1.header()), Color::kWhite);
@@ -407,7 +408,7 @@ TEST_F(ConcurrentMarkAndSweepTest, ObjectReferencedFromRootSet) {
ASSERT_THAT(GetColor(object3.header()), Color::kWhite);
ASSERT_THAT(GetColor(object4.header()), Color::kWhite);
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(
Alive(threadData),
@@ -456,7 +457,7 @@ TEST_F(ConcurrentMarkAndSweepTest, ObjectsWithCycles) {
ASSERT_THAT(GetColor(object5.header()), Color::kWhite);
ASSERT_THAT(GetColor(object6.header()), Color::kWhite);
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(
Alive(threadData),
@@ -507,7 +508,8 @@ TEST_F(ConcurrentMarkAndSweepTest, ObjectsWithCyclesAndFinalizers) {
EXPECT_CALL(finalizerHook(), Call(object5.header()));
EXPECT_CALL(finalizerHook(), Call(object6.header()));
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGCWithFinalizers();
threadData.gc().StopFinalizerThreadForTests();
EXPECT_THAT(
Alive(threadData),
@@ -540,7 +542,7 @@ TEST_F(ConcurrentMarkAndSweepTest, ObjectsWithCyclesIntoRootSet) {
ASSERT_THAT(GetColor(object1.header()), Color::kWhite);
ASSERT_THAT(GetColor(object2.header()), Color::kWhite);
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(Alive(threadData), testing::UnorderedElementsAre(global.header(), stack.header(), object1.header(), object2.header()));
EXPECT_THAT(GetColor(global.header()), Color::kWhite);
@@ -584,8 +586,8 @@ TEST_F(ConcurrentMarkAndSweepTest, RunGCTwice) {
ASSERT_THAT(GetColor(object5.header()), Color::kWhite);
ASSERT_THAT(GetColor(object6.header()), Color::kWhite);
threadData.gc().PerformFullGC();
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(
Alive(threadData),
@@ -615,7 +617,7 @@ TEST_F(ConcurrentMarkAndSweepTest, PermanentObjects) {
ASSERT_THAT(Alive(threadData), testing::UnorderedElementsAre(global2.header()));
EXPECT_THAT(GetColor(global2.header()), Color::kWhite);
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(Alive(threadData), testing::UnorderedElementsAre(global2.header()));
EXPECT_THAT(GetColor(global2.header()), Color::kWhite);
@@ -635,7 +637,7 @@ TEST_F(ConcurrentMarkAndSweepTest, SameObjectInRootSet) {
EXPECT_THAT(GetColor(global.header()), Color::kWhite);
EXPECT_THAT(GetColor(object.header()), Color::kWhite);
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(Alive(threadData), testing::UnorderedElementsAre(global.header(), object.header()));
EXPECT_THAT(GetColor(global.header()), Color::kWhite);
@@ -771,7 +773,7 @@ TEST_F(ConcurrentMarkAndSweepTest, MultipleMutatorsCollect) {
KStdVector<std::future<void>> gcFutures(kDefaultThreadCount);
gcFutures[0] = mutators[0].Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().PerformFullGC(); });
gcFutures[0] = mutators[0].Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().ScheduleAndWaitFullGC(); });
// Spin until thread suspension is requested.
while (!mm::IsThreadSuspensionRequested()) {
@@ -827,15 +829,27 @@ TEST_F(ConcurrentMarkAndSweepTest, MultipleMutatorsAllCollect) {
KStdVector<std::future<void>> gcFutures(kDefaultThreadCount);
// TODO: Maybe check that only one GC is performed.
for (int i = 0; i < kDefaultThreadCount; ++i) {
gcFutures[i] = mutators[i].Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().PerformFullGC(); });
gcFutures[i] = mutators[i].Execute([](mm::ThreadData& threadData, Mutator& mutator) {
threadData.gc().ScheduleAndWaitFullGC();
// If GC starts before all thread executed line above, two gc will be run
// So we are temporary switch threads to native state and then return them back after all GC runs are done
SwitchThreadState(mm::GetMemoryState(), kotlin::ThreadState::kNative);
});
}
for (auto& future : gcFutures) {
future.wait();
}
for (int i = 0; i < kDefaultThreadCount; ++i) {
mutators[i]
.Execute([](mm::ThreadData& threadData, Mutator& mutator) {
SwitchThreadState(mm::GetMemoryState(), kotlin::ThreadState::kRunnable);
})
.wait();
}
KStdVector<ObjHeader*> expectedAlive;
for (auto& global : globals) {
expectedAlive.push_back(global);
@@ -891,7 +905,7 @@ TEST_F(ConcurrentMarkAndSweepTest, MultipleMutatorsAddToRootSetAfterCollectionRe
}
KStdVector<std::future<void>> gcFutures(kDefaultThreadCount);
gcFutures[0] = mutators[0].Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().PerformFullGC(); });
gcFutures[0] = mutators[0].Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().ScheduleAndWaitFullGC(); });
// Spin until thread suspension is requested.
while (!mm::IsThreadSuspensionRequested()) {
@@ -956,7 +970,7 @@ TEST_F(ConcurrentMarkAndSweepTest, CrossThreadReference) {
KStdVector<std::future<void>> gcFutures(kDefaultThreadCount);
gcFutures[0] = mutators[0].Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().PerformFullGC(); });
gcFutures[0] = mutators[0].Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().ScheduleAndWaitFullGC(); });
// Spin until thread suspension is requested.
while (!mm::IsThreadSuspensionRequested()) {
@@ -1018,7 +1032,7 @@ TEST_F(ConcurrentMarkAndSweepTest, MultipleMutatorsWeaks) {
KStdVector<std::future<void>> gcFutures(kDefaultThreadCount);
gcFutures[0] = mutators[0].Execute([weak](mm::ThreadData& threadData, Mutator& mutator) {
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT((*weak)->referred, nullptr);
});
@@ -1069,7 +1083,7 @@ TEST_F(ConcurrentMarkAndSweepTest, NewThreadsWhileRequestingCollection) {
KStdVector<std::future<void>> gcFutures(kDefaultThreadCount);
gcFutures[0] = mutators[0].Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().PerformFullGC(); });
gcFutures[0] = mutators[0].Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().ScheduleAndWaitFullGC(); });
// Spin until thread suspension is requested.
while (!mm::IsThreadSuspensionRequested()) {
@@ -1135,7 +1149,7 @@ TEST_F(ConcurrentMarkAndSweepTest, FreeObjectWithFreeWeakReversedOrder) {
object1 = &object1_local;
global1->field1 = object1_local.header();
while (weak.load() == nullptr);
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
ASSERT_THAT(Alive(threadData), testing::UnorderedElementsAre(object1_local.header(), weak.load()->header(), global1.header()));
ASSERT_THAT(GetColor(global1.header()), Color::kWhite);
@@ -1145,7 +1159,7 @@ TEST_F(ConcurrentMarkAndSweepTest, FreeObjectWithFreeWeakReversedOrder) {
global1->field1 = nullptr;
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(Alive(threadData), testing::UnorderedElementsAre(global1.header()));
done = true;
kotlin-native/runtime/src/gc/cms/cpp/FinalizerProcessor.cpp
+74
View File
@@ -0,0 +1,74 @@
/*
* Copyright 2010-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
* that can be found in the LICENSE file.
*/
#include "FinalizerProcessor.hpp"
#include "ObjectFactory.hpp"
#include "Runtime.h"
#include <thread>
void kotlin::gc::FinalizerProcessor::StartFinalizerThreadIfNone() noexcept {
if (finalizerThread_.joinable()) return;
finalizerThread_ = std::thread([this] {
Kotlin_initRuntimeIfNeeded();
int64_t finalizersEpoch = 0;
while (true) {
std::unique_lock lock(finalizerQueueMutex_);
finalizerQueueCondVar_.wait(lock, [this, &finalizersEpoch] {
return finalizerQueue_.size() > 0 || finalizerQueueEpoch_ != finalizersEpoch || shutdownFlag_;
});
if (finalizerQueue_.size() == 0 && finalizerQueueEpoch_ == finalizersEpoch) {
newTasksAllowed_ = false;
RuntimeAssert(shutdownFlag_, "Nothing to do, but no shutdownFlag_ is set on wakeup");
break;
}
auto queue = std::move(finalizerQueue_);
finalizersEpoch = finalizerQueueEpoch_;
lock.unlock();
if (queue.size() > 0) {
ThreadStateGuard guard(ThreadState::kRunnable);
queue.Finalize();
}
epochDoneCallback_(finalizersEpoch);
}
});
}
void kotlin::gc::FinalizerProcessor::StopFinalizerThread() noexcept {
{
std::unique_lock guard(finalizerQueueMutex_);
if (!finalizerThread_.joinable()) return;
shutdownFlag_ = true;
finalizerQueueCondVar_.notify_all();
}
finalizerThread_.join();
shutdownFlag_ = false;
RuntimeAssert(finalizerQueue_.size() == 0, "Finalizer queue should be empty when killing finalizer thread");
std::unique_lock guard(finalizerQueueMutex_);
newTasksAllowed_ = true;
finalizerQueueCondVar_.notify_all();
}
void kotlin::gc::FinalizerProcessor::ScheduleTasks(Queue&& tasks, int64_t epoch) noexcept {
std::unique_lock guard(finalizerQueueMutex_);
if (tasks.size() == 0 && !IsRunning()) {
epochDoneCallback_(epoch);
return;
}
finalizerQueueCondVar_.wait(guard, [this] { return newTasksAllowed_; });
StartFinalizerThreadIfNone();
finalizerQueue_.MergeWith(std::move(tasks));
finalizerQueueEpoch_ = epoch;
finalizerQueueCondVar_.notify_all();
}
bool kotlin::gc::FinalizerProcessor::IsRunning() noexcept {
return finalizerThread_.joinable();
}
kotlin::gc::FinalizerProcessor::~FinalizerProcessor() {
StopFinalizerThread();
}
kotlin-native/runtime/src/gc/cms/cpp/FinalizerProcessor.hpp
+34
View File
@@ -0,0 +1,34 @@
/*
* Copyright 2010-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
* that can be found in the LICENSE file.
*/
#pragma once
#include "ObjectFactory.hpp"
#include "ConcurrentMarkAndSweep.hpp"
#include "GCState.hpp"
namespace kotlin::gc {
class FinalizerProcessor : Pinned {
public:
using Queue = typename kotlin::mm::ObjectFactory<ConcurrentMarkAndSweep>::FinalizerQueue;
// epochDoneCallback could be called on any subset of them.
// If no new tasks are set, epochDoneCallback will be eventually called on last epoch
explicit FinalizerProcessor(std::function<void(int64_t)> epochDoneCallback): epochDoneCallback_(std::move(epochDoneCallback)) {}
void ScheduleTasks(Queue&& tasks, int64_t epoch) noexcept;
void StopFinalizerThread() noexcept;
bool IsRunning() noexcept;
~FinalizerProcessor();
private:
void StartFinalizerThreadIfNone() noexcept;
std::thread finalizerThread_;
Queue finalizerQueue_;
std::condition_variable finalizerQueueCondVar_;
std::mutex finalizerQueueMutex_;
std::function<void(int64_t)> epochDoneCallback_;
int64_t finalizerQueueEpoch_ = 0;
bool shutdownFlag_ = false;
bool newTasksAllowed_ = true;
};
}
kotlin-native/runtime/src/gc/cms/cpp/FinalizerProcessorTest.cpp
+146
View File
@@ -0,0 +1,146 @@
/*
* Copyright 2010-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
* that can be found in the LICENSE file.
*/
#include "ConcurrentMarkAndSweep.hpp"
#include <condition_variable>
#include <future>
#include <mutex>
#include <thread>
#include "gmock/gmock.h"
#include "gtest/gtest.h"
#include "ExtraObjectData.hpp"
#include "FinalizerHooksTestSupport.hpp"
#include "GlobalData.hpp"
#include "ObjectOps.hpp"
#include "ObjectTestSupport.hpp"
#include "TestSupport.hpp"
#include "ThreadData.hpp"
#include "FinalizerProcessor.hpp"
using namespace kotlin;
// These tests can only work if `GC` is `ConcurrentMarkAndSweep`.
namespace {
struct Payload {
ObjHeader* field1;
ObjHeader* field2;
ObjHeader* field3;
static constexpr std::array kFields = {
&Payload::field1,
&Payload::field2,
&Payload::field3,
};
};
test_support::TypeInfoHolder typeHolder{test_support::TypeInfoHolder::ObjectBuilder<Payload>()};
test_support::TypeInfoHolder typeHolderWithFinalizer{test_support::TypeInfoHolder::ObjectBuilder<Payload>().addFlag(TF_HAS_FINALIZER)};
test_support::Object<Payload>& AllocateObjectWithFinalizer(mm::ThreadData& threadData) {
ObjHolder holder;
mm::AllocateObject(&threadData, typeHolderWithFinalizer.typeInfo(), holder.slot());
return test_support::Object<Payload>::FromObjHeader(holder.obj());
}
class FinalizerProcessorTest : public testing::Test {
public:
~FinalizerProcessorTest() {
mm::GlobalsRegistry::Instance().ClearForTests();
mm::GlobalData::Instance().extraObjectDataFactory().ClearForTests();
mm::GlobalData::Instance().objectFactory().ClearForTests();
}
testing::MockFunction<void(ObjHeader*)>& finalizerHook() { return finalizerHooks_.finalizerHook(); }
private:
FinalizerHooksTestSupport finalizerHooks_;
};
int threadsCount() {
int result = 0;
for (auto &thread: mm::ThreadRegistry::Instance().LockForIter()) {
static_cast<void>(thread); // to avoid unused warning
result++;
}
return result;
};
} // namespace
TEST_F(FinalizerProcessorTest, NotRunningThreadWhenUnused) {
GCStateHolder state;
gc::FinalizerProcessor processor([](int64_t) {});
ASSERT_EQ(threadsCount(), 0);
ASSERT_FALSE(processor.IsRunning());
mm::ObjectFactory<kotlin::gc::ConcurrentMarkAndSweep>::FinalizerQueue queue;
processor.ScheduleTasks(std::move(queue), 1);
ASSERT_EQ(threadsCount(), 0);
ASSERT_FALSE(processor.IsRunning());
}
TEST_F(FinalizerProcessorTest, RemoveObject) {
RunInNewThread([this] {
ASSERT_EQ(threadsCount(), 1);
std::atomic<int64_t> done = 0;
gc::FinalizerProcessor processor([&](int64_t epoch) { done = epoch; });
mm::ObjectFactory<kotlin::gc::ConcurrentMarkAndSweep>::FinalizerQueue queue;
auto &object = AllocateObjectWithFinalizer(*mm::ThreadRegistry::Instance().CurrentThreadData());
mm::ThreadRegistry::Instance().CurrentThreadData()->Publish();
auto &factory = mm::GlobalData::Instance().objectFactory();
auto iter = factory.LockForIter();
auto iterator = iter.begin();
iter.MoveAndAdvance(queue, iterator);
ASSERT_EQ(queue.size(), 1u);
EXPECT_CALL(finalizerHook(), Call(object.header()));
processor.ScheduleTasks(std::move(queue), 1);
while (done != 1) {}
ASSERT_EQ(threadsCount(), 2);
ASSERT_TRUE(processor.IsRunning());
processor.StopFinalizerThread();
ASSERT_EQ(threadsCount(), 1);
});
}
TEST_F(FinalizerProcessorTest, ScheduleTasksWhileFinalizing) {
RunInNewThread([this] {
std::atomic<int64_t> done = 0;
gc::FinalizerProcessor processor([&done](int64_t epoch) { done = epoch; });
std::vector<mm::ObjectFactory<kotlin::gc::ConcurrentMarkAndSweep>::FinalizerQueue> queues;
int epochs = 100;
std::vector<ObjHeader*> headers;
for (int epoch = 0; epoch < epochs; epoch++) {
for (int i = 0; i < 10; i++) {
auto& object = AllocateObjectWithFinalizer(*mm::ThreadRegistry::Instance().CurrentThreadData());
headers.push_back(object.header());
}
auto& factory = mm::GlobalData::Instance().objectFactory();
mm::ThreadRegistry::Instance().CurrentThreadData()->Publish();
auto iter = factory.LockForIter();
mm::ObjectFactory<kotlin::gc::ConcurrentMarkAndSweep>::FinalizerQueue queue;
for (auto iterator = iter.begin(); iterator != iter.end();) {
iter.MoveAndAdvance(queue, iterator);
}
queues.push_back(std::move(queue));
}
for (auto header: headers) {
EXPECT_CALL(finalizerHook(), Call(header));
}
for (int epoch = 0; epoch < epochs; epoch++) {
processor.ScheduleTasks(std::move(queues[epoch]), epoch + 1);
}
while (done != epochs) {}
ASSERT_EQ(threadsCount(), 2);
ASSERT_TRUE(processor.IsRunning());
processor.StopFinalizerThread();
ASSERT_EQ(threadsCount(), 1);
});
}
kotlin-native/runtime/src/gc/cms/cpp/GCState.hpp
+73
View File
@@ -0,0 +1,73 @@
/*
* Copyright 2010-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
* that can be found in the LICENSE file.
*/
#pragma once
#include <condition_variable>
#include <mutex>
#include <atomic>
#include <optional>
class GCStateHolder {
public:
int64_t schedule() {
std::unique_lock lock(mutex_);
if (scheduledEpoch <= startedEpoch) {
scheduledEpoch = startedEpoch + 1;
cond_.notify_all();
}
return scheduledEpoch;
}
void shutdown() {
std::unique_lock lock(mutex_);
shutdownFlag_ = true;
cond_.notify_all();
}
void start(int64_t epoch) {
std::unique_lock lock(mutex_);
startedEpoch = epoch;
cond_.notify_all();
}
void finish(int64_t epoch) {
std::unique_lock lock(mutex_);
finishedEpoch = epoch;
cond_.notify_all();
}
void finalized(int64_t epoch) {
std::unique_lock lock(mutex_);
finalizedEpoch = epoch;
cond_.notify_all();
}
void waitEpochFinished(int64_t epoch) {
std::unique_lock lock(mutex_);
cond_.wait(lock, [this, epoch] { return finishedEpoch >= epoch || shutdownFlag_; });
}
void waitEpochFinalized(int64_t epoch) {
std::unique_lock lock(mutex_);
cond_.wait(lock, [this, epoch] { return finalizedEpoch >= epoch || shutdownFlag_; });
}
std::optional<int64_t> waitScheduled() {
std::unique_lock lock(mutex_);
cond_.wait(lock, [this] { return scheduledEpoch > finishedEpoch || shutdownFlag_; });
if (shutdownFlag_) return std::nullopt;
return scheduledEpoch;
}
private:
std::mutex mutex_;
std::condition_variable cond_;
int64_t startedEpoch = 0;
int64_t finishedEpoch = 0;
int64_t scheduledEpoch = 0;
int64_t finalizedEpoch = 0;
bool shutdownFlag_ = false;
};
kotlin-native/runtime/src/gc/common/cpp/MarkAndSweepUtils.hpp
+10 -5
View File
@@ -85,26 +85,31 @@ void SweepExtraObjects(typename Traits::ExtraObjectsFactory& objectFactory) noex
}
template <typename Traits>
typename Traits::ObjectFactory::FinalizerQueue Sweep(typename Traits::ObjectFactory& objectFactory) noexcept {
typename Traits::ObjectFactory::FinalizerQueue Sweep(typename Traits::ObjectFactory::Iterable& objectFactoryIter) noexcept {
typename Traits::ObjectFactory::FinalizerQueue finalizerQueue;
auto iter = objectFactory.LockForIter();
for (auto it = iter.begin(); it != iter.end();) {
for (auto it = objectFactoryIter.begin(); it != objectFactoryIter.end();) {
if (Traits::TryResetMark(*it)) {
++it;
continue;
}
auto* objHeader = it->IsArray() ? it->GetArrayHeader()->obj() : it->GetObjHeader();
if (HasFinalizers(objHeader)) {
iter.MoveAndAdvance(finalizerQueue, it);
objectFactoryIter.MoveAndAdvance(finalizerQueue, it);
} else {
iter.EraseAndAdvance(it);
objectFactoryIter.EraseAndAdvance(it);
}
}
return finalizerQueue;
}
template <typename Traits>
typename Traits::ObjectFactory::FinalizerQueue Sweep(typename Traits::ObjectFactory& objectFactory) noexcept {
auto iter = objectFactory.LockForIter();
return Sweep<Traits>(iter);
}
KStdVector<ObjHeader*> collectRootSet();
} // namespace gc
kotlin-native/runtime/src/gc/noop/cpp/NoOpGC.hpp
+3 -1
View File
@@ -40,7 +40,8 @@ public:
void SafePointLoopBody() noexcept {}
void SafePointAllocation(size_t size) noexcept {}
void PerformFullGC() noexcept {}
void ScheduleAndWaitFullGC() noexcept {}
void ScheduleAndWaitFullGCWithFinalizers() noexcept {}
void OnOOM(size_t size) noexcept {}
@@ -53,6 +54,7 @@ public:
~NoOpGC() = default;
GCScheduler& scheduler() noexcept { return scheduler_; }
void StopFinalizerThreadForTests() noexcept {}
private:
GCScheduler scheduler_;
kotlin-native/runtime/src/gc/stms/cpp/SameThreadMarkAndSweep.cpp
+3 -3
View File
@@ -80,7 +80,7 @@ void gc::SameThreadMarkAndSweep::ThreadData::SafePointAllocation(size_t size) no
}
}
void gc::SameThreadMarkAndSweep::ThreadData::PerformFullGC() noexcept {
void gc::SameThreadMarkAndSweep::ThreadData::ScheduleAndWaitFullGC() noexcept {
auto didGC = gc_.PerformFullGC();
if (!didGC) {
@@ -91,7 +91,7 @@ void gc::SameThreadMarkAndSweep::ThreadData::PerformFullGC() noexcept {
void gc::SameThreadMarkAndSweep::ThreadData::OnOOM(size_t size) noexcept {
RuntimeLogDebug({kTagGC}, "Attempt to GC on OOM at size=%zu", size);
PerformFullGC();
ScheduleAndWaitFullGC();
}
ALWAYS_INLINE void gc::SameThreadMarkAndSweep::ThreadData::SafePointRegular(size_t weight) noexcept {
@@ -108,7 +108,7 @@ NO_INLINE void gc::SameThreadMarkAndSweep::ThreadData::SafePointSlowPath(Safepoi
threadData_.suspensionData().suspendIfRequested();
if (flag == SafepointFlag::kNeedsGC) {
RuntimeLogDebug({kTagGC}, "Attempt to GC at SafePoint");
PerformFullGC();
ScheduleAndWaitFullGC();
}
}
kotlin-native/runtime/src/gc/stms/cpp/SameThreadMarkAndSweep.hpp
+3 -1
View File
@@ -59,7 +59,8 @@ public:
void SafePointExceptionUnwind() noexcept;
void SafePointAllocation(size_t size) noexcept;
void PerformFullGC() noexcept;
void ScheduleAndWaitFullGC() noexcept;
void ScheduleAndWaitFullGCWithFinalizers() noexcept { ScheduleAndWaitFullGC(); }
void OnOOM(size_t size) noexcept;
@@ -77,6 +78,7 @@ public:
SameThreadMarkAndSweep() noexcept;
~SameThreadMarkAndSweep() = default;
void StopFinalizerThreadForTests() noexcept {}
private:
// Returns `true` if GC has happened, and `false` if not (because someone else has suspended the threads).
kotlin-native/runtime/src/gc/stms/cpp/SameThreadMarkAndSweepTest.cpp
+22 -23
View File
@@ -24,7 +24,6 @@
using namespace kotlin;
// These tests can only work if `GC` is `SameThreadMarkAndSweep`.
// TODO: Extracting GC into a separate module will help with this.
namespace {
@@ -251,7 +250,7 @@ TEST_F(SameThreadMarkAndSweepTest, RootSet) {
ASSERT_THAT(GetColor(stack2.header()), Color::kWhite);
ASSERT_THAT(GetColor(stack3.header()), Color::kWhite);
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(
Alive(threadData),
@@ -297,7 +296,7 @@ TEST_F(SameThreadMarkAndSweepTest, InterconnectedRootSet) {
ASSERT_THAT(GetColor(stack2.header()), Color::kWhite);
ASSERT_THAT(GetColor(stack3.header()), Color::kWhite);
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(
Alive(threadData),
@@ -321,7 +320,7 @@ TEST_F(SameThreadMarkAndSweepTest, FreeObjects) {
ASSERT_THAT(GetColor(object1.header()), Color::kWhite);
ASSERT_THAT(GetColor(object2.header()), Color::kWhite);
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(Alive(threadData), testing::UnorderedElementsAre());
});
@@ -338,7 +337,7 @@ TEST_F(SameThreadMarkAndSweepTest, FreeObjectsWithFinalizers) {
EXPECT_CALL(finalizerHook(), Call(object1.header()));
EXPECT_CALL(finalizerHook(), Call(object2.header()));
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(Alive(threadData), testing::UnorderedElementsAre());
});
@@ -357,7 +356,7 @@ TEST_F(SameThreadMarkAndSweepTest, FreeObjectWithFreeWeak) {
ASSERT_THAT(GetColor(weak1.header()), Color::kWhite);
ASSERT_THAT(weak1->referred, object1.header());
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(Alive(threadData), testing::UnorderedElementsAre());
});
@@ -374,7 +373,7 @@ TEST_F(SameThreadMarkAndSweepTest, FreeObjectWithHoldedWeak) {
ASSERT_THAT(GetColor(weak1.header()), Color::kWhite);
ASSERT_THAT(weak1->referred, object1.header());
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(Alive(threadData), testing::UnorderedElementsAre(weak1.header(), stack.header()));
EXPECT_THAT(GetColor(weak1.header()), Color::kWhite);
@@ -407,7 +406,7 @@ TEST_F(SameThreadMarkAndSweepTest, ObjectReferencedFromRootSet) {
ASSERT_THAT(GetColor(object3.header()), Color::kWhite);
ASSERT_THAT(GetColor(object4.header()), Color::kWhite);
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(
Alive(threadData),
@@ -456,7 +455,7 @@ TEST_F(SameThreadMarkAndSweepTest, ObjectsWithCycles) {
ASSERT_THAT(GetColor(object5.header()), Color::kWhite);
ASSERT_THAT(GetColor(object6.header()), Color::kWhite);
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(
Alive(threadData),
@@ -507,7 +506,7 @@ TEST_F(SameThreadMarkAndSweepTest, ObjectsWithCyclesAndFinalizers) {
EXPECT_CALL(finalizerHook(), Call(object5.header()));
EXPECT_CALL(finalizerHook(), Call(object6.header()));
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(
Alive(threadData),
@@ -540,7 +539,7 @@ TEST_F(SameThreadMarkAndSweepTest, ObjectsWithCyclesIntoRootSet) {
ASSERT_THAT(GetColor(object1.header()), Color::kWhite);
ASSERT_THAT(GetColor(object2.header()), Color::kWhite);
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(Alive(threadData), testing::UnorderedElementsAre(global.header(), stack.header(), object1.header(), object2.header()));
EXPECT_THAT(GetColor(global.header()), Color::kWhite);
@@ -584,8 +583,8 @@ TEST_F(SameThreadMarkAndSweepTest, RunGCTwice) {
ASSERT_THAT(GetColor(object5.header()), Color::kWhite);
ASSERT_THAT(GetColor(object6.header()), Color::kWhite);
threadData.gc().PerformFullGC();
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(
Alive(threadData),
@@ -615,7 +614,7 @@ TEST_F(SameThreadMarkAndSweepTest, PermanentObjects) {
ASSERT_THAT(Alive(threadData), testing::UnorderedElementsAre(global2.header()));
EXPECT_THAT(GetColor(global2.header()), Color::kWhite);
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(Alive(threadData), testing::UnorderedElementsAre(global2.header()));
EXPECT_THAT(GetColor(global2.header()), Color::kWhite);
@@ -635,7 +634,7 @@ TEST_F(SameThreadMarkAndSweepTest, SameObjectInRootSet) {
EXPECT_THAT(GetColor(global.header()), Color::kWhite);
EXPECT_THAT(GetColor(object.header()), Color::kWhite);
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(Alive(threadData), testing::UnorderedElementsAre(global.header(), object.header()));
EXPECT_THAT(GetColor(global.header()), Color::kWhite);
@@ -771,7 +770,7 @@ TEST_F(SameThreadMarkAndSweepTest, MultipleMutatorsCollect) {
KStdVector<std::future<void>> gcFutures(kDefaultThreadCount);
gcFutures[0] = mutators[0].Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().PerformFullGC(); });
gcFutures[0] = mutators[0].Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().ScheduleAndWaitFullGC(); });
// Spin until thread suspension is requested.
while (!mm::IsThreadSuspensionRequested()) {
@@ -829,7 +828,7 @@ TEST_F(SameThreadMarkAndSweepTest, MultipleMutatorsAllCollect) {
// TODO: Maybe check that only one GC is performed.
for (int i = 0; i < kDefaultThreadCount; ++i) {
gcFutures[i] = mutators[i].Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().PerformFullGC(); });
gcFutures[i] = mutators[i].Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().ScheduleAndWaitFullGC(); });
}
for (auto& future : gcFutures) {
@@ -891,7 +890,7 @@ TEST_F(SameThreadMarkAndSweepTest, MultipleMutatorsAddToRootSetAfterCollectionRe
}
KStdVector<std::future<void>> gcFutures(kDefaultThreadCount);
gcFutures[0] = mutators[0].Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().PerformFullGC(); });
gcFutures[0] = mutators[0].Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().ScheduleAndWaitFullGC(); });
// Spin until thread suspension is requested.
while (!mm::IsThreadSuspensionRequested()) {
@@ -956,7 +955,7 @@ TEST_F(SameThreadMarkAndSweepTest, CrossThreadReference) {
KStdVector<std::future<void>> gcFutures(kDefaultThreadCount);
gcFutures[0] = mutators[0].Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().PerformFullGC(); });
gcFutures[0] = mutators[0].Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().ScheduleAndWaitFullGC(); });
// Spin until thread suspension is requested.
while (!mm::IsThreadSuspensionRequested()) {
@@ -1018,7 +1017,7 @@ TEST_F(SameThreadMarkAndSweepTest, MultipleMutatorsWeaks) {
KStdVector<std::future<void>> gcFutures(kDefaultThreadCount);
gcFutures[0] = mutators[0].Execute([weak](mm::ThreadData& threadData, Mutator& mutator) {
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT((*weak)->referred, nullptr);
});
@@ -1069,7 +1068,7 @@ TEST_F(SameThreadMarkAndSweepTest, NewThreadsWhileRequestingCollection) {
KStdVector<std::future<void>> gcFutures(kDefaultThreadCount);
gcFutures[0] = mutators[0].Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().PerformFullGC(); });
gcFutures[0] = mutators[0].Execute([](mm::ThreadData& threadData, Mutator& mutator) { threadData.gc().ScheduleAndWaitFullGC(); });
// Spin until thread suspension is requested.
while (!mm::IsThreadSuspensionRequested()) {
@@ -1135,7 +1134,7 @@ TEST_F(SameThreadMarkAndSweepTest, FreeObjectWithFreeWeakReversedOrder) {
object1 = &object1_local;
global1->field1 = object1_local.header();
while (weak.load() == nullptr);
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
ASSERT_THAT(Alive(threadData), testing::UnorderedElementsAre(object1_local.header(), weak.load()->header(), global1.header()));
ASSERT_THAT(GetColor(global1.header()), Color::kWhite);
@@ -1145,7 +1144,7 @@ TEST_F(SameThreadMarkAndSweepTest, FreeObjectWithFreeWeakReversedOrder) {
global1->field1 = nullptr;
threadData.gc().PerformFullGC();
threadData.gc().ScheduleAndWaitFullGC();
EXPECT_THAT(Alive(threadData), testing::UnorderedElementsAre(global1.header()));
done = true;
kotlin-native/runtime/src/mm/cpp/CallsChecker.cpp
+1
View File
@@ -257,6 +257,7 @@ extern "C" const char* Kotlin_callsCheckerGoodFunctionNames[] = {
"llvm.va_start",
"llvm.x86.avx2.*",
"llvm.x86.ssse3.*",
"llvm.x86.sse2.*",
"llvm.uadd.sat.*",
"llvm.aarch64.neon.*",
kotlin-native/runtime/src/mm/cpp/Memory.cpp
+3 -3
View File
@@ -111,7 +111,7 @@ extern "C" void DeinitMemory(MemoryState* state, bool destroyRuntime) {
auto* node = mm::FromMemoryState(state);
if (destroyRuntime) {
ThreadStateGuard guard(state, ThreadState::kRunnable);
node->Get()->gc().PerformFullGC();
node->Get()->gc().ScheduleAndWaitFullGC();
// TODO: Also make sure that finalizers are run.
}
mm::ThreadRegistry::Instance().Unregister(node);
@@ -293,7 +293,7 @@ extern "C" RUNTIME_NOTHROW void GC_CollectorCallback(void* worker) {
extern "C" void Kotlin_native_internal_GC_collect(ObjHeader*) {
auto* threadData = mm::ThreadRegistry::Instance().CurrentThreadData();
threadData->gc().PerformFullGC();
threadData->gc().ScheduleAndWaitFullGC();
}
extern "C" void Kotlin_native_internal_GC_collectCyclic(ObjHeader*) {
@@ -402,7 +402,7 @@ extern "C" void Kotlin_Any_share(ObjHeader* thiz) {
}
extern "C" RUNTIME_NOTHROW void PerformFullGC(MemoryState* memory) {
memory->GetThreadData()->gc().PerformFullGC();
memory->GetThreadData()->gc().ScheduleAndWaitFullGC();
}
extern "C" bool TryAddHeapRef(const ObjHeader* object) {
kotlin-native/runtime/src/mm/cpp/ObjectFactory.hpp
+34 -2
View File
@@ -266,8 +266,20 @@ public:
Consumer() noexcept = default;
Consumer(Consumer&&) noexcept = default;
Consumer& operator=(Consumer&&) noexcept = default;
Consumer(Consumer&& other) noexcept {
root_ = std::move(other.root_);
size_ = other.size_;
last_ = other.last_;
other.size_ = 0;
other.last_ = nullptr;
}
Consumer& operator=(Consumer&& other) noexcept {
Consumer temp = std::move(other);
std::swap(root_, temp.root_);
std::swap(size_, temp.size_);
std::swap(last_, temp.last_);
return *this;
}
~Consumer() {
// Make sure not to blow up the stack by nested `~Node` calls.
@@ -280,6 +292,22 @@ public:
Iterator begin() noexcept { return Iterator(root_.get()); }
Iterator end() noexcept { return Iterator(nullptr); }
void MergeWith(Consumer &&other) {
AssertCorrect();
if (other.root_) {
if (!root_) {
root_ = std::move(other.root_);
} else {
last_->next_ = std::move(other.root_);
}
last_ = other.last_;
size_ += other.size_;
other.last_ = nullptr;
other.size_ = 0;
}
AssertCorrect();
}
private:
friend class ObjectFactoryStorage;
@@ -624,6 +652,10 @@ public:
}
}
void MergeWith(FinalizerQueue &&other) {
consumer_.MergeWith(std::move(other.consumer_));
}
Iterable IterForTests() noexcept { return Iterable(*this); }
private:
kotlin-native/runtime/src/mm/cpp/ObjectFactoryTest.cpp
+85
View File
@@ -478,6 +478,91 @@ TEST(ObjectFactoryStorageTest, MoveAll) {
EXPECT_THAT(consumer.size(), 3);
}
TEST(ObjectFactoryStorageTest, MergeWith) {
ObjectFactoryStorageRegular storage;
Producer<ObjectFactoryStorageRegular> producer(storage, SimpleAllocator());
Consumer<ObjectFactoryStorageRegular> consumer1;
Consumer<ObjectFactoryStorageRegular> consumer2;
producer.Insert<int>(1);
producer.Insert<int>(2);
producer.Insert<int>(3);
producer.Insert<int>(4);
producer.Insert<int>(5);
producer.Publish();
{
auto iter = storage.LockForIter();
for (auto it = iter.begin(); it != iter.end();) {
if (it->Data<int>() % 2 == 0) {
iter.MoveAndAdvance(consumer1, it);
} else {
++it;
}
}
}
{
auto iter = storage.LockForIter();
for (auto it = iter.begin(); it != iter.end();) {
iter.MoveAndAdvance(consumer2, it);
}
}
auto actual = Collect<int>(storage);
EXPECT_THAT(actual, testing::IsEmpty());
EXPECT_THAT(storage.GetSizeUnsafe(), 0);
EXPECT_THAT(producer.size(), 0);
{
auto actualConsumer1 = Collect<int, alignof(void*)>(consumer1);
auto actualConsumer2 = Collect<int, alignof(void*)>(consumer2);
EXPECT_THAT(actualConsumer1, testing::ElementsAre(2, 4));
EXPECT_THAT(consumer1.size(), 2);
EXPECT_THAT(actualConsumer2, testing::ElementsAre(1, 3, 5));
EXPECT_THAT(consumer2.size(), 3);
}
consumer1.MergeWith(std::move(consumer2));
{
auto actualConsumer1 = Collect<int, alignof(void*)>(consumer1);
auto actualConsumer2 = Collect<int, alignof(void*)>(consumer2);
EXPECT_THAT(actualConsumer1, testing::ElementsAre(2, 4, 1, 3, 5));
EXPECT_THAT(consumer1.size(), 5);
EXPECT_THAT(actualConsumer2, testing::ElementsAre());
EXPECT_THAT(consumer2.size(), 0);
}
Consumer<ObjectFactoryStorageRegular> consumer3;
consumer1.MergeWith(std::move(consumer3));
{
auto actualConsumer1 = Collect<int, alignof(void*)>(consumer1);
auto actualConsumer2 = Collect<int, alignof(void*)>(consumer2);
auto actualConsumer3 = Collect<int, alignof(void*)>(consumer3);
EXPECT_THAT(actualConsumer1, testing::ElementsAre(2, 4, 1, 3, 5));
EXPECT_THAT(consumer1.size(), 5);
EXPECT_THAT(actualConsumer2, testing::ElementsAre());
EXPECT_THAT(consumer2.size(), 0);
EXPECT_THAT(actualConsumer3, testing::ElementsAre());
EXPECT_THAT(consumer3.size(), 0);
}
consumer3.MergeWith(std::move(consumer1));
{
auto actualConsumer1 = Collect<int, alignof(void*)>(consumer1);
auto actualConsumer2 = Collect<int, alignof(void*)>(consumer2);
auto actualConsumer3 = Collect<int, alignof(void*)>(consumer3);
EXPECT_THAT(actualConsumer1, testing::ElementsAre());
EXPECT_THAT(consumer1.size(), 0);
EXPECT_THAT(actualConsumer2, testing::ElementsAre());
EXPECT_THAT(consumer2.size(), 0);
EXPECT_THAT(actualConsumer3, testing::ElementsAre(2, 4, 1, 3, 5));
EXPECT_THAT(consumer3.size(), 5);
}
}
TEST(ObjectFactoryStorageTest, MoveTheOnlyElement) {
ObjectFactoryStorageRegular storage;
Producer<ObjectFactoryStorageRegular> producer(storage, SimpleAllocator());
kotlin-native/runtime/src/mm/cpp/ThreadSuspension.cpp
+4 -1
View File
@@ -54,10 +54,13 @@ NO_EXTERNAL_CALLS_CHECK void kotlin::mm::ThreadSuspensionData::suspendIfRequeste
std::unique_lock lock(gSuspensionMutex);
if (IsThreadSuspensionRequested()) {
auto threadId = konan::currentThreadId();
auto suspendStartMs = konan::getTimeMicros();
RuntimeLogDebug({kTagGC, kTagMM}, "Suspending thread %d", threadId);
AutoReset scopedAssign(&suspended_, true);
gSuspendsionCondVar.wait(lock, []() { return !IsThreadSuspensionRequested(); });
RuntimeLogDebug({kTagGC, kTagMM}, "Resuming thread %d", threadId);
auto suspendEndMs = konan::getTimeMicros();
RuntimeLogDebug({kTagGC, kTagMM}, "Resuming thread %d after %" PRIu64 " microseconds of suspension",
threadId, suspendEndMs - suspendStartMs);
}
}