Using a different Concurrent Queue code.

This commit is contained in:
Nicolas Pixel Noble
2022-05-03 20:04:57 -07:00
parent 8e31328f5c
commit db04b7239b
9 changed files with 517 additions and 16 deletions
-3
View File
@@ -67,9 +67,6 @@
[submodule "third_party/capstone"]
path = third_party/capstone
url = https://github.com/capstone-engine/capstone.git
[submodule "third_party/concurrentqueue"]
path = third_party/concurrentqueue
url = https://github.com/cameron314/concurrentqueue.git
[submodule "third_party/curl"]
path = third_party/curl
url = https://github.com/curl/curl
+7 -7
View File
@@ -51,7 +51,7 @@ size_t PCSX::UvThreadOp::s_dataDownloadSinceLastTick;
std::atomic<size_t> PCSX::UvThreadOp::s_dataReadLastTick;
std::atomic<size_t> PCSX::UvThreadOp::s_dataWrittenLastTick;
std::atomic<size_t> PCSX::UvThreadOp::s_dataDownloadLastTick;
moodycamel::ConcurrentQueue<PCSX::UvThreadOp::UvRequest> PCSX::UvThreadOp::s_queue;
ConcurrentQueue<PCSX::UvThreadOp::UvRequest> PCSX::UvThreadOp::s_queue;
PCSX::UvThreadOpListType PCSX::UvThreadOp::s_allOps;
uv_loop_t PCSX::UvThreadOp::s_uvLoop;
uv_timer_t PCSX::UvThreadOp::s_curlTimeout;
@@ -82,7 +82,7 @@ void PCSX::UvThreadOp::startThread() {
s_dataDownloadLastTick = 0;
uv_async_init(&s_uvLoop, &s_kicker, [](uv_async_t *async) {
UvRequest req;
while (s_queue.try_dequeue(req)) {
while (s_queue.Dequeue(req)) {
req(async->loop);
}
});
@@ -712,7 +712,7 @@ PCSX::UvFifo::UvFifo(uv_tcp_t *tcp) : File(File::FileType::RW_STREAM) {
fifo->m_buffer = nullptr;
Slice slice;
slice.acquire(b, nread);
fifo->m_queue.enqueue(std::move(slice));
fifo->m_queue.Enqueue(std::move(slice));
fifo->m_size.fetch_add(nread);
});
}
@@ -739,7 +739,7 @@ ssize_t PCSX::UvFifo::read(void *dest_, size_t size) {
if (m_size.load() == 0) {
return ret == 0 ? -1 : ret;
}
while (!m_queue.try_dequeue(m_slice))
while (!m_queue.Dequeue(m_slice))
;
}
auto toRead = std::min(size, static_cast<size_t>(m_slice.size()) - m_currentPtr);
@@ -803,7 +803,7 @@ void PCSX::UvFifoListener::start(unsigned port, uv_loop_t *loop, uv_async_t *asy
uv_async_init(loop, async, [](uv_async_t *async) {
UvFifoListener *self = reinterpret_cast<UvFifoListener *>(async->data);
UvFifo *fifo = nullptr;
while (self->m_pending.try_dequeue(fifo)) {
while (self->m_pending.Dequeue(fifo)) {
self->m_cb(fifo);
}
});
@@ -830,7 +830,7 @@ void PCSX::UvFifoListener::start(unsigned port, uv_loop_t *loop, uv_async_t *asy
uv_tcp_init(loop, tcp);
if (uv_accept(reinterpret_cast<uv_stream_t *>(server), reinterpret_cast<uv_stream_t *>(tcp)) == 0) {
UvFifo *fifo = new UvFifo(tcp);
listener->m_pending.enqueue(fifo);
listener->m_pending.Enqueue(fifo);
uv_async_send(listener->m_async);
} else {
uv_close(reinterpret_cast<uv_handle_t *>(tcp),
@@ -848,7 +848,7 @@ void PCSX::UvFifoListener::stop() {
request([this](auto loop) {
uv_close(reinterpret_cast<uv_handle_t *>(&m_server), [](uv_handle_t *handle) {
UvFifoListener *listener = reinterpret_cast<UvFifoListener *>(handle->data);
listener->m_pending.enqueue(nullptr);
listener->m_pending.Enqueue(nullptr);
uv_async_send(listener->m_async);
});
});
+5 -5
View File
@@ -28,7 +28,7 @@
#include <string>
#include <thread>
#include "concurrentqueue/concurrentqueue.h"
#include "cq/concurrent_queue.h"
#include "support/file.h"
#include "support/list.h"
@@ -83,7 +83,7 @@ class UvThreadOp : public UvThreadOpListType::Node {
std::promise<void> m_cacheBarrier;
typedef std::function<void(uv_loop_t*)> UvRequest;
static void request(UvRequest&& req) {
s_queue.enqueue(std::move(req));
s_queue.Enqueue(std::move(req));
uv_async_send(&s_kicker);
}
static CURLM* s_curlMulti;
@@ -109,7 +109,7 @@ class UvThreadOp : public UvThreadOpListType::Node {
static std::atomic<size_t> s_dataDownloadLastTick;
static constexpr uint64_t c_tick = 500;
static moodycamel::ConcurrentQueue<UvRequest> s_queue;
static ConcurrentQueue<UvRequest> s_queue;
static UvThreadOpListType s_allOps;
};
@@ -216,7 +216,7 @@ class UvFifo : public File, public UvThreadOp {
std::atomic<bool> m_closed = false;
uv_write_t m_writeReq;
const size_t c_chunkSize = 4096;
moodycamel::ConcurrentQueue<Slice> m_queue;
ConcurrentQueue<Slice> m_queue;
std::atomic<size_t> m_size = 0;
Slice m_slice;
size_t m_currentPtr = 0;
@@ -233,7 +233,7 @@ class UvFifoListener : public UvThreadOp {
uv_async_t* m_async = nullptr;
uv_tcp_t m_server = {};
std::function<void(UvFifo*)> m_cb;
moodycamel::ConcurrentQueue<UvFifo*> m_pending;
ConcurrentQueue<UvFifo*> m_pending;
};
} // namespace PCSX
+201
View File
@@ -0,0 +1,201 @@
#pragma once
#include <atomic>
#include "reclaimer.h"
template <typename T>
class ConcreteReclaimer;
template <typename T>
class ConcurrentQueue {
static_assert(std::is_copy_constructible_v<T>, "T requires copy constructor");
struct Node;
struct RegularNode;
friend ConcreteReclaimer<T>;
public:
ConcurrentQueue()
: head_(new Node),
tail_(head_.load(std::memory_order_relaxed)),
size_(0) {}
~ConcurrentQueue() {
Node* p = head_.load(std::memory_order_acquire);
while (p != nullptr) {
Node* tmp = p;
p = p->next.load(std::memory_order_acquire);
tmp->Release();
}
}
ConcurrentQueue(const ConcurrentQueue&) = delete;
ConcurrentQueue(ConcurrentQueue&&) = delete;
ConcurrentQueue& operator=(const ConcurrentQueue& other) = delete;
ConcurrentQueue& operator=(ConcurrentQueue&& other) = delete;
template <typename... Args>
void Emplace(Args&&... args);
void Enqueue(const T& value) {
static_assert(std::is_copy_constructible<T>::value,
"T must be copy constructible");
Emplace(value);
};
void Enqueue(T&& value) {
static_assert(std::is_constructible_v<T, T&&>,
"T must be constructible with T&&");
Emplace(std::forward<T>(value));
}
bool Dequeue(T& data);
size_t size() const { return size_.load(std::memory_order_relaxed); }
private:
Node* get_head() const { return head_.load(std::memory_order_acquire); }
Node* get_tail() const { return tail_.load(std::memory_order_acquire); }
// Get safe node and its next, ensure next is the succeed of node
// and both pointer are safety.
// REQUIRE: atomic_node is head_ or tail_.
void AcquireSafeNodeAndNext(std::atomic<Node*>& atomic_node, Node** node_ptr,
Node** next_ptr, HazardPointer& node_hp,
HazardPointer& next_hp);
// Invoke this function when the node can be reclaimed
static void OnDeleteNode(void* ptr) { static_cast<Node*>(ptr)->Release(); }
struct Node {
Node() : next(nullptr) {}
virtual ~Node() = default;
virtual void Release() { delete this; }
Node* get_next() const { return next.load(std::memory_order_acquire); }
std::atomic<Node*> next;
};
struct RegularNode : Node {
template <typename... Args>
RegularNode(Args&&... args) : value(std::forward<Args>(args)...) {}
~RegularNode() override = default;
void Release() override { delete this; }
T value;
};
std::atomic<Node*> head_;
std::atomic<Node*> tail_;
std::atomic<size_t> size_;
static Reclaimer::HazardPointerList global_hp_list_;
};
template <typename T>
Reclaimer::HazardPointerList ConcurrentQueue<T>::global_hp_list_;
template <typename T>
class ConcreteReclaimer : public Reclaimer {
friend ConcurrentQueue<T>;
private:
ConcreteReclaimer(HazardPointerList& hp_list) : Reclaimer(hp_list) {}
~ConcreteReclaimer() override = default;
static ConcreteReclaimer<T>& GetInstance() {
thread_local static ConcreteReclaimer reclaimer(
ConcurrentQueue<T>::global_hp_list_);
return reclaimer;
}
};
template <typename T>
void ConcurrentQueue<T>::AcquireSafeNodeAndNext(std::atomic<Node*>& atomic_node,
Node** node_ptr,
Node** next_ptr,
HazardPointer& node_hp,
HazardPointer& next_hp) {
Node* node = atomic_node.load(std::memory_order_acquire);
Node* next;
Node* temp_node;
Node* temp_next;
auto& reclaimer = ConcreteReclaimer<T>::GetInstance();
do {
do {
// 1.UnMark old node;
node_hp.UnMark();
temp_node = node;
// 2. Mark node.
node_hp = HazardPointer(&reclaimer, node);
node = atomic_node.load(std::memory_order_acquire);
// 3. Make sure the node is still the one we mark before.
} while (temp_node != node);
// 4. UnMark old next.
next_hp.UnMark();
next = node->get_next();
temp_next = next;
// 5. Mark next.
next_hp = HazardPointer(&reclaimer, next);
next = node->get_next();
// 6. Make sure the next is still the succeed of first.
} while (temp_next != next);
*node_ptr = node;
*next_ptr = next;
}
template <typename T>
template <typename... Args>
void ConcurrentQueue<T>::Emplace(Args&&... args) {
static_assert(std::is_constructible_v<T, Args&&...>,
"T must be constructible with Args&&...");
RegularNode* new_node = new RegularNode(std::forward<Args>(args)...);
Node* tail;
Node* next;
HazardPointer tail_hp, next_hp;
while (true) {
AcquireSafeNodeAndNext(tail_, &tail, &next, tail_hp, next_hp);
if (tail != get_tail()) continue; // Are tail and next consistent?
if (nullptr == next) { // Was tail point to last node?
// Try to link node at the end of the linked list.
if (tail->next.compare_exchange_strong(next, new_node)) break;
}
else {
// Try to swing tail to the next node.
tail_.compare_exchange_weak(tail, next);
}
}
// Enqueue is done. Try to swing tail to the inserted node.
tail_.compare_exchange_weak(tail, new_node);
size_.fetch_add(1, std::memory_order_relaxed);
}
template <typename T>
bool ConcurrentQueue<T>::Dequeue(T& value) {
HazardPointer head_hp;
HazardPointer next_hp;
Node* head;
Node* next;
Node* tail;
while (true) {
AcquireSafeNodeAndNext(head_, &head, &next, head_hp, next_hp);
tail = get_tail();
if (head != get_head()) continue; // Are head, tail, and next consistent?
if (head == tail) { // Is queue empty or tail falling behind?
if (nullptr == next) return false; // Queue is empty;
// Tail is falling behind. Try to advance it.
tail_.compare_exchange_weak(tail, next);
}
else {
// Try to swing head to the next node.
if (head_.compare_exchange_strong(head, next)) break;
}
}
size_.fetch_sub(1, std::memory_order_relaxed);
auto& reclaimer = ConcreteReclaimer<T>::GetInstance();
reclaimer.ReclaimLater(head, ConcurrentQueue<T>::OnDeleteNode);
reclaimer.ReclaimNoHazardPointer();
value = std::move(static_cast<RegularNode*>(next)->value);
return true;
}
+115
View File
@@ -0,0 +1,115 @@
#include "reclaimer.h"
Reclaimer::~Reclaimer() {
// The Reclaimer destruct when the thread exit
// 1.Hand over the hazard pointer
for (int i = 0; i < hp_list_.size(); ++i) {
// If assert fired, you should make sure no pointer is marked as hazard
// before thread exit
assert(nullptr == hp_list_[i]->ptr.load(std::memory_order_relaxed));
hp_list_[i]->flag.clear();
}
// 2.Make sure reclaim all no hazard pointers
for (auto it = reclaim_map_.begin(); it != reclaim_map_.end();) {
// Wait until pointer is no hazard
while (Hazard(it->first)) {
std::this_thread::yield();
}
ReclaimNode* node = it->second;
node->delete_func(node->ptr);
delete node;
it = reclaim_map_.erase(it);
}
}
Reclaimer::HPIndex Reclaimer::MarkHazard(void* ptr) {
if (nullptr == ptr) return HP_INDEX_NULL;
for (int i = 0; i < hp_list_.size(); ++i) {
InternalHazardPointer* hp = hp_list_[i];
if (nullptr == hp->ptr.load(std::memory_order_relaxed)) {
hp->ptr.store(ptr, std::memory_order_release);
return i;
}
}
TryAcquireHazardPointer();
int index = hp_list_.size() - 1;
hp_list_[index]->ptr.store(ptr, std::memory_order_release);
return index;
}
void Reclaimer::ReclaimNoHazardPointer() {
if (reclaim_map_.size() < kCoefficient * global_hp_list_.get_size()) {
return;
}
// Used to speed up the inspection of the ptr.
std::unordered_set<void*> not_allow_delete_set;
std::atomic<InternalHazardPointer*>& head = global_hp_list_.head;
InternalHazardPointer* p = head.load(std::memory_order_acquire);
do {
void* const ptr = p->ptr.load(std::memory_order_consume);
if (nullptr != ptr) {
not_allow_delete_set.insert(ptr);
}
p = p->next.load(std::memory_order_acquire);
} while (p);
for (auto it = reclaim_map_.begin(); it != reclaim_map_.end();) {
if (not_allow_delete_set.find(it->first) == not_allow_delete_set.end()) {
ReclaimNode* node = it->second;
node->delete_func(node->ptr);
reclaim_pool_.Push(node);
it = reclaim_map_.erase(it);
}
else {
++it;
}
}
}
bool Reclaimer::Hazard(void* const ptr) {
std::atomic<InternalHazardPointer*>& head = global_hp_list_.head;
InternalHazardPointer* p = head.load(std::memory_order_acquire);
do {
if (p->ptr.load(std::memory_order_consume) == ptr) {
return true;
}
p = p->next.load(std::memory_order_acquire);
} while (p != nullptr);
return false;
}
void Reclaimer::TryAcquireHazardPointer() {
std::atomic<InternalHazardPointer*>& head = global_hp_list_.head;
InternalHazardPointer* p = head.load(std::memory_order_acquire);
InternalHazardPointer* hp = nullptr;
do {
// Try to get the idle hazard pointer.
if (!p->flag.test_and_set()) {
hp = p;
break;
}
p = p->next.load(std::memory_order_acquire);
} while (p != nullptr);
if (nullptr == hp) {
// No idle hazard pointer, allocate new one.
InternalHazardPointer* new_head = new InternalHazardPointer();
new_head->flag.test_and_set();
hp = new_head;
global_hp_list_.size.fetch_add(1, std::memory_order_release);
InternalHazardPointer* old_head = head.load(std::memory_order_acquire);
do {
new_head->next = old_head;
} while (!head.compare_exchange_weak(old_head, new_head,
std::memory_order_release,
std::memory_order_relaxed));
}
hp_list_.push_back(hp);
}
+177
View File
@@ -0,0 +1,177 @@
#pragma once
#include <atomic>
#include <cassert>
#include <functional>
#include <thread>
#include <unordered_map>
#include <unordered_set>
#include <vector>
class HazardPointer;
class Reclaimer {
typedef int HPIndex;
friend HazardPointer;
private:
// A coefficient that used to calcuate the max number
// of reclaim node in reclaim list.
static const int kCoefficient = 4 + 1 / 4;
static const HPIndex HP_INDEX_NULL = -1;
struct InternalHazardPointer {
InternalHazardPointer() : flag(), ptr(nullptr), next(nullptr) { flag.clear(); }
~InternalHazardPointer() {}
InternalHazardPointer(const InternalHazardPointer& other) = delete;
InternalHazardPointer(InternalHazardPointer&& other) = delete;
InternalHazardPointer& operator=(const InternalHazardPointer& other) =
delete;
InternalHazardPointer& operator=(InternalHazardPointer&& other) = delete;
std::atomic_flag flag;
// We must use atomic pointer to ensure the modification
// is visible to other threads.
std::atomic<void*> ptr;
std::atomic<InternalHazardPointer*> next;
};
public:
struct HazardPointerList {
HazardPointerList() : head(new InternalHazardPointer()), size(0) {}
~HazardPointerList() {
// HazardPointerList destruct when program exit.
InternalHazardPointer* p = head.load(std::memory_order_acquire);
while (p) {
InternalHazardPointer* temp = p;
p = p->next.load(std::memory_order_consume);
delete temp;
}
}
size_t get_size() const { return size.load(std::memory_order_consume); }
std::atomic<InternalHazardPointer*> head;
std::atomic<int> size;
};
Reclaimer(const Reclaimer&) = delete;
Reclaimer(Reclaimer&&) = delete;
Reclaimer& operator=(const Reclaimer&) = delete;
Reclaimer& operator=(Reclaimer&&) = delete;
// Mark ptr as hazard.
HPIndex MarkHazard(void* ptr);
void UnMarkHazard(HPIndex index) {
if (index == HP_INDEX_NULL) return;
assert(index >= 0 && index < hp_list_.size());
hp_list_[index]->ptr.store(nullptr, std::memory_order_release);
}
// Get ptr that marked as hazard at the index of hazard_pointers_ array.
void* GetHazardPtr(HPIndex index) {
if (index == HP_INDEX_NULL) return nullptr;
assert(index >= 0 && index < hp_list_.size());
return hp_list_[index]->ptr.load(std::memory_order_relaxed);
}
// If ptr is hazard then reclaim it later.
void ReclaimLater(void* const ptr, std::function<void(void*)>&& func) {
ReclaimNode* new_node = reclaim_pool_.Pop();
new_node->ptr = ptr;
new_node->delete_func = std::move(func);
reclaim_map_.insert(std::make_pair(ptr, new_node));
}
// Try to reclaim all no hazard pointers.
void ReclaimNoHazardPointer();
protected:
Reclaimer(HazardPointerList& hp_list) : global_hp_list_(hp_list) {}
virtual ~Reclaimer();
private:
// Check if the ptr is hazard.
bool Hazard(void* const ptr);
void TryAcquireHazardPointer();
struct ReclaimNode {
ReclaimNode() : ptr(nullptr), next(nullptr), delete_func(nullptr) {}
~ReclaimNode() {}
void* ptr;
ReclaimNode* next;
std::function<void(void*)> delete_func;
};
// Reuse ReclaimNode
struct ReclaimPool {
ReclaimPool() : head(new ReclaimNode()) {}
~ReclaimPool() {
ReclaimNode* temp;
while (head) {
temp = head;
head = head->next;
delete temp;
}
}
void Push(ReclaimNode* node) {
node->next = head;
head = node;
}
ReclaimNode* Pop() {
if (nullptr == head->next) {
return new ReclaimNode();
}
ReclaimNode* temp = head;
head = head->next;
temp->next = nullptr;
return temp;
}
ReclaimNode* head;
};
std::vector<InternalHazardPointer*> hp_list_;
std::unordered_map<void*, ReclaimNode*> reclaim_map_;
ReclaimPool reclaim_pool_;
HazardPointerList& global_hp_list_;
};
// Use Hazard Pointer in RAII style.
class HazardPointer {
public:
HazardPointer() : reclaimer_(nullptr), index(Reclaimer::HP_INDEX_NULL) {}
HazardPointer(Reclaimer* reclaimer_, void* ptr)
: reclaimer_(reclaimer_), index(reclaimer_->MarkHazard(ptr)) {}
~HazardPointer() { UnMark(); }
HazardPointer(const HazardPointer& other) = delete;
HazardPointer(HazardPointer&& other) {
*this = std::move(other);
}
HazardPointer& operator=(const HazardPointer& other) = delete;
HazardPointer& operator=(HazardPointer&& other) {
reclaimer_ = other.reclaimer_;
index = other.index;
// If move assign is called, we must disable the other's index to avoid
// incorrectly unmark index when other destruct.
other.index = Reclaimer::HP_INDEX_NULL;
return *this;
}
void UnMark() { reclaimer_->UnMarkHazard(index); }
public:
Reclaimer* reclaimer_;
Reclaimer::HPIndex index;
};
+3
View File
@@ -303,6 +303,8 @@
<ClInclude Include="..\..\src\support\windowswrapper.h" />
<ClInclude Include="..\..\src\support\zfile.h" />
<ClInclude Include="..\..\src\support\zip.h" />
<ClInclude Include="..\..\third_party\cq\concurrent_queue.h" />
<ClInclude Include="..\..\third_party\cq\reclaimer.h" />
<ClInclude Include="..\..\third_party\interval_tree.h" />
<ClInclude Include="..\..\third_party\json.hpp" />
<ClInclude Include="..\..\third_party\typestring.hh" />
@@ -318,6 +320,7 @@
<ClCompile Include="..\..\src\support\version.cc" />
<ClCompile Include="..\..\src\support\zfile.cc" />
<ClCompile Include="..\..\src\support\zip.cc" />
<ClCompile Include="..\..\third_party\cq\reclaimer.cc" />
</ItemGroup>
<Import Project="$(VCTargetsPath)\Microsoft.Cpp.targets" />
<ImportGroup Label="ExtensionTargets">
@@ -90,6 +90,12 @@
<ClInclude Include="..\..\src\support\version.h">
<Filter>Header Files</Filter>
</ClInclude>
<ClInclude Include="..\..\third_party\cq\concurrent_queue.h">
<Filter>Header Files</Filter>
</ClInclude>
<ClInclude Include="..\..\third_party\cq\reclaimer.h">
<Filter>Header Files</Filter>
</ClInclude>
</ItemGroup>
<ItemGroup>
<ClCompile Include="..\..\src\support\file.cc">
@@ -122,5 +128,8 @@
<ClCompile Include="..\..\src\support\version-windows.cc">
<Filter>Source Files</Filter>
</ClCompile>
<ClCompile Include="..\..\third_party\cq\reclaimer.cc">
<Filter>Source Files</Filter>
</ClCompile>
</ItemGroup>
</Project>