eden-emu/eden
24
3
Fork 15
Code Issues 138 Pull requests 55 Projects 1 Releases Packages Wiki Activity Actions

[video_core] reduce SPSC/MPSC queue contention for commands
All checks were successful
eden-license / license-header (pull_request) Successful in 39s
Details

Browse source 
Signed-off-by: lizzie <lizzie@eden-emu.dev>
This commit is contained in:
lizzie 2025-08-30 08:22:03 +00:00 • committed by crueter
parent 13ecc1e481
commit d3dbe43553
1 changed files with 44 additions and 52 deletions
Download patch file Download diff file Expand all files Collapse all files

96
src/common/bounded_threadsafe_queue.h
View file

@ -1,3 +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
@ -23,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;
@ -71,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};
Common::CondvarWait(consumer_cv, lock, stop_token, [this, read_index] {
return read_index != m_write_index.load(std::memory_order::acquire);
std::unique_lock lock{consumer.cv_mutex};
Common::CondvarWait(consumer.cv, lock, stop_token, [this, read_index] {
return read_index != producer.index.load(std::memory_order::acquire);
});
if (stop_token.stop_requested()) {
return false;
@ -132,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>
@ -242,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