early-access version 3464

README.md

@@ -1,7 +1,7 @@
 yuzu emulator early access
 =============
 
-This is the source code for early-access 3463.
+This is the source code for early-access 3464.
 
 ## Legal Notice
 

src/common/bounded_threadsafe_queue.h

@@ -1,159 +1,323 @@
-// SPDX-FileCopyrightText: Copyright (c) 2020 Erik Rigtorp <erik@rigtorp.se>
+// SPDX-FileCopyrightText: Copyright 2023 yuzu Emulator Project
-// SPDX-License-Identifier: MIT
+// SPDX-License-Identifier: GPL-2.0-or-later
 
 #pragma once
 
 #include <atomic>
-#include <bit>
 #include <condition_variable>
-#include <memory>
+#include <cstddef>
 #include <mutex>
 #include <new>
-#include <type_traits>
+#include <version>
-#include <utility>
 
 #include "common/polyfill_thread.h"
 
 namespace Common {
 
-#if defined(__cpp_lib_hardware_interference_size)
+namespace detail {
-constexpr size_t hardware_interference_size = std::hardware_destructive_interference_size;
+constexpr size_t DefaultCapacity = 0x1000;
-#else
+} // namespace detail
-constexpr size_t hardware_interference_size = 64;
+
-#endif
+template <typename T, size_t Capacity = detail::DefaultCapacity>
+class SPSCQueue {
+    static_assert((Capacity & (Capacity - 1)) == 0, "Capacity must be a power of two.");
 
-template <typename T, size_t capacity = 0x400>
-class MPSCQueue {
 public:
-    explicit MPSCQueue() : allocator{std::allocator<Slot<T>>()} {
+    bool TryPush(T&& t) {
-        // Allocate one extra slot to prevent false sharing on the last slot
+        const size_t write_index = m_write_index.load();
-        slots = allocator.allocate(capacity + 1);
+
-        // Allocators are not required to honor alignment for over-aligned types
+        // Check if we have free slots to write to.
-        // (see http://eel.is/c++draft/allocator.requirements#10) so we verify
+        if ((write_index - m_read_index.load()) == Capacity) {
-        // alignment here
+            return false;
-        if (reinterpret_cast<uintptr_t>(slots) % alignof(Slot<T>) != 0) {
-            allocator.deallocate(slots, capacity + 1);
-            throw std::bad_alloc();
-        }
-        for (size_t i = 0; i < capacity; ++i) {
-            std::construct_at(&slots[i]);
-        }
-        static_assert(std::has_single_bit(capacity), "capacity must be an integer power of 2");
-        static_assert(alignof(Slot<T>) == hardware_interference_size,
-                      "Slot must be aligned to cache line boundary to prevent false sharing");
-        static_assert(sizeof(Slot<T>) % hardware_interference_size == 0,
-                      "Slot size must be a multiple of cache line size to prevent "
-                      "false sharing between adjacent slots");
-        static_assert(sizeof(MPSCQueue) % hardware_interference_size == 0,
-                      "Queue size must be a multiple of cache line size to "
-                      "prevent false sharing between adjacent queues");
         }
 
-    ~MPSCQueue() noexcept {
+        // Determine the position to write to.
-        for (size_t i = 0; i < capacity; ++i) {
+        const size_t pos = write_index % Capacity;
-            std::destroy_at(&slots[i]);
-        }
-        allocator.deallocate(slots, capacity + 1);
-    }
 
-    // The queue must be both non-copyable and non-movable
+        // Push into the queue.
-    MPSCQueue(const MPSCQueue&) = delete;
+        m_data[pos] = std::move(t);
-    MPSCQueue& operator=(const MPSCQueue&) = delete;
 
-    MPSCQueue(MPSCQueue&&) = delete;
+        // Increment the write index.
-    MPSCQueue& operator=(MPSCQueue&&) = delete;
+        ++m_write_index;
 
-    void Push(const T& v) noexcept {
+        // Notify the consumer that we have pushed into the queue.
-        static_assert(std::is_nothrow_copy_constructible_v<T>,
+        std::scoped_lock lock{cv_mutex};
-                      "T must be nothrow copy constructible");
+        cv.notify_one();
-        emplace(v);
-    }
 
-    template <typename P, typename = std::enable_if_t<std::is_nothrow_constructible_v<T, P&&>>>
+        return true;
-    void Push(P&& v) noexcept {
-        emplace(std::forward<P>(v));
-    }
-
-    void Pop(T& v, std::stop_token stop) noexcept {
-        auto const tail = tail_.fetch_add(1);
-        auto& slot = slots[idx(tail)];
-        if (!slot.turn.test()) {
-            std::unique_lock lock{cv_mutex};
-            Common::CondvarWait(cv, lock, stop, [&slot] { return slot.turn.test(); });
-        }
-        v = slot.move();
-        slot.destroy();
-        slot.turn.clear();
-        slot.turn.notify_one();
-    }
-
-private:
-    template <typename U = T>
-    struct Slot {
-        ~Slot() noexcept {
-            if (turn.test()) {
-                destroy();
-            }
     }
 
     template <typename... Args>
-        void construct(Args&&... args) noexcept {
+    bool TryPush(Args&&... args) {
-            static_assert(std::is_nothrow_constructible_v<U, Args&&...>,
+        const size_t write_index = m_write_index.load();
-                          "T must be nothrow constructible with Args&&...");
+
-            std::construct_at(reinterpret_cast<U*>(&storage), std::forward<Args>(args)...);
+        // Check if we have free slots to write to.
+        if ((write_index - m_read_index.load()) == Capacity) {
+            return false;
         }
 
-        void destroy() noexcept {
+        // Determine the position to write to.
-            static_assert(std::is_nothrow_destructible_v<U>, "T must be nothrow destructible");
+        const size_t pos = write_index % Capacity;
-            std::destroy_at(reinterpret_cast<U*>(&storage));
+
+        // Emplace into the queue.
+        std::construct_at(std::addressof(m_data[pos]), std::forward<Args>(args)...);
+
+        // Increment the write index.
+        ++m_write_index;
+
+        // Notify the consumer that we have pushed into the queue.
+        std::scoped_lock lock{cv_mutex};
+        cv.notify_one();
+
+        return true;
     }
 
-        U&& move() noexcept {
+    void Push(T&& t) {
-            return reinterpret_cast<U&&>(storage);
+        const size_t write_index = m_write_index.load();
+
+        // Wait until we have free slots to write to.
+        while ((write_index - m_read_index.load()) == Capacity) {
+            std::this_thread::yield();
         }
 
-        // Align to avoid false sharing between adjacent slots
+        // Determine the position to write to.
-        alignas(hardware_interference_size) std::atomic_flag turn{};
+        const size_t pos = write_index % Capacity;
-        struct aligned_store {
-            struct type {
-                alignas(U) unsigned char data[sizeof(U)];
-            };
-        };
-        typename aligned_store::type storage;
-    };
 
-    template <typename... Args>
+        // Push into the queue.
-    void emplace(Args&&... args) noexcept {
+        m_data[pos] = std::move(t);
-        static_assert(std::is_nothrow_constructible_v<T, Args&&...>,
+
-                      "T must be nothrow constructible with Args&&...");
+        // Increment the write index.
-        auto const head = head_.fetch_add(1);
+        ++m_write_index;
-        auto& slot = slots[idx(head)];
+
-        slot.turn.wait(true);
+        // Notify the consumer that we have pushed into the queue.
-        slot.construct(std::forward<Args>(args)...);
+        std::scoped_lock lock{cv_mutex};
-        slot.turn.test_and_set();
         cv.notify_one();
     }
 
-    constexpr size_t idx(size_t i) const noexcept {
+    template <typename... Args>
-        return i & mask;
+    void Push(Args&&... args) {
+        const size_t write_index = m_write_index.load();
+
+        // Wait until we have free slots to write to.
+        while ((write_index - m_read_index.load()) == Capacity) {
+            std::this_thread::yield();
         }
 
-    static constexpr size_t mask = capacity - 1;
+        // Determine the position to write to.
+        const size_t pos = write_index % Capacity;
 
-    // Align to avoid false sharing between head_ and tail_
+        // Emplace into the queue.
-    alignas(hardware_interference_size) std::atomic<size_t> head_{0};
+        std::construct_at(std::addressof(m_data[pos]), std::forward<Args>(args)...);
-    alignas(hardware_interference_size) std::atomic<size_t> tail_{0};
+
+        // Increment the write index.
+        ++m_write_index;
+
+        // Notify the consumer that we have pushed into the queue.
+        std::scoped_lock lock{cv_mutex};
+        cv.notify_one();
+    }
+
+    bool TryPop(T& t) {
+        return Pop(t);
+    }
+
+    void PopWait(T& t, std::stop_token stop_token) {
+        Wait(stop_token);
+        Pop(t);
+    }
+
+    T PopWait(std::stop_token stop_token) {
+        Wait(stop_token);
+        T t;
+        Pop(t);
+        return t;
+    }
+
+    void Clear() {
+        while (!Empty()) {
+            Pop();
+        }
+    }
+
+    bool Empty() const {
+        return m_read_index.load() == m_write_index.load();
+    }
+
+    size_t Size() const {
+        return m_write_index.load() - m_read_index.load();
+    }
+
+private:
+    void Pop() {
+        const size_t read_index = m_read_index.load();
+
+        // Check if the queue is empty.
+        if (read_index == m_write_index.load()) {
+            return;
+        }
+
+        // Determine the position to read from.
+        const size_t pos = read_index % Capacity;
+
+        // Pop the data off the queue, deleting it.
+        std::destroy_at(std::addressof(m_data[pos]));
+
+        // Increment the read index.
+        ++m_read_index;
+    }
+
+    bool Pop(T& t) {
+        const size_t read_index = m_read_index.load();
+
+        // Check if the queue is empty.
+        if (read_index == m_write_index.load()) {
+            return false;
+        }
+
+        // Determine the position to read from.
+        const 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;
+
+        return true;
+    }
+
+    void Wait(std::stop_token stop_token) {
+        std::unique_lock lock{cv_mutex};
+        Common::CondvarWait(cv, lock, stop_token, [this] { return !Empty(); });
+    }
+
+#ifdef __cpp_lib_hardware_interference_size
+    alignas(std::hardware_destructive_interference_size) std::atomic_size_t m_read_index{0};
+    alignas(std::hardware_destructive_interference_size) std::atomic_size_t m_write_index{0};
+#else
+    alignas(64) std::atomic_size_t m_read_index{0};
+    alignas(64) std::atomic_size_t m_write_index{0};
+#endif
+
+    std::array<T, Capacity> m_data;
 
-    std::mutex cv_mutex;
     std::condition_variable_any cv;
+    std::mutex cv_mutex;
+};
 
-    Slot<T>* slots;
+template <typename T, size_t Capacity = detail::DefaultCapacity>
-    [[no_unique_address]] std::allocator<Slot<T>> allocator;
+class MPSCQueue {
+public:
+    void TryPush(T&& t) {
+        std::scoped_lock lock{write_mutex};
+        spsc_queue.TryPush(std::move(t));
+    }
 
-    static_assert(std::is_nothrow_copy_assignable_v<T> || std::is_nothrow_move_assignable_v<T>,
+    template <typename... Args>
-                  "T must be nothrow copy or move assignable");
+    void TryPush(Args&&... args) {
+        std::scoped_lock lock{write_mutex};
+        spsc_queue.TryPush(std::forward<Args>(args)...);
+    }
 
-    static_assert(std::is_nothrow_destructible_v<T>, "T must be nothrow destructible");
+    void Push(T&& t) {
+        std::scoped_lock lock{write_mutex};
+        spsc_queue.Push(std::move(t));
+    }
+
+    template <typename... Args>
+    void Push(Args&&... args) {
+        std::scoped_lock lock{write_mutex};
+        spsc_queue.Push(std::forward<Args>(args)...);
+    }
+
+    bool TryPop(T& t) {
+        return spsc_queue.TryPop(t);
+    }
+
+    void PopWait(T& t, std::stop_token stop_token) {
+        spsc_queue.PopWait(t, stop_token);
+    }
+
+    T PopWait(std::stop_token stop_token) {
+        return spsc_queue.PopWait(stop_token);
+    }
+
+    void Clear() {
+        spsc_queue.Clear();
+    }
+
+    bool Empty() {
+        return spsc_queue.Empty();
+    }
+
+    size_t Size() {
+        return spsc_queue.Size();
+    }
+
+private:
+    SPSCQueue<T, Capacity> spsc_queue;
+    std::mutex write_mutex;
+};
+
+template <typename T, size_t Capacity = detail::DefaultCapacity>
+class MPMCQueue {
+public:
+    void TryPush(T&& t) {
+        std::scoped_lock lock{write_mutex};
+        spsc_queue.TryPush(std::move(t));
+    }
+
+    template <typename... Args>
+    void TryPush(Args&&... args) {
+        std::scoped_lock lock{write_mutex};
+        spsc_queue.TryPush(std::forward<Args>(args)...);
+    }
+
+    void Push(T&& t) {
+        std::scoped_lock lock{write_mutex};
+        spsc_queue.Push(std::move(t));
+    }
+
+    template <typename... Args>
+    void Push(Args&&... args) {
+        std::scoped_lock lock{write_mutex};
+        spsc_queue.Push(std::forward<Args>(args)...);
+    }
+
+    bool TryPop(T& t) {
+        std::scoped_lock lock{read_mutex};
+        return spsc_queue.TryPop(t);
+    }
+
+    void PopWait(T& t, std::stop_token stop_token) {
+        std::scoped_lock lock{read_mutex};
+        spsc_queue.PopWait(t, stop_token);
+    }
+
+    T PopWait(std::stop_token stop_token) {
+        std::scoped_lock lock{read_mutex};
+        return spsc_queue.PopWait(stop_token);
+    }
+
+    void Clear() {
+        std::scoped_lock lock{read_mutex};
+        spsc_queue.Clear();
+    }
+
+    bool Empty() {
+        std::scoped_lock lock{read_mutex};
+        return spsc_queue.Empty();
+    }
+
+    size_t Size() {
+        std::scoped_lock lock{read_mutex};
+        return spsc_queue.Size();
+    }
+
+private:
+    SPSCQueue<T, Capacity> spsc_queue;
+    std::mutex write_mutex;
+    std::mutex read_mutex;
 };
 
 } // namespace Common

src/common/logging/backend.cpp

@@ -28,7 +28,7 @@
 #ifdef _WIN32
 #include "common/string_util.h"
 #endif
-#include "common/threadsafe_queue.h"
+#include "common/bounded_threadsafe_queue.h"
 
 namespace Common::Log {
 
@@ -204,11 +204,11 @@ public:
 
     void PushEntry(Class log_class, Level log_level, const char* filename, unsigned int line_num,
                    const char* function, std::string&& message) {
-        if (!filter.CheckMessage(log_class, log_level))
+        if (!filter.CheckMessage(log_class, log_level)) {
             return;
-        const Entry& entry =
+        }
-            CreateEntry(log_class, log_level, filename, line_num, function, std::move(message));
+        message_queue.Push(
-        message_queue.Push(entry);
+            CreateEntry(log_class, log_level, filename, line_num, function, std::move(message)));
     }
 
 private:
@@ -225,7 +225,7 @@ private:
                 ForEachBackend([&entry](Backend& backend) { backend.Write(entry); });
             };
             while (!stop_token.stop_requested()) {
-                entry = message_queue.PopWait(stop_token);
+                message_queue.PopWait(entry, stop_token);
                 if (entry.filename != nullptr) {
                     write_logs();
                 }
@@ -233,7 +233,7 @@ private:
             // Drain the logging queue. Only writes out up to MAX_LOGS_TO_WRITE to prevent a
             // case where a system is repeatedly spamming logs even on close.
             int max_logs_to_write = filter.IsDebug() ? INT_MAX : 100;
-            while (max_logs_to_write-- && message_queue.Pop(entry)) {
+            while (max_logs_to_write-- && message_queue.TryPop(entry)) {
                 write_logs();
             }
         });
@@ -273,7 +273,7 @@ private:
     ColorConsoleBackend color_console_backend{};
     FileBackend file_backend;
 
-    MPSCQueue<Entry, true> message_queue{};
+    MPSCQueue<Entry> message_queue{};
     std::chrono::steady_clock::time_point time_origin{std::chrono::steady_clock::now()};
     std::jthread backend_thread;
 };

src/video_core/gpu_thread.cpp

@@ -31,9 +31,10 @@ static void RunThread(std::stop_token stop_token, Core::System& system,
     auto current_context = context.Acquire();
     VideoCore::RasterizerInterface* const rasterizer = renderer.ReadRasterizer();
 
-    while (!stop_token.stop_requested()) {
     CommandDataContainer next;
-        state.queue.Pop(next, stop_token);
+
+    while (!stop_token.stop_requested()) {
+        state.queue.PopWait(next, stop_token);
         if (stop_token.stop_requested()) {
             break;
         }
@@ -58,6 +59,9 @@ static void RunThread(std::stop_token stop_token, Core::System& system,
             state.cv.notify_all();
         }
     }
+
+    // Drain the queue.
+    state.queue.Clear();
 }
 
 ThreadManager::ThreadManager(Core::System& system_, bool is_async_)
@@ -117,7 +121,7 @@ u64 ThreadManager::PushCommand(CommandData&& command_data, bool block) {
 
     std::unique_lock lk(state.write_lock);
     const u64 fence{++state.last_fence};
-    state.queue.Push(CommandDataContainer(std::move(command_data), fence, block));
+    state.queue.Push(std::move(command_data), fence, block);
 
     if (block) {
         Common::CondvarWait(state.cv, lk, thread.get_stop_token(), [this, fence] {