[K/N][Runtime] Support suspending mutators in the new MM

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)

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<ThreadState> expected) noexcept {
+    // no-op, used by the new MM only.
+}
+
 MemoryState* kotlin::mm::GetMemoryState() {
     return ::memoryState;
 }

kotlin-native/runtime/src/main/cpp/Memory.h

@@ -17,6 +17,7 @@
 #ifndef RUNTIME_MEMORY_H
 #define RUNTIME_MEMORY_H
 
+#include <ostream>
 #include <utility>
 
 #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<ThreadState> 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<ThreadState> expected) noexcept {
+    AssertThreadState(mm::GetMemoryState(), expected);
+}
+
 // Scopely sets the given thread state for the given thread.
 class ThreadStateGuard final : private Pinned {
 public:

kotlin-native/runtime/src/main/cpp/Utils.hpp

@@ -6,6 +6,8 @@
 #ifndef RUNTIME_UTILS_H
 #define RUNTIME_UTILS_H
 
+#include <type_traits>
+
 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<typename T1, typename T2>
+class AutoReset final : private Pinned {
+    static_assert(std::is_assignable<T1, T2>::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

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<gc::GC>::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<ThreadState> state_;
     ShadowStack shadowStack_;
     gc::GC::ThreadData gc_;
     ObjectFactory<gc::GC>::ThreadQueue objectFactoryThreadQueue_;
     KStdVector<std::pair<ObjHeader**, ObjHeader*>> initializingSingletons_;
+    ThreadSuspensionData suspensionData_;
 };
 
 } // namespace mm

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<ThreadState> 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<ThreadState> expected) noexcept {
+    AssertThreadState(thread->GetThreadData(), expected);
+}
+
 ThreadState kotlin::GetThreadState(MemoryState* thread) noexcept {
     return thread->GetThreadData()->state();
 }

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 <ostream>
+
 #include <Common.h>
 #include <Utils.hpp>
 
 #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<ThreadState> 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<ThreadState> 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

kotlin-native/runtime/src/mm/cpp/ThreadStateTest.cpp

@@ -41,6 +41,9 @@ private:
 //static
 testing::MockFunction<int32_t(int32_t)>* 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();

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 <condition_variable>
+#include <thread>
+#include <mutex>
+
+namespace {
+
+bool isSuspendedOrNative(kotlin::mm::ThreadData& thread) noexcept {
+    auto& suspensionData = thread.suspensionData();
+    return suspensionData.suspended() || suspensionData.state() == kotlin::ThreadState::kNative;
+}
+
+template<typename F>
+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<bool> 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();
+}

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 <atomic>
+
+#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<ThreadState> state_;
+    std::atomic<bool> 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

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 <gtest/gtest.h>
+#include <gmock/gmock.h>
+
+#include <thread>
+#include <TestSupport.hpp>
+
+#include <iostream>
+
+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<mm::ThreadData*> collectThreadData() {
+    KStdVector<mm::ThreadData*> result;
+    auto iter = mm::ThreadRegistry::Instance().Iter();
+    for (auto& thread : iter) {
+        result.push_back(&thread);
+    }
+    return result;
+}
+
+template<typename T, typename F>
+KStdVector<T> collectFromThreadData(F extractFunction) {
+    KStdVector<T> result;
+    auto threadData = collectThreadData();
+    std::transform(threadData.begin(), threadData.end(), std::back_inserter(result), extractFunction);
+    return result;
+}
+
+KStdVector<bool> collectSuspended() {
+    return collectFromThreadData<bool>(
+            [](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<std::thread> threads;
+    std::array<std::atomic<bool>, kThreadCount> ready{false};
+    std::atomic<bool> canStart{false};
+    std::atomic<bool> 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<size_t> 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);
+}