[K/N] Cooperative intrusive list (in preparations for parallel mark)

Merge-request: KOTLIN-MR-670
Merged-by: Aleksej Glushko <aleksei.glushko@jetbrains.com>
This commit is contained in:
Aleksei.Glushko
2023-05-09 15:08:32 +00:00
committed by Space Cloud
parent f0a8d7c86f
commit 57707180c6
6 changed files with 512 additions and 43 deletions
@@ -0,0 +1,116 @@
/*
* Copyright 2010-2023 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
* that can be found in the LICENSE file.
*/
#pragma once
#include <mutex>
#include "IntrusiveList.hpp"
#include "Porting.h"
#include "Mutex.hpp"
namespace kotlin {
/**
* A list for thread-local use with a separate shared part that can be accessed by other threads.
*/
template<typename T, typename Traits = DefaultIntrusiveForwardListTraits<T>>
class CooperativeIntrusiveList : Pinned {
using ListImpl = intrusive_forward_list<T, Traits>;
public:
using value_type = typename ListImpl::value_type;
using size_type = typename ListImpl::size_type;
using reference = typename ListImpl::reference;
using pointer = typename ListImpl::pointer;
CooperativeIntrusiveList() = default;
bool localEmpty() const {
return local_.empty();
}
size_type localSize() const {
return localSize_;
}
/**
* Tries to add `value` to the local list.
* See `intrusive_forward_list.try_push_front`.
*/
bool tryPushLocal(reference value) {
auto pushed = local_.try_push_front(value);
if (pushed) ++localSize_;
return pushed;
}
/**
* Tries to pop a value from the local list.
* See `intrusive_forward_list.try_pop_front`.
*/
pointer tryPopLocal() {
auto popped = local_.try_pop_front();
if (popped) {
--localSize_;
} else {
RuntimeAssert(localEmpty(), "Pop can only fail if the list is empty");
}
return popped;
}
void clearLocal() {
local_.clear();
localSize_ = 0;
}
bool sharedEmpty() const {
return shared_.empty();
}
/**
* Tries to move at most `maxAmount` elements from a from's shared list into `this`'s local list.
* In case some other thread is currently operating with the from's shared list, returns `0`.
* @return the number of elements stolen
*/
size_type tryTransferFrom(CooperativeIntrusiveList<T, Traits>& from, size_type maxAmount) noexcept {
std::unique_lock guard(from.sharedLocked_, std::try_to_lock);
if (!guard || from.sharedEmpty()) {
return 0;
}
auto amount = local_.splice_after(local_.before_begin(),
from.shared_.before_begin(),
from.shared_.end(),
maxAmount);
from.sharedSize_ -= amount;
localSize_ += amount;
return amount;
}
/**
* Moves all of the local items into own shared list.
* @return `0` if the shared list is busy, amount of newly shared items otherwise.
*/
size_type shareAll() noexcept {
RuntimeAssert(!localEmpty(), "Nothing to share");
std::unique_lock guard(sharedLocked_, std::try_to_lock);
if (!guard) return 0;
auto amount = shared_.splice_after(shared_.before_begin(), local_.before_begin(), local_.end(), localSize_);
sharedSize_ += amount;
localSize_ -= amount;
return amount;
}
private:
ListImpl local_;
size_type localSize_ = 0;
ListImpl shared_;
size_type sharedSize_ = 0;
SpinLock<MutexThreadStateHandling::kIgnore> sharedLocked_;
};
}
@@ -0,0 +1,236 @@
/*
* Copyright 2010-2023 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
* that can be found in the LICENSE file.
*/
#include "CooperativeIntrusiveList.hpp"
#include "gmock/gmock.h"
#include "gtest/gtest.h"
#include "ScopedThread.hpp"
#include "TestSupport.hpp"
#include "std_support/Deque.hpp"
#include "std_support/Vector.hpp"
#include "std_support/List.hpp"
using namespace kotlin;
using ::testing::_;
namespace {
class Node : private Pinned {
public:
explicit Node(int value) : value_(value) {}
int& operator*() { return value_; }
const int& operator*() const { return value_; }
void clearNext() noexcept { next_ = nullptr; }
int value() const {
return value_;
}
private:
friend struct DefaultIntrusiveForwardListTraits<Node>;
Node* next() const noexcept { return next_; }
void setNext(Node* next) noexcept {
RuntimeAssert(next, "next cannot be nullptr");
next_ = next;
}
bool trySetNext(Node* next) noexcept {
RuntimeAssert(next, "next cannot be nullptr");
if (next_) return false;
next_ = next;
return true;
}
int value_;
Node* next_ = nullptr;
};
using TestSubject = CooperativeIntrusiveList<Node>;
std_support::vector<int> range(int first, int lastExclusive) {
std_support::vector<int> values;
for (int i = first; i < lastExclusive; ++i) {
values.push_back(i);
}
return values;
}
template<typename Values>
[[nodiscard]] std_support::list<typename TestSubject::value_type> fill(TestSubject& list, Values&& values) {
std_support::list<typename TestSubject::value_type> nodesHandle;
for (int value: values) {
auto& elem = nodesHandle.emplace_back(value);
list.tryPushLocal(elem);
}
return nodesHandle;
}
void drainLocalInto(TestSubject& list, std_support::vector<int>& dest) {
while (auto elem = list.tryPopLocal()) {
dest.push_back(elem->value());
}
}
} // namespace
TEST(CooperativeIntrusiveListTest, Init) {
TestSubject list;
EXPECT_THAT(list.localEmpty(), true);
EXPECT_THAT(list.localSize(), 0);
EXPECT_THAT(list.sharedEmpty(), true);
}
TEST(CooperativeIntrusiveListTest, TryPopLocalEmpty) {
TestSubject list;
auto res = list.tryPopLocal();
EXPECT_THAT(res, nullptr);
}
TEST(CooperativeIntrusiveListTest, TryPushLocalPopLocal) {
TestSubject list;
typename TestSubject::value_type value1(1);
typename TestSubject::value_type value2(2);
bool pushed1 = list.tryPushLocal(value1);
bool pushed2 = list.tryPushLocal(value2);
EXPECT_THAT(pushed1, true);
EXPECT_THAT(pushed2, true);
EXPECT_THAT(list.localEmpty(), false);
EXPECT_THAT(list.localSize(), 2);
EXPECT_THAT(list.sharedEmpty(), true);
std_support::vector<int> popped;
drainLocalInto(list, popped);
EXPECT_THAT(list.localEmpty(), true);
EXPECT_THAT(list.localSize(), 0);
EXPECT_THAT(list.sharedEmpty(), true);
EXPECT_THAT(popped, testing::UnorderedElementsAre(1, 2));
}
TEST(CooperativeIntrusiveListTest, TryPushLocalTwice) {
TestSubject list;
typename TestSubject::value_type value(1);
bool pushed1 = list.tryPushLocal(value);
EXPECT_THAT(pushed1, true);
bool pushed2 = list.tryPushLocal(value);
EXPECT_THAT(pushed2, false);
EXPECT_THAT(list.localEmpty(), false);
EXPECT_THAT(list.localSize(), 1);
EXPECT_THAT(list.sharedEmpty(), true);
}
TEST(CooperativeIntrusiveListTest, ShareSome) {
TestSubject list;
auto values = range(0, 10);
auto nodeHandle = fill(list, values);
EXPECT_THAT(list.localEmpty(), false);
EXPECT_THAT(list.localSize(), values.size());
EXPECT_THAT(list.sharedEmpty(), true);
auto sharedAmount = list.shareAll();
EXPECT_THAT(sharedAmount, values.size());
EXPECT_THAT(list.localEmpty(), true);
EXPECT_THAT(list.sharedEmpty(), false);
}
TEST(CooperativeIntrusiveListTest, TryTransferFromEmpty) {
TestSubject from;
TestSubject thief;
auto stolenAmount = thief.tryTransferFrom(from, 1);
EXPECT_THAT(stolenAmount, 0);
}
TEST(CooperativeIntrusiveListTest, TryTransferHalf) {
TestSubject from;
auto values = range(0, 10);
auto nodeHandle = fill(from, values);
from.shareAll();
TestSubject thief;
auto toTransfer = values.size() / 2;
auto stolenAmount = thief.tryTransferFrom(from, toTransfer);
EXPECT_THAT(stolenAmount, toTransfer);
EXPECT_THAT(thief.localSize(), stolenAmount);
from.tryTransferFrom(from, values.size());
EXPECT_THAT(from.sharedEmpty(), true);
std_support::vector<int> allTheElements;
drainLocalInto(from, allTheElements);
drainLocalInto(thief, allTheElements);
EXPECT_THAT(allTheElements, testing::UnorderedElementsAreArray(values));
}
TEST(CooperativeIntrusiveListTest, TryTransferAllEventually) {
TestSubject from;
auto values = range(0, 10);
auto nodeHandle = fill(from, values);
from.shareAll();
TestSubject thief;
for (std::size_t i = 0; i < values.size(); ++i) {
auto stolenAmount = thief.tryTransferFrom(from, 1);
EXPECT_THAT(stolenAmount, 1);
}
EXPECT_THAT(from.sharedEmpty(), true);
EXPECT_THAT(thief.localSize(), values.size());
std_support::vector<int> allTheElements;
drainLocalInto(from, allTheElements);
drainLocalInto(thief, allTheElements);
EXPECT_THAT(allTheElements, testing::UnorderedElementsAreArray(values));
}
TEST(CooperativeIntrusiveListTest, TransferingPingPong) {
TestSubject list1;
TestSubject list2;
const auto size = 100;
auto values = range(0, size);
auto nodesHandle1 = fill(list1, values);
auto nodesHandle2 = fill(list2, values);
std::atomic ready = false;
auto kIters = 10000;
std_support::vector<ScopedThread> threads;
for (int tIdx = 0; tIdx < 2; ++tIdx) {
threads.emplace_back([&ready, kIters, tIdx, &list1, &list2] {
TestSubject& self = tIdx % 2 == 0 ? list1 : list2;
TestSubject& from = tIdx % 2 == 0 ? list2 : list1;
while (!ready.load()) {
std::this_thread::yield();
}
for (int iter = 0; iter < kIters; ++iter) {
if (!self.localEmpty()) self.shareAll();
self.tryTransferFrom(from, size / 2);
if (!self.localEmpty()) self.shareAll();
self.tryTransferFrom(from, size);
if (auto popped = self.tryPopLocal()) {
popped->clearNext();
self.tryPushLocal(*popped);
}
}
});
}
ready = true;
for (auto& thr: threads) {
thr.join();
}
// check nothing is lost
list1.tryTransferFrom(list1, size * 2);
list2.tryTransferFrom(list2, size * 2);
std_support::vector<int> allTheElements;
drainLocalInto(list1, allTheElements);
drainLocalInto(list2, allTheElements);
std_support::vector<int> expected;
expected.insert(expected.end(), values.begin(), values.end());
expected.insert(expected.end(), values.begin(), values.end());
EXPECT_THAT(allTheElements, testing::UnorderedElementsAreArray(expected));
}
@@ -339,7 +339,26 @@ public:
}
}
// TODO: Implement splice_after.
// Moves at most `maxCount` first elements from the range `(firstExcl, lastExcl)` after the element pointed by `insertAfter`.
// No elements are copied or moved, only the internal pointers of the list nodes are re-pointed.
// The behavior is undefined if `insertAfter` is an iterator in the range `[firstExcl, lastExcl)`.
// Complexity: O(min(maxCount, std::distance(first, last)))
size_type splice_after(iterator insertAfter, iterator firstExcl, iterator lastExcl, size_type maxCount) {
auto firstIncl = std::next(firstExcl);
if (firstIncl == lastExcl) return 0;
auto lastIncl = firstExcl;
size_type count = 0;
while (std::next(lastIncl) != lastExcl && count < maxCount) {
RuntimeAssert(lastIncl != insertAfter, "Position to splice after must not be in the spliced range");
++lastIncl;
++count;
}
lastExcl = std::next(lastIncl);
setNext(firstExcl.node_, lastExcl.node_);
setNext(lastIncl.node_, next(insertAfter.node_));
setNext(insertAfter.node_, firstIncl.node_);
return count;
}
private:
static pointer next(const_pointer node) noexcept { return Traits::next(*node); }
@@ -348,8 +367,8 @@ private:
static bool trySetNext(pointer node, pointer next) noexcept { return Traits::trySetNext(*node, next); }
pointer head() noexcept { return reinterpret_cast<pointer>(headStorage_); }
const_pointer head() const noexcept { return reinterpret_cast<const_pointer>(headStorage_); }
pointer head() noexcept { return &head_; }
const_pointer head() const noexcept { return &head_; }
static pointer tail() noexcept { return reinterpret_cast<pointer>(tailStorage_); }
@@ -364,7 +383,10 @@ private:
return iterator(&value);
}
alignas(value_type) char headStorage_[sizeof(value_type)] = {0};
union {
value_type head_; // for debugger
alignas(value_type) char headStorage_[sizeof(value_type)] = {0};
};
alignas(value_type) static inline char tailStorage_[sizeof(value_type)] = {0};
};
@@ -629,7 +629,7 @@ TYPED_TEST(ForwardListTest, EraseAfterEmptyRangeFront) {
EXPECT_ELEMENTS_ARE(list, 1, 2, 3, 4);
}
TEST(InstrusiveForwardListTest, TryPushFrontSuccess) {
TEST(IntrusiveForwardListTest, TryPushFrontSuccess) {
using List = intrusive_forward_list<Node>;
auto values = create<List>({1, 2, 3, 4});
List list(values.begin(), values.end());
@@ -640,7 +640,7 @@ TEST(InstrusiveForwardListTest, TryPushFrontSuccess) {
EXPECT_ELEMENTS_ARE(list, 5, 1, 2, 3, 4);
}
TEST(InstrusiveForwardListTest, TryPushFrontFailure) {
TEST(IntrusiveForwardListTest, TryPushFrontFailure) {
using List = intrusive_forward_list<Node>;
auto values = create<List>({1, 2, 3, 4});
List list(values.begin(), values.end());
@@ -650,7 +650,7 @@ TEST(InstrusiveForwardListTest, TryPushFrontFailure) {
EXPECT_ELEMENTS_ARE(list, 1, 2, 3, 4);
}
TEST(InstrusiveForwardListTest, TryPushFrontEmptySuccess) {
TEST(IntrusiveForwardListTest, TryPushFrontEmptySuccess) {
using List = intrusive_forward_list<Node>;
List list;
typename List::value_type value(5);
@@ -660,7 +660,7 @@ TEST(InstrusiveForwardListTest, TryPushFrontEmptySuccess) {
EXPECT_ELEMENTS_ARE(list, 5);
}
TEST(InstrusiveForwardListTest, TryPushFrontEmptyFailure) {
TEST(IntrusiveForwardListTest, TryPushFrontEmptyFailure) {
using List = intrusive_forward_list<Node>;
List list;
typename List::value_type value(5);
@@ -697,3 +697,84 @@ TEST(IntrusiveForwardListTest, TryPopFrontFromEmpty) {
EXPECT_THAT(result, nullptr);
EXPECT_ELEMENTS_ARE(list);
}
template<typename List>
typename List::iterator before_end(List& list) {
auto cur = list.before_begin();
auto next = list.begin();
while (next != list.end()) {
++cur;
++next;
}
return cur;
}
TEST(IntrusiveForwardListTest, SpliceAfterBeforeBeginAll) {
using List = intrusive_forward_list<Node>;
auto values1 = create<List>({1, 2, 3, 4});
auto values2 = create<List>({11, 12, 13, 14});
List l1(values1.begin(), values1.end());
List l2(values2.begin(), values2.end());
auto spliced = l1.splice_after(l1.before_begin(), l2.before_begin(), l2.end(), values2.size());
EXPECT_THAT(spliced, values2.size());
EXPECT_ELEMENTS_ARE(l1, 11, 12, 13, 14, 1, 2, 3, 4);
EXPECT_ELEMENTS_ARE(l2);
}
TEST(IntrusiveForwardListTest, SpliceAfterBeforeEndAll) {
using List = intrusive_forward_list<Node>;
auto values1 = create<List>({1, 2, 3, 4});
auto values2 = create<List>({11, 12, 13, 14});
List l1(values1.begin(), values1.end());
List l2(values2.begin(), values2.end());
auto spliced = l1.splice_after(before_end(l1), l2.before_begin(), l2.end(), values2.size());
EXPECT_THAT(spliced, values2.size());
EXPECT_ELEMENTS_ARE(l1, 1, 2, 3, 4, 11, 12, 13, 14);
EXPECT_ELEMENTS_ARE(l2);
}
TEST(IntrusiveForwardListTest, SpliceAfterMidMidHalf) {
using List = intrusive_forward_list<Node>;
auto values1 = create<List>({1, 2, 3, 4});
auto values2 = create<List>({11, 12, 13, 14});
List l1(values1.begin(), values1.end());
List l2(values2.begin(), values2.end());
auto spliced = l1.splice_after(std::next(l1.begin()), l2.begin(), l2.end(), 2);
EXPECT_THAT(spliced, 2);
EXPECT_ELEMENTS_ARE(l1, 1, 2, 12, 13, 3, 4);
EXPECT_ELEMENTS_ARE(l2, 11, 14);
}
TEST(IntrusiveForwardListTest, SpliceAfterBeforeBeginOwnTail) {
using List = intrusive_forward_list<Node>;
auto values = create<List>({1, 2, 3, 4});
List list(values.begin(), values.end());
auto spliced = list.splice_after(list.before_begin(), std::next(list.begin()), list.end(), 2);
EXPECT_THAT(spliced, 2);
EXPECT_ELEMENTS_ARE(list, 3, 4, 1, 2);
}
TEST(IntrusiveForwardListTest, SpliceAfterLessThanAvailable) {
using List = intrusive_forward_list<Node>;
auto values = create<List>({1, 2, 3, 4});
List from(values.begin(), values.end());
List to;
auto spliced = to.splice_after(to.before_begin(), from.before_begin(), from.end(), 2);
EXPECT_THAT(spliced, 2);
EXPECT_ELEMENTS_ARE(to, 1, 2);
EXPECT_ELEMENTS_ARE(from, 3, 4);
}
TEST(IntrusiveForwardListTest, SpliceAfterMoreThanAvailable) {
using List = intrusive_forward_list<Node>;
auto values = create<List>({1, 2, 3, 4});
List from(values.begin(), values.end());
List to;
auto spliced = to.splice_after(to.before_begin(), from.before_begin(), std::next(from.begin(), 2), 4);
EXPECT_THAT(spliced, 2);
EXPECT_ELEMENTS_ARE(to, 1, 2);
EXPECT_ELEMENTS_ARE(from, 3, 4);
}
+21 -16
View File
@@ -34,13 +34,12 @@ enum class MutexThreadStateHandling {
template <MutexThreadStateHandling threadStateHandling>
class SpinLock;
template <>
class SpinLock<MutexThreadStateHandling::kIgnore> : private Pinned {
namespace internal {
class SpinLockBase : private Pinned {
public:
void lock() noexcept {
while(flag_.test_and_set(std::memory_order_acquire)) {
yield();
}
bool try_lock() noexcept {
return !flag_.test_and_set(std::memory_order_acquire);
}
void unlock() noexcept {
@@ -49,7 +48,20 @@ public:
private:
std::atomic_flag flag_ = ATOMIC_FLAG_INIT;
};
}
template <>
class SpinLock<MutexThreadStateHandling::kIgnore> : public internal::SpinLockBase {
public:
void lock() noexcept {
while(!try_lock()) {
yield();
}
}
private:
// No need to check for external calls, because we explicitly ignore thread state.
static NO_EXTERNAL_CALLS_CHECK NO_INLINE void yield() {
std::this_thread::yield();
@@ -57,26 +69,19 @@ private:
};
template <>
class SpinLock<MutexThreadStateHandling::kSwitchIfRegistered> : private Pinned {
class SpinLock<MutexThreadStateHandling::kSwitchIfRegistered> : public internal::SpinLockBase {
public:
void lock() noexcept {
// Fast path without thread state switching.
if (!flag_.test_and_set(std::memory_order_acquire)) {
if (try_lock()) {
return;
}
kotlin::NativeOrUnregisteredThreadGuard guard(/* reentrant = */ true);
while (flag_.test_and_set(std::memory_order_acquire)) {
while (!try_lock()) {
std::this_thread::yield();
}
}
void unlock() noexcept {
flag_.clear(std::memory_order_release);
}
private:
std::atomic_flag flag_ = ATOMIC_FLAG_INIT;
};
template <MutexThreadStateHandling threadStateHandling>
@@ -19,9 +19,36 @@ using namespace kotlin;
template <typename T>
class MutexTest : public testing::Test {};
using LockTypes = testing::Types<SpinLock<MutexThreadStateHandling::kIgnore>, SpinLock<MutexThreadStateHandling::kSwitchIfRegistered>>;
using LockTypes = testing::Types<
SpinLock<MutexThreadStateHandling::kIgnore>,
SpinLock<MutexThreadStateHandling::kSwitchIfRegistered>,
RWSpinLock<MutexThreadStateHandling::kIgnore>
>;
TYPED_TEST_SUITE(MutexTest, LockTypes);
TYPED_TEST(MutexTest, Lock) {
using LockUnderTest = TypeParam;
LockUnderTest mutex;
mutex.lock();
mutex.unlock();
// Check that unlock unlocked.
mutex.lock();
mutex.unlock();
}
TYPED_TEST(MutexTest, TryLock) {
using LockUnderTest = TypeParam;
LockUnderTest mutex;
EXPECT_TRUE(mutex.try_lock());
EXPECT_FALSE(mutex.try_lock());
mutex.unlock();
mutex.lock();
EXPECT_FALSE(mutex.try_lock());
mutex.unlock();
}
TYPED_TEST(MutexTest, SmokeDetachedThread) {
using LockUnderTest = TypeParam;
@@ -49,24 +76,6 @@ TYPED_TEST(MutexTest, SmokeDetachedThread) {
EXPECT_EQ(protectedCounter, 2);
}
TEST(RWSpinLockTest, Lock) {
RWSpinLock<MutexThreadStateHandling::kIgnore> mutex;
mutex.lock();
mutex.unlock();
// Check that unlock unlocked.
mutex.lock();
mutex.unlock();
}
TEST(RWSpinLockTest, TryLock) {
RWSpinLock<MutexThreadStateHandling::kIgnore> mutex;
EXPECT_TRUE(mutex.try_lock());
EXPECT_FALSE(mutex.try_lock());
mutex.unlock();
mutex.lock();
EXPECT_FALSE(mutex.try_lock());
mutex.unlock();
}
TEST(RWSpinLockTest, LockShared) {
RWSpinLock<MutexThreadStateHandling::kIgnore> mutex;