eden-emu
/
eden
mirror of https://git.eden-emu.dev/eden-emu/eden.git
1
0
Fork 0
Code Releases Wiki Activity
Browse Source

[video_core] reduce SPSC/MPSC queue contention for commands (#289) 

Signed-off-by: lizzie <lizzie@eden-emu.dev>
Reviewed-on: https://git.eden-emu.dev/eden-emu/eden/pulls/289
Reviewed-by: Caio Oliveira <caiooliveirafarias0@gmail.com>
Reviewed-by: crueter <crueter@eden-emu.dev>
Co-authored-by: lizzie <lizzie@eden-emu.dev>
Co-committed-by: lizzie <lizzie@eden-emu.dev>
pull/2500/head^2
lizzie 1 month ago
committed by crueter
parent
7d239df065
commit
9807fa02ed
No known key found for this signature in database GPG Key ID: 425ACD2D4830EBC6
1 changed files with 42 additions and 53 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 95
      src/common/bounded_threadsafe_queue.h

95
src/common/bounded_threadsafe_queue.h
View File

@ -1,6 +1,5 @@
// SPDX-FileCopyrightText: Copyright 2025 Eden Emulator Project
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: Copyright 2023 yuzu Emulator Project
// SPDX-License-Identifier: GPL-2.0-or-later
@ -26,34 +25,34 @@ class SPSCQueue {
public:
template <typename... Args>
bool TryEmplace(Args&&... args) {
bool TryEmplace(Args&&... args) noexcept {
return Emplace<PushMode::Try>(std::forward<Args>(args)...);
}
template <typename... Args>
void EmplaceWait(Args&&... args) {
void EmplaceWait(Args&&... args) noexcept {
Emplace<PushMode::Wait>(std::forward<Args>(args)...);
}
bool TryPop(T& t) {
bool TryPop(T& t) noexcept {
return Pop<PopMode::Try>(t);
}
void PopWait(T& t) {
void PopWait(T& t) noexcept {
Pop<PopMode::Wait>(t);
}
void PopWait(T& t, std::stop_token stop_token) {
void PopWait(T& t, const std::stop_token stop_token) noexcept {
Pop<PopMode::WaitWithStopToken>(t, stop_token);
}
T PopWait() {
T PopWait() noexcept {
T t{};
Pop<PopMode::Wait>(t);
return t;
}
T PopWait(std::stop_token stop_token) {
T PopWait(const std::stop_token stop_token) noexcept {
T t{};
Pop<PopMode::WaitWithStopToken>(t, stop_token);
return t;
@ -74,60 +73,53 @@ private:
};
template <PushMode Mode, typename... Args>
bool Emplace(Args&&... args) {
const size_t write_index = m_write_index.load(std::memory_order::relaxed);
bool Emplace(Args&&... args) noexcept {
const std::size_t write_index = producer.index.load(std::memory_order::relaxed);
if constexpr (Mode == PushMode::Try) {
// Check if we have free slots to write to.
if ((write_index - m_read_index.load(std::memory_order::acquire)) == Capacity) {
if ((write_index - consumer.index.load(std::memory_order::acquire)) == Capacity) {
return false;
}
} else if constexpr (Mode == PushMode::Wait) {
// Wait until we have free slots to write to.
std::unique_lock lock{producer_cv_mutex};
producer_cv.wait(lock, [this, write_index] {
return (write_index - m_read_index.load(std::memory_order::acquire)) < Capacity;
std::unique_lock lock{producer.cv_mutex};
producer.cv.wait(lock, [this, write_index] {
return (write_index - consumer.index.load(std::memory_order::acquire)) < Capacity;
});
} else {
static_assert(Mode < PushMode::Count, "Invalid PushMode.");
}
// Determine the position to write to.
const size_t pos = write_index % Capacity;
const std::size_t pos = write_index % Capacity;
// Emplace into the queue.
std::construct_at(std::addressof(m_data[pos]), std::forward<Args>(args)...);
// Increment the write index.
++m_write_index;
++producer.index;
// Notify the consumer that we have pushed into the queue.
std::scoped_lock lock{consumer_cv_mutex};
consumer_cv.notify_one();
std::scoped_lock lock{consumer.cv_mutex};
consumer.cv.notify_one();
return true;
}
template <PopMode Mode>
bool Pop(T& t, [[maybe_unused]] std::stop_token stop_token = {}) {
const size_t read_index = m_read_index.load(std::memory_order::relaxed);
bool Pop(T& t, [[maybe_unused]] std::stop_token stop_token = {}) noexcept {
const std::size_t read_index = consumer.index.load(std::memory_order::relaxed);
if constexpr (Mode == PopMode::Try) {
// Check if the queue is empty.
if (read_index == m_write_index.load(std::memory_order::acquire)) {
if (read_index == producer.index.load(std::memory_order::acquire)) {
return false;
}
} else if constexpr (Mode == PopMode::Wait) {
// Wait until the queue is not empty.
std::unique_lock lock{consumer_cv_mutex};
consumer_cv.wait(lock, [this, read_index] {
return read_index != m_write_index.load(std::memory_order::acquire);
std::unique_lock lock{consumer.cv_mutex};
consumer.cv.wait(lock, [this, read_index] {
return read_index != producer.index.load(std::memory_order::acquire);
});
} else if constexpr (Mode == PopMode::WaitWithStopToken) {
// Wait until the queue is not empty.
std::unique_lock lock{consumer_cv_mutex};
consumer_cv.wait(lock, stop_token, [this, read_index] {
return read_index != m_write_index.load(std::memory_order::acquire);
std::unique_lock lock{consumer.cv_mutex};
consumer.cv.wait(lock, stop_token, [this, read_index] {
return read_index != producer.index.load(std::memory_order::acquire);
});
if (stop_token.stop_requested()) {
return false;
@ -135,32 +127,29 @@ private:
} else {
static_assert(Mode < PopMode::Count, "Invalid PopMode.");
}
// Determine the position to read from.
const size_t pos = read_index % Capacity;
const std::size_t pos = read_index % Capacity;
// Pop the data off the queue, moving it.
t = std::move(m_data[pos]);
// Increment the read index.
++m_read_index;
++consumer.index;
// Notify the producer that we have popped off the queue.
std::scoped_lock lock{producer_cv_mutex};
producer_cv.notify_one();
std::scoped_lock lock{producer.cv_mutex};
producer.cv.notify_one();
return true;
}
alignas(128) std::atomic_size_t m_read_index{0};
alignas(128) std::atomic_size_t m_write_index{0};
std::array<T, Capacity> m_data;
std::condition_variable_any producer_cv;
std::mutex producer_cv_mutex;
std::condition_variable_any consumer_cv;
std::mutex consumer_cv_mutex;
alignas(64) struct {
std::atomic_size_t index{0};
std::condition_variable_any cv;
std::mutex cv_mutex;
} producer;
alignas(64) struct {
std::atomic_size_t index{0};
std::condition_variable_any cv;
std::mutex cv_mutex;
} consumer;
};
template <typename T, size_t Capacity = detail::DefaultCapacity>
@ -245,8 +234,8 @@ public:
private:
SPSCQueue<T, Capacity> spsc_queue;
std::mutex write_mutex;
std::mutex read_mutex;
alignas(64) std::mutex write_mutex;
alignas(64) std::mutex read_mutex;
};
} // namespace Common
Write Preview
Loading…
Cancel
Save