From de5dcfcbd83268d0781984601fdc3587217af164 Mon Sep 17 00:00:00 2001 From: Ilya Matveev Date: Mon, 31 May 2021 09:48:46 +0000 Subject: [PATCH] [K/N][Runtime] Support suspending mutators in the new MM --- kotlin-native/runtime/build.gradle.kts | 6 + .../runtime/src/legacymm/cpp/Memory.cpp | 4 + kotlin-native/runtime/src/main/cpp/Memory.h | 6 + kotlin-native/runtime/src/main/cpp/Utils.hpp | 24 ++ .../runtime/src/mm/cpp/ThreadData.hpp | 13 +- .../runtime/src/mm/cpp/ThreadState.cpp | 18 +- .../runtime/src/mm/cpp/ThreadState.hpp | 22 +- .../runtime/src/mm/cpp/ThreadStateTest.cpp | 25 ++- .../runtime/src/mm/cpp/ThreadSuspension.cpp | 95 ++++++++ .../runtime/src/mm/cpp/ThreadSuspension.hpp | 60 +++++ .../src/mm/cpp/ThreadSuspensionTest.cpp | 205 ++++++++++++++++++ 11 files changed, 466 insertions(+), 12 deletions(-) create mode 100644 kotlin-native/runtime/src/mm/cpp/ThreadSuspension.cpp create mode 100644 kotlin-native/runtime/src/mm/cpp/ThreadSuspension.hpp create mode 100644 kotlin-native/runtime/src/mm/cpp/ThreadSuspensionTest.cpp diff --git a/kotlin-native/runtime/build.gradle.kts b/kotlin-native/runtime/build.gradle.kts index 591e469aac9..2334fa95913 100644 --- a/kotlin-native/runtime/build.gradle.kts +++ b/kotlin-native/runtime/build.gradle.kts @@ -272,6 +272,12 @@ val assemble by tasks.registering { }) } +val hostAssemble by tasks.registering { + dependsOn(tasks.withType(CompileToBitcode::class).matching { + it.outputGroup == "main" && it.target == hostName + }) +} + val clean by tasks.registering { doFirst { delete(buildDir) diff --git a/kotlin-native/runtime/src/legacymm/cpp/Memory.cpp b/kotlin-native/runtime/src/legacymm/cpp/Memory.cpp index faa01669bc4..c28e24edc5a 100644 --- a/kotlin-native/runtime/src/legacymm/cpp/Memory.cpp +++ b/kotlin-native/runtime/src/legacymm/cpp/Memory.cpp @@ -3733,6 +3733,10 @@ ALWAYS_INLINE void kotlin::AssertThreadState(MemoryState* thread, ThreadState ex // no-op, used by the new MM only. } +ALWAYS_INLINE void kotlin::AssertThreadState(MemoryState* thread, std::initializer_list expected) noexcept { + // no-op, used by the new MM only. +} + MemoryState* kotlin::mm::GetMemoryState() { return ::memoryState; } diff --git a/kotlin-native/runtime/src/main/cpp/Memory.h b/kotlin-native/runtime/src/main/cpp/Memory.h index b22c6f3f565..586c79ee37a 100644 --- a/kotlin-native/runtime/src/main/cpp/Memory.h +++ b/kotlin-native/runtime/src/main/cpp/Memory.h @@ -17,6 +17,7 @@ #ifndef RUNTIME_MEMORY_H #define RUNTIME_MEMORY_H +#include #include #include "KAssert.h" @@ -410,12 +411,17 @@ ALWAYS_INLINE ThreadState SwitchThreadState(MemoryState* thread, ThreadState new // Asserts that the given thread is in the given state. ALWAYS_INLINE void AssertThreadState(MemoryState* thread, ThreadState expected) noexcept; +ALWAYS_INLINE void AssertThreadState(MemoryState* thread, std::initializer_list expected) noexcept; // Asserts that the current thread is in the the given state. ALWAYS_INLINE inline void AssertThreadState(ThreadState expected) noexcept { AssertThreadState(mm::GetMemoryState(), expected); } +ALWAYS_INLINE inline void AssertThreadState(std::initializer_list expected) noexcept { + AssertThreadState(mm::GetMemoryState(), expected); +} + // Scopely sets the given thread state for the given thread. class ThreadStateGuard final : private Pinned { public: diff --git a/kotlin-native/runtime/src/main/cpp/Utils.hpp b/kotlin-native/runtime/src/main/cpp/Utils.hpp index 1f10f53082a..a55bd75c427 100644 --- a/kotlin-native/runtime/src/main/cpp/Utils.hpp +++ b/kotlin-native/runtime/src/main/cpp/Utils.hpp @@ -6,6 +6,8 @@ #ifndef RUNTIME_UTILS_H #define RUNTIME_UTILS_H +#include + namespace kotlin { // A helper for implementing classes with disabled copy constructor and copy assignment. @@ -52,6 +54,28 @@ protected: ~Pinned() = default; }; +// A helper that scopley assings 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 +// the variable it works with to avoid invalid memory access. +template +class AutoReset final : private Pinned { + static_assert(std::is_assignable::value); + +public: + AutoReset(T1* variable, T2 value) : variable_(variable), oldValue_(*variable) { + *variable_ = value; + } + + ~AutoReset() { + *variable_ = oldValue_; + } + +private: + T1* variable_; + T2 oldValue_; +}; + } // namespace kotlin #endif // RUNTIME_UTILS_H diff --git a/kotlin-native/runtime/src/mm/cpp/ThreadData.hpp b/kotlin-native/runtime/src/mm/cpp/ThreadData.hpp index c994561b879..b6407c466c4 100644 --- a/kotlin-native/runtime/src/mm/cpp/ThreadData.hpp +++ b/kotlin-native/runtime/src/mm/cpp/ThreadData.hpp @@ -18,6 +18,7 @@ #include "ThreadLocalStorage.hpp" #include "Types.h" #include "Utils.hpp" +#include "ThreadSuspension.hpp" struct ObjHeader; @@ -32,9 +33,9 @@ public: threadId_(threadId), globalsThreadQueue_(GlobalsRegistry::Instance()), stableRefThreadQueue_(StableRefRegistry::Instance()), - state_(ThreadState::kRunnable), gc_(GlobalData::Instance().gc()), - objectFactoryThreadQueue_(GlobalData::Instance().objectFactory(), gc_) {} + objectFactoryThreadQueue_(GlobalData::Instance().objectFactory(), gc_), + suspensionData_(ThreadState::kRunnable) {} ~ThreadData() = default; @@ -46,9 +47,9 @@ public: StableRefRegistry::ThreadQueue& stableRefThreadQueue() noexcept { return stableRefThreadQueue_; } - ThreadState state() noexcept { return state_; } + ThreadState state() noexcept { return suspensionData_.state(); } - ThreadState setState(ThreadState state) noexcept { return state_.exchange(state); } + ThreadState setState(ThreadState state) noexcept { return suspensionData_.setState(state); } ObjectFactory::ThreadQueue& objectFactoryThreadQueue() noexcept { return objectFactoryThreadQueue_; } @@ -58,6 +59,8 @@ public: gc::GC::ThreadData& gc() noexcept { return gc_; } + ThreadSuspensionData& suspensionData() { return suspensionData_; } + void Publish() noexcept { // TODO: These use separate locks, which is inefficient. globalsThreadQueue_.Publish(); @@ -76,11 +79,11 @@ private: GlobalsRegistry::ThreadQueue globalsThreadQueue_; ThreadLocalStorage tls_; StableRefRegistry::ThreadQueue stableRefThreadQueue_; - std::atomic state_; ShadowStack shadowStack_; gc::GC::ThreadData gc_; ObjectFactory::ThreadQueue objectFactoryThreadQueue_; KStdVector> initializingSingletons_; + ThreadSuspensionData suspensionData_; }; } // namespace mm diff --git a/kotlin-native/runtime/src/mm/cpp/ThreadState.cpp b/kotlin-native/runtime/src/mm/cpp/ThreadState.cpp index 8486ca4719e..b4c00643b2e 100644 --- a/kotlin-native/runtime/src/mm/cpp/ThreadState.cpp +++ b/kotlin-native/runtime/src/mm/cpp/ThreadState.cpp @@ -7,7 +7,7 @@ #include "ThreadData.hpp" #include "ThreadState.hpp" -const char* kotlin::internal::stateToString(ThreadState state) noexcept { +const char* kotlin::ThreadStateName(ThreadState state) noexcept { switch (state) { case ThreadState::kRunnable: return "RUNNABLE"; @@ -16,6 +16,18 @@ const char* kotlin::internal::stateToString(ThreadState state) noexcept { } } +std::string kotlin::internal::statesToString(std::initializer_list states) noexcept { + std::string result = "{ "; + for (size_t i = 0; i < states.size(); i++) { + if (i != 0) { + result += ", "; + } + result += ThreadStateName(data(states)[i]); + } + result += " }"; + return result; +} + ALWAYS_INLINE ThreadState kotlin::SwitchThreadState(MemoryState* thread, ThreadState newState, bool reentrant) noexcept { return SwitchThreadState(thread->GetThreadData(), newState, reentrant); } @@ -24,6 +36,10 @@ ALWAYS_INLINE void kotlin::AssertThreadState(MemoryState* thread, ThreadState ex AssertThreadState(thread->GetThreadData(), expected); } +ALWAYS_INLINE void kotlin::AssertThreadState(MemoryState* thread, std::initializer_list expected) noexcept { + AssertThreadState(thread->GetThreadData(), expected); +} + ThreadState kotlin::GetThreadState(MemoryState* thread) noexcept { return thread->GetThreadData()->state(); } \ No newline at end of file diff --git a/kotlin-native/runtime/src/mm/cpp/ThreadState.hpp b/kotlin-native/runtime/src/mm/cpp/ThreadState.hpp index 6b3c8ee49d9..f5196604174 100644 --- a/kotlin-native/runtime/src/mm/cpp/ThreadState.hpp +++ b/kotlin-native/runtime/src/mm/cpp/ThreadState.hpp @@ -6,10 +6,13 @@ #ifndef RUNTIME_MM_THREAD_STATE_H #define RUNTIME_MM_THREAD_STATE_H +#include + #include #include #include "ThreadData.hpp" +#include "ThreadSuspension.hpp" namespace kotlin { @@ -19,17 +22,23 @@ ALWAYS_INLINE inline bool isStateSwitchAllowed(ThreadState oldState, ThreadState return oldState != newState || reentrant; } -const char* stateToString(ThreadState state) noexcept; +std::string statesToString(std::initializer_list states) noexcept; } // namespace internal +const char* ThreadStateName(ThreadState state) noexcept; + +inline std::ostream& operator<<(std::ostream& stream, ThreadState state) { + return stream << ThreadStateName(state); +} + // Switches the state of the given thread to `newState` and returns the previous thread state. ALWAYS_INLINE inline ThreadState SwitchThreadState(mm::ThreadData* threadData, ThreadState newState, bool reentrant = false) noexcept { auto oldState = threadData->setState(newState); // TODO(perf): Mesaure the impact of this assert in debug and opt modes. RuntimeAssert(internal::isStateSwitchAllowed(oldState, newState, reentrant), "Illegal thread state switch. Old state: %s. New state: %s.", - internal::stateToString(oldState), internal::stateToString(newState)); + ThreadStateName(oldState), ThreadStateName(newState)); return oldState; } @@ -38,7 +47,14 @@ ALWAYS_INLINE inline void AssertThreadState(mm::ThreadData* threadData, ThreadSt auto actual = threadData->state(); RuntimeAssert(actual == expected, "Unexpected thread state. Expected: %s. Actual: %s.", - internal::stateToString(expected), internal::stateToString(actual)); + ThreadStateName(expected), ThreadStateName(actual)); +} + +ALWAYS_INLINE inline void AssertThreadState(mm::ThreadData* threadData, std::initializer_list expected) noexcept { + auto actual = threadData->state(); + RuntimeAssert(std::any_of(expected.begin(), expected.end(), [actual](ThreadState expected) { return expected == actual; }), + "Unexpected thread state. Expected one of: %s. Actual: %s", + internal::statesToString(expected).c_str(), ThreadStateName(actual)); } } // namespace kotlin diff --git a/kotlin-native/runtime/src/mm/cpp/ThreadStateTest.cpp b/kotlin-native/runtime/src/mm/cpp/ThreadStateTest.cpp index c556820a123..98612df95e2 100644 --- a/kotlin-native/runtime/src/mm/cpp/ThreadStateTest.cpp +++ b/kotlin-native/runtime/src/mm/cpp/ThreadStateTest.cpp @@ -41,6 +41,9 @@ private: //static testing::MockFunction* ThreadStateTest::globalSomeFunctionMock = nullptr; +#define EXPECT_NO_DEATH(statement) \ + do { EXPECT_EXIT({statement; exit(0);}, testing::ExitedWithCode(0), testing::_); } while(false) + } // namespace TEST_F(ThreadStateTest, StateSwitchWithThreadData) { @@ -148,13 +151,11 @@ TEST(ThreadStateDeathTest, StateAsserts) { }); } -TEST(ThreadStateDeathTest, IncorrectStateSwitch) { +TEST(ThreadStateDeathTest, IncorrectStateSwitchWithDifferentFunctions) { RunInNewThread([](MemoryState* memoryState) { auto* threadData = memoryState->GetThreadData(); EXPECT_DEATH(SwitchThreadState(memoryState, ThreadState::kRunnable), "runtime assert: Illegal thread state switch. Old state: RUNNABLE. New state: RUNNABLE"); - EXPECT_DEATH(SwitchThreadState(threadData, ThreadState::kRunnable), - "runtime assert: Illegal thread state switch. Old state: RUNNABLE. New state: RUNNABLE"); EXPECT_DEATH(Kotlin_mm_switchThreadStateRunnable(), "runtime assert: Illegal thread state switch. Old state: RUNNABLE. New state: RUNNABLE"); @@ -165,6 +166,24 @@ TEST(ThreadStateDeathTest, IncorrectStateSwitch) { }); } +TEST(ThreadStateDeathTest, StateSwitchCorrectness) { + mm::ThreadData threadData(pthread_self()); + + // Allowed state switches: runnable <-> native + threadData.setState(ThreadState::kRunnable); + ASSERT_EQ(threadData.state(), ThreadState::kRunnable); + EXPECT_DEATH(SwitchThreadState(&threadData, ThreadState::kRunnable), + "runtime assert: Illegal thread state switch. Old state: RUNNABLE. New state: RUNNABLE"); + // Each EXPECT_NO_DEATH is executed in a fork process, so the global state of the test is not affected. + EXPECT_NO_DEATH(SwitchThreadState(&threadData, ThreadState::kNative)); + + threadData.setState(ThreadState::kNative); + ASSERT_EQ(threadData.state(), ThreadState::kNative); + EXPECT_NO_DEATH(SwitchThreadState(&threadData, ThreadState::kRunnable)); + EXPECT_DEATH(SwitchThreadState(&threadData, ThreadState::kNative), + "runtime assert: Illegal thread state switch. Old state: NATIVE. New state: NATIVE"); +} + TEST(ThreadStateDeathTest, ReentrantStateSwitch) { RunInNewThread([](MemoryState* memoryState) { auto* threadData = memoryState->GetThreadData(); diff --git a/kotlin-native/runtime/src/mm/cpp/ThreadSuspension.cpp b/kotlin-native/runtime/src/mm/cpp/ThreadSuspension.cpp new file mode 100644 index 00000000000..424c61c35d2 --- /dev/null +++ b/kotlin-native/runtime/src/mm/cpp/ThreadSuspension.cpp @@ -0,0 +1,95 @@ +/* + * 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 "ThreadData.hpp" +#include "ThreadSuspension.hpp" + +#include +#include +#include + +namespace { + +bool isSuspendedOrNative(kotlin::mm::ThreadData& thread) noexcept { + auto& suspensionData = thread.suspensionData(); + return suspensionData.suspended() || suspensionData.state() == kotlin::ThreadState::kNative; +} + +template +bool allThreads(F predicate) noexcept { + auto& threadRegistry = kotlin::mm::ThreadRegistry::Instance(); + auto* currentThread = threadRegistry.CurrentThreadData(); + kotlin::mm::ThreadRegistry::Iterable threads = kotlin::mm::ThreadRegistry::Instance().Iter(); + for (auto& thread : threads) { + // Handle if suspension was initiated by the mutator thread. + if (&thread == currentThread) + continue; + if (!predicate(thread)) { + return false; + } + } + return true; +} + +void yield() noexcept { + std::this_thread::yield(); +} + +std::atomic gSuspensionRequested = false; +THREAD_LOCAL_VARIABLE bool gSuspensionRequestedByCurrentThread = false; +std::mutex gSuspensionMutex; +std::condition_variable gSuspendsionCondVar; + +} // namespace + +bool kotlin::mm::ThreadSuspensionData::suspendIfRequested() noexcept { + if (IsThreadSuspensionRequested()) { + std::unique_lock lock(gSuspensionMutex); + if (IsThreadSuspensionRequested()) { + AutoReset scopedAssign(&suspended_, true); + gSuspendsionCondVar.wait(lock, []() { return !IsThreadSuspensionRequested(); }); + return true; + } + } + return false; +} + +bool kotlin::mm::IsThreadSuspensionRequested() noexcept { + // TODO: Consider using a more relaxed memory order. + return gSuspensionRequested.load(); +} + +bool kotlin::mm::SuspendThreads() noexcept { + RuntimeAssert(gSuspensionRequestedByCurrentThread == false, "Current thread already suspended threads."); + { + std::unique_lock lock(gSuspensionMutex); + bool actual = false; + gSuspensionRequested.compare_exchange_strong(actual, true); + if (actual) { + return false; + } + } + gSuspensionRequestedByCurrentThread = true; + + // Spin wating for threads to suspend. Ignore Native threads. + while(!allThreads(isSuspendedOrNative)) { + yield(); + } + + return true; +} + +void kotlin::mm::ResumeThreads() noexcept { + // From the std::condition_variable docs: + // Even if the shared variable is atomic, it must be modified under + // the mutex in order to correctly publish the modification to the waiting thread. + // https://en.cppreference.com/w/cpp/thread/condition_variable + { + std::unique_lock lock(gSuspensionMutex); + gSuspensionRequested = false; + } + gSuspensionRequestedByCurrentThread = false; + gSuspendsionCondVar.notify_all(); +} diff --git a/kotlin-native/runtime/src/mm/cpp/ThreadSuspension.hpp b/kotlin-native/runtime/src/mm/cpp/ThreadSuspension.hpp new file mode 100644 index 00000000000..e57584f33c2 --- /dev/null +++ b/kotlin-native/runtime/src/mm/cpp/ThreadSuspension.hpp @@ -0,0 +1,60 @@ +/* + * 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. + */ + +#ifndef RUNTIME_MM_THREAD_SUSPENSION_UTILS_H +#define RUNTIME_MM_THREAD_SUSPENSION_UTILS_H + +#include + +#include "Memory.h" + +namespace kotlin { +namespace mm { + +class ThreadSuspensionData : private Pinned { +public: + explicit ThreadSuspensionData(ThreadState initialState) noexcept : state_(initialState), suspended_(false) {} + + ~ThreadSuspensionData() = default; + + ThreadState state() noexcept { return state_; } + + ThreadState setState(ThreadState newState) noexcept { + ThreadState oldState = state_.exchange(newState); + if (oldState == ThreadState::kNative && newState == ThreadState::kRunnable) { + suspendIfRequested(); + } + return oldState; + } + + bool suspended() noexcept { return suspended_; } + + bool suspendIfRequested() noexcept; + +private: + std::atomic state_; + std::atomic suspended_; +}; + +bool IsThreadSuspensionRequested() 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 + * of this call will be suspended on exit from the Native state. + * Returns false if some other thread has suspended the threads. + */ +bool SuspendThreads() noexcept; + +/** + * Resumes all threads registered in ThreadRegistry that were suspended by the SuspendThreads call. + * Does not wait until all such threads are actually resumed. + */ +void ResumeThreads() noexcept; + +} // namespace mm +} // namespace kotlin + +#endif // RUNTIME_MM_THREAD_SUSPENSION_UTILS_H diff --git a/kotlin-native/runtime/src/mm/cpp/ThreadSuspensionTest.cpp b/kotlin-native/runtime/src/mm/cpp/ThreadSuspensionTest.cpp new file mode 100644 index 00000000000..4d53f66c52f --- /dev/null +++ b/kotlin-native/runtime/src/mm/cpp/ThreadSuspensionTest.cpp @@ -0,0 +1,205 @@ +/* + * 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 "MemoryPrivate.hpp" +#include "ThreadSuspension.hpp" +#include "ThreadState.hpp" + +#include +#include + +#include +#include + +#include + +using namespace kotlin; + +namespace { + +#ifdef KONAN_WINDOWS +constexpr size_t kDefaultIterations = 1000; +constexpr size_t kDefaultReportingStep = 100; +#else +constexpr size_t kDefaultIterations = 10000; +constexpr size_t kDefaultReportingStep = 1000; +#endif // #ifdef KONAN_WINDOWS + +KStdVector collectThreadData() { + KStdVector result; + auto iter = mm::ThreadRegistry::Instance().Iter(); + for (auto& thread : iter) { + result.push_back(&thread); + } + return result; +} + +template +KStdVector collectFromThreadData(F extractFunction) { + KStdVector result; + auto threadData = collectThreadData(); + std::transform(threadData.begin(), threadData.end(), std::back_inserter(result), extractFunction); + return result; +} + +KStdVector collectSuspended() { + return collectFromThreadData( + [](mm::ThreadData* threadData) { return threadData->suspensionData().suspended(); }); +} + +void reportProgress(size_t currentIteration, size_t totalIterations) { + if (currentIteration % kDefaultReportingStep == 0) { + std::cout << "Iteration: " << currentIteration << " of " << totalIterations << std::endl; + } +} + +} // namespace + +class ThreadSuspensionTest : public ::testing::Test { +public: + ~ThreadSuspensionTest() { + canStart = true; + shouldStop = true; + for (auto& thread : threads) { + if (thread.joinable()) thread.join(); + } + } + + static constexpr size_t kThreadCount = kDefaultThreadCount; + static constexpr size_t kIterations = kDefaultIterations; + + KStdVector threads; + std::array, kThreadCount> ready{false}; + std::atomic canStart{false}; + std::atomic shouldStop{false}; + + void waitUntilCanStart(size_t threadNumber) { + ready[threadNumber] = true; + while(!canStart) { + std::this_thread::yield(); + } + ready[threadNumber] = false; + } + + void waitUntilThreadsAreReady() { + canStart = false; + while (!std::all_of(ready.begin(), ready.end(), [](bool it) { return it; })) { + std::this_thread::yield(); + } + } +}; + +TEST_F(ThreadSuspensionTest, SimpleStartStop) { + ASSERT_THAT(collectThreadData(), testing::IsEmpty()); + for (size_t i = 0; i < kThreadCount; i++) { + threads.emplace_back([this, i]() { + ScopedMemoryInit init; + auto& suspensionData = init.memoryState()->GetThreadData()->suspensionData(); + EXPECT_EQ(mm::IsThreadSuspensionRequested(), false); + + while(!shouldStop) { + waitUntilCanStart(i); + + EXPECT_FALSE(suspensionData.suspended()); + suspensionData.suspendIfRequested(); + EXPECT_FALSE(suspensionData.suspended()); + } + }); + } + waitUntilThreadsAreReady(); + + for (size_t i = 0; i < kIterations; i++) { + reportProgress(i, kIterations); + canStart = true; + + mm::SuspendThreads(); + auto suspended = collectSuspended(); + EXPECT_THAT(suspended, testing::Each(true)); + EXPECT_EQ(mm::IsThreadSuspensionRequested(), true); + + mm::ResumeThreads(); + + // Wait for threads to run and sync for the next iteration + waitUntilThreadsAreReady(); + + suspended = collectSuspended(); + EXPECT_THAT(suspended, testing::Each(false)); + EXPECT_EQ(mm::IsThreadSuspensionRequested(), false); + } +} + + +TEST_F(ThreadSuspensionTest, SwitchStateToNative) { + ASSERT_THAT(collectThreadData(), testing::IsEmpty()); + + for (size_t i = 0; i < kThreadCount; i++) { + threads.emplace_back([this, i]() { + ScopedMemoryInit init; + auto* threadData = init.memoryState()->GetThreadData(); + EXPECT_EQ(mm::IsThreadSuspensionRequested(), false); + + while(!shouldStop) { + waitUntilCanStart(i); + + EXPECT_EQ(threadData->state(), ThreadState::kRunnable); + SwitchThreadState(threadData, ThreadState::kNative); + EXPECT_EQ(threadData->state(), ThreadState::kNative); + SwitchThreadState(threadData, ThreadState::kRunnable); + EXPECT_EQ(threadData->state(), ThreadState::kRunnable); + } + }); + } + waitUntilThreadsAreReady(); + + for (size_t i = 0; i < kIterations; i++) { + reportProgress(i, kIterations); + canStart = true; + + mm::SuspendThreads(); + EXPECT_EQ(mm::IsThreadSuspensionRequested(), true); + + mm::ResumeThreads(); + EXPECT_EQ(mm::IsThreadSuspensionRequested(), false); + + // Sync for the next iteration. + waitUntilThreadsAreReady(); + } +} + +TEST_F(ThreadSuspensionTest, ConcurrentSuspend) { + ASSERT_THAT(collectThreadData(), testing::IsEmpty()); + std::atomic successCount = 0; + + for (size_t i = 0; i < kThreadCount; i++) { + threads.emplace_back([this, i, &successCount]() { + ScopedMemoryInit init; + auto* currentThreadData = init.memoryState()->GetThreadData(); + EXPECT_EQ(mm::IsThreadSuspensionRequested(), false); + + // Sync with other threads. + ready[i] = true; + waitUntilThreadsAreReady(); + + bool success = mm::SuspendThreads(); + if (success) { + successCount++; + auto allThreadData = collectThreadData(); + auto isCurrentOrSuspended = [currentThreadData](mm::ThreadData* data) { + return data == currentThreadData || data->suspensionData().suspended(); + }; + EXPECT_THAT(allThreadData, testing::Each(testing::Truly(isCurrentOrSuspended))); + EXPECT_FALSE(currentThreadData->suspensionData().suspended()); + mm::ResumeThreads(); + } else { + EXPECT_TRUE(mm::IsThreadSuspensionRequested()); + currentThreadData->suspensionData().suspendIfRequested(); + } + }); + } + for (auto& thread : threads) { + thread.join(); + } + EXPECT_EQ(successCount, 1u); +}