diff --git a/Source/Core/Common/SPSCQueue.h b/Source/Core/Common/SPSCQueue.h deleted file mode 100644 index ca1a33555e..0000000000 --- a/Source/Core/Common/SPSCQueue.h +++ /dev/null @@ -1,155 +0,0 @@ -// Copyright 2010 Dolphin Emulator Project -// SPDX-License-Identifier: GPL-2.0-or-later - -#pragma once - -// a simple lockless thread-safe, -// single producer, single consumer queue - -#include -#include - -#if defined(__APPLE__) -#include -#if (defined(__TV_OS_VERSION_MIN_REQUIRED) && __TV_OS_VERSION_MIN_REQUIRED < 140000) || \ - (defined(__IPHONE_OS_VERSION_MIN_REQUIRED) && __IPHONE_OS_VERSION_MIN_REQUIRED < 140000) -#define DOLPHIN_NEEDS_ATOMIC_WAIT_FALLBACK 1 -#include -#include -#endif -#endif - -#include "Common/TypeUtils.h" - -namespace Common -{ - -namespace detail -{ -template -class SPSCQueueBase final -{ -public: - SPSCQueueBase() = default; - ~SPSCQueueBase() - { - Clear(); - delete m_read_ptr; - } - - SPSCQueueBase(const SPSCQueueBase&) = delete; - SPSCQueueBase& operator=(const SPSCQueueBase&) = delete; - - std::size_t Size() const { return m_size.load(std::memory_order_acquire); } - bool Empty() const { return Size() == 0; } - - // The following are only safe from the "producer thread": - void Push(const T& arg) { Emplace(arg); } - void Push(T&& arg) { Emplace(std::move(arg)); } - template - void Emplace(Args&&... args) - { - m_write_ptr->value.Construct(std::forward(args)...); - - Node* const new_ptr = new Node; - m_write_ptr->next = new_ptr; - m_write_ptr = new_ptr; - - AdjustSize(1); - } - - void WaitForEmpty() requires(IncludeWaitFunctionality) - { -#ifdef DOLPHIN_NEEDS_ATOMIC_WAIT_FALLBACK - std::unique_lock lock(m_wait_mutex); - while (Size() > 0) - m_wait_cv.wait(lock); -#else - while (const std::size_t old_size = Size()) - m_size.wait(old_size, std::memory_order_acquire); -#endif - } - - // The following are only safe from the "consumer thread": - T& Front() { return m_read_ptr->value.Ref(); } - const T& Front() const { return m_read_ptr->value.Ref(); } - - void Pop() - { - assert(!Empty()); - - m_read_ptr->value.Destroy(); - - Node* const old_node = m_read_ptr; - m_read_ptr = old_node->next; - delete old_node; - - AdjustSize(-1); - } - - bool Pop(T& result) - { - if (Empty()) - return false; - - result = std::move(Front()); - Pop(); - return true; - } - - void WaitForData() requires(IncludeWaitFunctionality) - { -#ifdef DOLPHIN_NEEDS_ATOMIC_WAIT_FALLBACK - std::unique_lock lock(m_wait_mutex); - while (Size() == 0) - m_wait_cv.wait(lock); -#else - m_size.wait(0, std::memory_order_acquire); -#endif - } - - void Clear() - { - while (!Empty()) - Pop(); - } - -private: - struct Node - { - ManuallyConstructedValue value; - Node* next; - }; - - Node* m_write_ptr = new Node; - Node* m_read_ptr = m_write_ptr; - - void AdjustSize(std::size_t value) - { - m_size.fetch_add(value, std::memory_order_release); - if constexpr (IncludeWaitFunctionality) - { -#ifdef DOLPHIN_NEEDS_ATOMIC_WAIT_FALLBACK - m_wait_cv.notify_one(); -#else - m_size.notify_one(); -#endif - } - } - - std::atomic m_size = 0; - -#ifdef DOLPHIN_NEEDS_ATOMIC_WAIT_FALLBACK - std::mutex m_wait_mutex; - std::condition_variable m_wait_cv; -#endif -}; -} // namespace detail - -template -using SPSCQueue = detail::SPSCQueueBase; - -template -using WaitableSPSCQueue = detail::SPSCQueueBase; - -} // namespace Common diff --git a/Source/Core/Common/WorkQueueThread.h b/Source/Core/Common/WorkQueueThread.h index 826ee778be..0ded21fbde 100644 --- a/Source/Core/Common/WorkQueueThread.h +++ b/Source/Core/Common/WorkQueueThread.h @@ -4,49 +4,73 @@ #pragma once #include -#include -#include #include -#include +#include -#include "Common/Event.h" -#include "Common/SPSCQueue.h" -#include "Common/Thread.h" +#include +#include +#include +#include namespace Common { namespace detail { +// A worker thread fed by a lock-free queue. Producers push onto an mpsc_stack, so any number of +// threads may push without a lock; the worker takes the whole stack at once and runs it in push +// order. Items are moved into their queue node on push and moved out of it once, by the worker. +// +// The control calls (Reset, Cancel, Shutdown, Stop, WaitForCompletion) are made from one thread. template class WorkQueueThreadBase final { public: using FunctionType = std::function; - WorkQueueThreadBase() = default; - WorkQueueThreadBase(std::string name, FunctionType function) + WorkQueueThreadBase() + { + mpsc_stack_init(&m_items); + retro_atomic_size_init(&m_pushed, 0); + retro_atomic_size_init(&m_done, 0); + retro_atomic_size_init(&m_cancel_to, 0); + retro_atomic_int_init(&m_stop, 0); + retro_eventcount_init(&m_wake); + retro_eventcount_init(&m_progress); + } + WorkQueueThreadBase(std::string name, FunctionType function) : WorkQueueThreadBase() { Reset(std::move(name), std::move(function)); } - ~WorkQueueThreadBase() { Shutdown(); } - - // Shuts the current work thread down (if any) and starts a new thread with the given function - // Note: Some consumers of this API push items to the queue before starting the thread. - void Reset(std::string name, FunctionType function) + ~WorkQueueThreadBase() { - auto lg = GetLockGuard(); Shutdown(); - m_thread = std::thread(std::bind_front(&WorkQueueThreadBase::ThreadLoop, this), std::move(name), - std::move(function)); + FreeChain(m_held); + FreeChain(reinterpret_cast(mpsc_stack_reverse(mpsc_stack_drain(&m_items)))); + retro_eventcount_free(&m_progress); + retro_eventcount_free(&m_wake); } - // Adds an item to the work queue + WorkQueueThreadBase(const WorkQueueThreadBase&) = delete; + WorkQueueThreadBase& operator=(const WorkQueueThreadBase&) = delete; + + // Shuts the current work thread down (if any) and starts a new thread with the given function. + // Items pushed before the thread starts are kept and run by it. + void Reset(std::string name, FunctionType function) + { + Shutdown(); + m_name = std::move(name); + m_function = std::move(function); + m_thread = sthread_create(ThreadEntry, this); + } + + // Adds an item to the work queue. Never blocks. template void EmplaceItem(Args&&... args) { - auto lg = GetLockGuard(); - m_items.Emplace(std::forward(args)...); - m_event.Set(); + Node* const node = new Node{{nullptr}, T(std::forward(args)...)}; + mpsc_stack_push(&m_items, &node->link); + retro_atomic_fetch_add_size(&m_pushed, 1); + retro_eventcount_notify(&m_wake); } void Push(T&& item) { EmplaceItem(std::move(item)); } void Push(const T& item) { EmplaceItem(item); } @@ -55,36 +79,46 @@ public: // Blocks until the current work is cancelled. void Cancel() { - auto lg = GetLockGuard(); - - // Fast path avoids round trip thread communication and saves ~20us. - if (m_items.Empty()) + const std::size_t pushed = retro_atomic_load_acquire_size(&m_pushed); + if (retro_atomic_load_acquire_size(&m_done) == pushed) return; - RunCommand([&] { m_items.Clear(); }); + if (!IsRunning()) + { + // No worker: the queue is ours. + Node* chain = TakeAll(); + std::size_t skipped = 0; + while (chain) + { + Node* const next = reinterpret_cast(chain->link.next); + delete chain; + chain = next; + ++skipped; + } + retro_atomic_fetch_add_size(&m_done, skipped); + return; + } + + retro_atomic_store_release_size(&m_cancel_to, pushed); + retro_eventcount_notify(&m_wake); + WaitForDone(pushed); } // Tells the worker thread to stop when its queue is empty. // Blocks until the worker thread exits. Does nothing if thread isn't running. void Shutdown() { - auto lg = GetLockGuard(); WaitForCompletion(); StopThread(); } // Tells the worker thread to stop immediately, potentially leaving work in the queue. // Blocks until the worker thread exits. Does nothing if thread isn't running. - void Stop() - { - auto lg = GetLockGuard(); - StopThread(); - } + void Stop() { StopThread(); } // Stops the worker thread ASAP and empties the queue. void StopAndCancel() { - auto lg = GetLockGuard(); Stop(); Cancel(); } @@ -93,93 +127,126 @@ public: // Does nothing if thread isn't running. void WaitForCompletion() { - auto lg = GetLockGuard(); if (IsRunning()) - m_items.WaitForEmpty(); + WaitForDone(retro_atomic_load_acquire_size(&m_pushed)); } - bool IsRunning() { return m_thread.joinable(); } + bool IsRunning() const { return m_thread != nullptr; } private: - using CommandFunction = std::function; - - // Blocking. - void RunCommand(CommandFunction cmd) + struct Node { - if (!IsRunning()) - { - std::invoke(cmd); - return; - } + mpsc_stack_node_t link; + T value; + }; - m_commands.Emplace(std::move(cmd)); - m_event.Set(); - m_commands.WaitForEmpty(); + static void FreeChain(Node* chain) + { + while (chain) + { + Node* const next = reinterpret_cast(chain->link.next); + delete chain; + chain = next; + } + } + + // Everything queued, oldest first: what a stopped worker held back, then the stack. + Node* TakeAll() + { + Node* chain = reinterpret_cast(mpsc_stack_reverse(mpsc_stack_drain(&m_items))); + if (!m_held) + return chain; + Node* tail = m_held; + while (tail->link.next) + tail = reinterpret_cast(tail->link.next); + tail->link.next = chain ? &chain->link : nullptr; + chain = m_held; + m_held = nullptr; + return chain; + } + + void WaitForDone(std::size_t target) + { + while (retro_atomic_load_acquire_size(&m_done) < target) + { + const int key = retro_eventcount_prepare_wait(&m_progress); + if (retro_atomic_load_acquire_size(&m_done) >= target) + { + retro_eventcount_cancel_wait(&m_progress); + break; + } + retro_eventcount_commit_wait(&m_progress, key); + } } - // Stop immediately. void StopThread() { - if (!m_thread.joinable()) + if (!m_thread) return; - // empty-function shutdown signal. - m_commands.Emplace(CommandFunction{}); - m_event.Set(); - m_thread.join(); - m_commands.Clear(); + retro_atomic_store_release_int(&m_stop, 1); + retro_eventcount_notify(&m_wake); + sthread_join(m_thread); + m_thread = nullptr; + retro_atomic_store_release_int(&m_stop, 0); } - auto GetLockGuard() + static void ThreadEntry(void* data) { static_cast(data)->ThreadLoop(); } + + void ThreadLoop() { - struct DummyLockGuard + sthread_setname(m_name.c_str()); + + // Items are numbered in the order they run; m_done is the last one finished. + std::size_t seq = retro_atomic_load_relaxed_size(&m_done); + Node* chain = m_held; + m_held = nullptr; + + for (;;) { - // Silences unused variable warning. - ~DummyLockGuard() { void(); } - }; + if (!chain) + chain = reinterpret_cast(mpsc_stack_reverse(mpsc_stack_drain(&m_items))); - if constexpr (IsSingleProducer) - return DummyLockGuard{}; - else - return std::lock_guard{m_mutex}; - } + if (retro_atomic_load_acquire_int(&m_stop)) + break; - void ThreadLoop(const std::string& thread_name, const FunctionType& function) - { - Common::SetCurrentThreadName(thread_name.c_str()); - - while (true) - { - while (!m_commands.Empty()) + if (chain) { - CommandFunction& command = m_commands.Front(); - // empty-function shutdown signal. - if (!command) - return; - - std::invoke(command); - m_commands.Pop(); - } - - if (m_items.Empty()) - { - m_event.Wait(); + Node* const node = chain; + chain = reinterpret_cast(node->link.next); + ++seq; + if (seq > retro_atomic_load_acquire_size(&m_cancel_to)) + m_function(std::move(node->value)); + delete node; + retro_atomic_store_release_size(&m_done, seq); + retro_eventcount_notify(&m_progress); continue; } - function(std::move(m_items.Front())); - m_items.Pop(); + const int key = retro_eventcount_prepare_wait(&m_wake); + if (!mpsc_stack_empty(&m_items) || retro_atomic_load_acquire_int(&m_stop)) + { + retro_eventcount_cancel_wait(&m_wake); + continue; + } + retro_eventcount_commit_wait(&m_wake, key); } + + // Stopped with work left: keep it, in order, for the next thread. + m_held = chain; } - std::thread m_thread; - Common::WaitableSPSCQueue m_items; - Common::WaitableSPSCQueue m_commands; - Common::Event m_event; - - using DummyMutex = std::type_identity; - using ProducerMutex = std::conditional_t; - ProducerMutex m_mutex; + sthread_t* m_thread = nullptr; + Node* m_held = nullptr; // worker-owned while running, ours while stopped + FunctionType m_function; + std::string m_name; + mpsc_stack_t m_items; + retro_eventcount_t m_wake; + retro_eventcount_t m_progress; + retro_atomic_size_t m_pushed; + retro_atomic_size_t m_done; + retro_atomic_size_t m_cancel_to; + retro_atomic_int_t m_stop; }; // A WorkQueueThread-like class that takes functions to invoke. @@ -189,8 +256,16 @@ class AsyncWorkThreadBase public: using FuncType = std::function; - AsyncWorkThreadBase() = default; - explicit AsyncWorkThreadBase(std::string thread_name) { Reset(std::move(thread_name)); } + AsyncWorkThreadBase() { retro_eventcount_init(&m_ran); } + explicit AsyncWorkThreadBase(std::string thread_name) : AsyncWorkThreadBase() + { + Reset(std::move(thread_name)); + } + ~AsyncWorkThreadBase() + { + m_worker.Shutdown(); + retro_eventcount_free(&m_ran); + } void Reset(std::string thread_name) { @@ -199,11 +274,28 @@ public: void Push(FuncType func) { m_worker.Push(std::move(func)); } - auto PushBlocking(FuncType func) + // Runs func on the worker and returns once it has run. + void PushBlocking(FuncType func) { - std::packaged_task task{std::move(func)}; - m_worker.EmplaceItem([&] { task(); }); - return task.get_future().get(); + // The flag lives on this stack; the eventcount outlives the call, since the worker may + // still be inside the notify when this returns. + retro_atomic_int_t done; + retro_atomic_int_init(&done, 0); + m_worker.EmplaceItem([&func, &done, this] { + func(); + retro_atomic_store_release_int(&done, 1); + retro_eventcount_notify(&m_ran); + }); + while (!retro_atomic_load_acquire_int(&done)) + { + const int key = retro_eventcount_prepare_wait(&m_ran); + if (retro_atomic_load_acquire_int(&done)) + { + retro_eventcount_cancel_wait(&m_ran); + break; + } + retro_eventcount_commit_wait(&m_ran, key); + } } void Cancel() { m_worker.Cancel(); } @@ -212,15 +304,15 @@ public: private: WorkThread m_worker; + retro_eventcount_t m_ran; }; } // namespace detail -// Multiple threads may use the public interface. +// Multiple threads may push. template using WorkQueueThread = detail::WorkQueueThreadBase; -// A "Single Producer" WorkQueueThread. -// It uses no mutex but only one thread can safely manipulate the queue. +// A "Single Producer" WorkQueueThread: only one thread pushes. template using WorkQueueThreadSP = detail::WorkQueueThreadBase; diff --git a/Source/Core/Core/Core.cpp b/Source/Core/Core/Core.cpp index 0e2c52b2f5..3de85299fb 100644 --- a/Source/Core/Core/Core.cpp +++ b/Source/Core/Core/Core.cpp @@ -967,21 +967,9 @@ void Callback_NewField(Core::System& system) { if (s_frame_step) { -#ifndef __LIBRETRO__ - // To ensure that s_stop_frame_step is up to date, wait for the GPU thread queue to empty, - // since it is may contain a swap event (which will call Callback_FramePresented). This hurts - // the performance a little, but luckily, performance matters less when using frame stepping. - AsyncRequests::GetInstance()->WaitForEmptyQueue(); - - // Only stop the frame stepping if a new frame was displayed - // (as opposed to the previous frame being displayed for another frame). - if (s_stop_frame_step.load()) -#endif - { - s_frame_step = false; - system.GetCPU().Break(); - NotifyStateChanged(Core::GetState(system)); - } + s_frame_step = false; + system.GetCPU().Break(); + NotifyStateChanged(Core::GetState(system)); } AchievementManager::GetInstance().DoFrame(); diff --git a/Source/Core/Core/CoreTiming.cpp b/Source/Core/Core/CoreTiming.cpp index b99a202ca7..ddb7a26732 100644 --- a/Source/Core/Core/CoreTiming.cpp +++ b/Source/Core/Core/CoreTiming.cpp @@ -4,7 +4,6 @@ #include "Core/CoreTiming.h" #include -#include #include #include #include @@ -14,7 +13,6 @@ #include "Common/Assert.h" #include "Common/ChunkFile.h" #include "Common/Logging/Log.h" -#include "Common/SPSCQueue.h" #include "Common/ScopeGuard.h" #include "Core/AchievementManager.h" @@ -42,6 +40,7 @@ static void EmptyTimedCallback(Core::System& system, u64 userdata, s64 cyclesLat CoreTimingManager::CoreTimingManager(Core::System& system) : m_system(system) { + mpsc_stack_init(&m_ts_queue); } // Changing the CPU speed in Dolphin isn't actually done by changing the physical clock rate, @@ -122,11 +121,21 @@ void CoreTimingManager::Init() }); } +CoreTimingManager::~CoreTimingManager() +{ + mpsc_stack_node_t* node = mpsc_stack_drain(&m_ts_queue); + while (node) + { + mpsc_stack_node_t* const next = node->next; + delete reinterpret_cast(node); + node = next; + } +} + void CoreTimingManager::Shutdown() { m_core_state_changed_hook.reset(); - std::lock_guard lk(m_ts_write_lock); MoveEvents(); ClearPendingEvents(); UnregisterAllEvents(); @@ -171,7 +180,6 @@ void CoreTimingManager::RefreshConfig() void CoreTimingManager::DoState(PointerWrap& p) { - std::lock_guard lk(m_ts_write_lock); p.Do(m_globals.slice_length); p.Do(m_globals.global_timer); p.Do(m_idled_cycles); @@ -294,8 +302,9 @@ void CoreTimingManager::ScheduleEvent(s64 cycles_into_future, EventType* event_t *event_type->name); } - std::lock_guard lk(m_ts_write_lock); - m_ts_queue.Push(Event{cycles_into_future, 0, userdata, event_type}); + ThreadSafeEvent* const node = + new ThreadSafeEvent{{nullptr}, Event{cycles_into_future, 0, userdata, event_type}}; + mpsc_stack_push(&m_ts_queue, &node->link); } } @@ -332,10 +341,17 @@ void CoreTimingManager::ForceExceptionCheck(s64 cycles) void CoreTimingManager::MoveEvents() { - while (!m_ts_queue.Empty()) + if (mpsc_stack_empty(&m_ts_queue)) + return; + + // Oldest first, so events keep the order they were scheduled in. + mpsc_stack_node_t* node = mpsc_stack_reverse(mpsc_stack_drain(&m_ts_queue)); + while (node) { - auto& ev = m_event_queue.emplace_back(m_ts_queue.Front()); - m_ts_queue.Pop(); + ThreadSafeEvent* const ts_event = reinterpret_cast(node); + node = node->next; + auto& ev = m_event_queue.emplace_back(ts_event->event); + delete ts_event; ev.fifo_order = m_event_fifo_id++; ev.time += m_globals.global_timer; diff --git a/Source/Core/Core/CoreTiming.h b/Source/Core/Core/CoreTiming.h index 29d636e433..e687ddf820 100644 --- a/Source/Core/Core/CoreTiming.h +++ b/Source/Core/Core/CoreTiming.h @@ -16,7 +16,6 @@ // inside callback: // ScheduleEvent(periodInCycles - cyclesLate, callback, "whatever") -#include #include #include #include @@ -25,10 +24,11 @@ #include "Common/CommonTypes.h" #include "Common/Functional.h" #include "Common/HookableEvent.h" -#include "Common/SPSCQueue.h" #include "Common/Timer.h" #include "Core/CPUThreadConfigCallback.h" +#include + class PointerWrap; namespace Core @@ -92,6 +92,7 @@ class CoreTimingManager { public: explicit CoreTimingManager(Core::System& system); + ~CoreTimingManager(); // CoreTiming begins at the boundary of timing slice -1. An initial call to Advance() is // required to end slice -1 and start slice 0 before the first cycle of code is executed. @@ -188,11 +189,15 @@ private: // by the standard adaptor class. std::vector m_event_queue; u64 m_event_fifo_id = 0; - std::mutex m_ts_write_lock; - // Event objects created from other threads. - // The time value of each Event here is a cycles_into_future value. - Common::SPSCQueue m_ts_queue; + // Event objects created from other threads, pushed lock-free by any of them and taken all at + // once by the CPU thread. The time value of each Event here is a cycles_into_future value. + struct ThreadSafeEvent + { + mpsc_stack_node_t link; + Event event; + }; + mpsc_stack_t m_ts_queue; float m_last_oc_factor = 0.0f; diff --git a/Source/Core/Core/HW/DVD/AMMediaboard.cpp b/Source/Core/Core/HW/DVD/AMMediaboard.cpp index 0f434dfcc0..3dc6153803 100644 --- a/Source/Core/Core/HW/DVD/AMMediaboard.cpp +++ b/Source/Core/Core/HW/DVD/AMMediaboard.cpp @@ -3,6 +3,8 @@ #include "Core/HW/DVD/AMMediaboard.h" +#include + #include #include #include diff --git a/Source/Core/Core/HW/DVD/DVDThread.cpp b/Source/Core/Core/HW/DVD/DVDThread.cpp index e7b40fa629..cc34b70bf4 100644 --- a/Source/Core/Core/HW/DVD/DVDThread.cpp +++ b/Source/Core/Core/HW/DVD/DVDThread.cpp @@ -13,7 +13,6 @@ #include "Common/CommonTypes.h" #include "Common/Logging/Log.h" #include "Common/MsgHandler.h" -#include "Common/SPSCQueue.h" #include "Common/Timer.h" #include "Core/ConfigManager.h" @@ -33,9 +32,60 @@ namespace DVD { DVDThread::DVDThread(Core::System& system) : m_system(system) { + mpsc_stack_init(&m_result_queue); + retro_eventcount_init(&m_result_ready); } -DVDThread::~DVDThread() = default; +DVDThread::~DVDThread() +{ + m_dvd_thread.Shutdown(); + FreeResults(); + retro_eventcount_free(&m_result_ready); +} + +void DVDThread::FreeResults() +{ + mpsc_stack_node_t* node = m_result_chain; + m_result_chain = nullptr; + while (node) + { + mpsc_stack_node_t* const next = node->next; + delete reinterpret_cast(node); + node = next; + } + mpsc_stack_node_t* queued = mpsc_stack_drain(&m_result_queue); + while (queued) + { + mpsc_stack_node_t* const next = queued->next; + delete reinterpret_cast(queued); + queued = next; + } +} + +// CPU thread. Unlinks and returns the result with this id, or every result taken so far when +// id is ~0, without waiting; nullptr when none has arrived. +DVDThread::ResultNode* DVDThread::TakeResult(u64 id) +{ + if (!mpsc_stack_empty(&m_result_queue)) + { + mpsc_stack_node_t* const arrived = mpsc_stack_reverse(mpsc_stack_drain(&m_result_queue)); + mpsc_stack_node_t** tail = &m_result_chain; + while (*tail) + tail = &(*tail)->next; + *tail = arrived; + } + + for (mpsc_stack_node_t** link = &m_result_chain; *link; link = &(*link)->next) + { + ResultNode* const node = reinterpret_cast(*link); + if (id == ~u64{0} || node->result.first.id == id) + { + *link = node->link.next; + return node; + } + } + return nullptr; +} void DVDThread::Start() { @@ -53,7 +103,7 @@ void DVDThread::Stop() m_dvd_thread.Cancel(); m_dvd_thread.Shutdown(); - m_result_queue.Clear(); + FreeResults(); m_result_map.clear(); m_disc.reset(); @@ -65,11 +115,13 @@ void DVDThread::DoState(PointerWrap& p) WaitUntilIdle(); // Move all results from result_queue to result_map because - // PointerWrap::Do supports std::map but not Common::WaitableSPSCQueue. + // PointerWrap::Do supports std::map but not the queue. // This won't affect the behavior of FinishRead. - ReadResult result; - while (m_result_queue.Pop(result)) - m_result_map.emplace(result.first.id, std::move(result)); + while (ResultNode* const node = TakeResult(~u64{0})) + { + m_result_map.emplace(node->result.first.id, std::move(node->result)); + delete node; + } p.Do(m_result_map); p.Do(m_next_id); @@ -225,16 +277,8 @@ void DVDThread::GlobalFinishRead(Core::System& system, u64 id, s64 cycles_late) void DVDThread::FinishRead(u64 id, s64 cycles_late) { - // We can't simply pop result_queue and always get the ReadResult - // we want, because the DVD thread may add ReadResults to the queue - // in a different order than we want to get them. What we do instead - // is to pop the queue until we find the ReadResult we want (the one - // whose ID matches userdata), which means we may end up popping - // ReadResults that we don't want. We can't add those unwanted results - // back to the queue, because the queue can only have one writer. - // Instead, we add them to a map that only is used by the CPU thread. - // When this function is called again later, it will check the map for - // the wanted ReadResult before it starts searching through the queue. + // The DVD thread may finish reads in a different order than they are wanted, so results + // are looked up by id: in the map a savestate load filled, else among those that arrived. ReadResult result; auto it = m_result_map.find(id); if (it != m_result_map.end()) @@ -244,15 +288,19 @@ void DVDThread::FinishRead(u64 id, s64 cycles_late) } else { - while (true) + ResultNode* node; + while (!(node = TakeResult(id))) { - m_result_queue.WaitForData(); - m_result_queue.Pop(result); - if (result.first.id == id) - break; - - m_result_map.emplace(result.first.id, std::move(result)); + const int key = retro_eventcount_prepare_wait(&m_result_ready); + if (!mpsc_stack_empty(&m_result_queue)) + { + retro_eventcount_cancel_wait(&m_result_ready); + continue; + } + retro_eventcount_commit_wait(&m_result_ready, key); } + result = std::move(node->result); + delete node; } // We have now obtained the right ReadResult. @@ -303,6 +351,9 @@ void DVDThread::ProcessReadRequest(ReadRequest&& request) request.realtime_done_us = Common::Timer::NowUs(); - m_result_queue.Push(ReadResult(std::move(request), std::move(buffer))); + ResultNode* const node = + new ResultNode{{nullptr}, ReadResult(std::move(request), std::move(buffer))}; + mpsc_stack_push(&m_result_queue, &node->link); + retro_eventcount_notify(&m_result_ready); } } // namespace DVD diff --git a/Source/Core/Core/HW/DVD/DVDThread.h b/Source/Core/Core/HW/DVD/DVDThread.h index 7b40e436c1..19d579071b 100644 --- a/Source/Core/Core/HW/DVD/DVDThread.h +++ b/Source/Core/Core/HW/DVD/DVDThread.h @@ -10,7 +10,6 @@ #include #include "Common/CommonTypes.h" -#include "Common/SPSCQueue.h" #include "Common/WorkQueueThread.h" #include "Core/HW/DVD/DVDInterface.h" @@ -18,6 +17,9 @@ #include "DiscIO/Volume.h" +#include +#include + class PointerWrap; namespace Core { @@ -128,7 +130,19 @@ private: Common::WorkQueueThreadSP m_dvd_thread; - Common::WaitableSPSCQueue m_result_queue; + // Results travel from the DVD thread to the CPU thread through a lock-free stack; the CPU thread + // takes them all at once into m_result_chain, oldest first, and wakes on m_result_ready. + struct ResultNode + { + mpsc_stack_node_t link; + ReadResult result; + }; + ResultNode* TakeResult(u64 id); + void FreeResults(); + + mpsc_stack_t m_result_queue; + retro_eventcount_t m_result_ready; + mpsc_stack_node_t* m_result_chain = nullptr; // CPU thread only std::map m_result_map; std::unique_ptr m_disc; diff --git a/Source/Core/Core/HW/WiimoteReal/IOWin.cpp b/Source/Core/Core/HW/WiimoteReal/IOWin.cpp index 42c4aaa117..ced5206c84 100644 --- a/Source/Core/Core/HW/WiimoteReal/IOWin.cpp +++ b/Source/Core/Core/HW/WiimoteReal/IOWin.cpp @@ -3,6 +3,8 @@ #include "Core/HW/WiimoteReal/IOWin.h" +#include + #include #include #include diff --git a/Source/Core/Core/HW/WiimoteReal/IOhidapi.cpp b/Source/Core/Core/HW/WiimoteReal/IOhidapi.cpp index 36c9a2377d..a41da7b716 100644 --- a/Source/Core/Core/HW/WiimoteReal/IOhidapi.cpp +++ b/Source/Core/Core/HW/WiimoteReal/IOhidapi.cpp @@ -3,6 +3,8 @@ #include "Core/HW/WiimoteReal/IOhidapi.h" +#include + #if defined(__linux__) #include #endif diff --git a/Source/Core/Core/HW/WiimoteReal/WiimoteReal.cpp b/Source/Core/Core/HW/WiimoteReal/WiimoteReal.cpp index 1c0529c887..18415e9b7f 100644 --- a/Source/Core/Core/HW/WiimoteReal/WiimoteReal.cpp +++ b/Source/Core/Core/HW/WiimoteReal/WiimoteReal.cpp @@ -9,6 +9,7 @@ #include +#include "Common/Assert.h" #include "Common/CommonTypes.h" #include "Common/Config/Config.h" #include "Common/Swap.h" @@ -145,8 +146,14 @@ void AddWiimoteToPool(std::unique_ptr wiimote) s_wiimote_pool.emplace_back(WiimotePoolEntry{std::move(wiimote)}); } +// Records each ring holds: 256 reports is over a second of input at the Wiimote's 200 Hz. +static constexpr size_t REPORT_RING_RECORDS = 256; + Wiimote::Wiimote() { + retro_spsc_init(&m_read_reports, REPORT_RING_RECORDS * sizeof(ReadRecord)); + retro_spsc_init(&m_write_reports, REPORT_RING_RECORDS * sizeof(WriteRecord)); + retro_eventcount_init(&m_write_event); m_config_changed_callback_id = Config::AddConfigChangedCallback([this] { RefreshConfig(); }); RefreshConfig(); } @@ -154,6 +161,22 @@ Wiimote::Wiimote() Wiimote::~Wiimote() { Config::RemoveConfigChangedCallback(m_config_changed_callback_id); + retro_eventcount_free(&m_write_event); + retro_spsc_free(&m_write_reports); + retro_spsc_free(&m_read_reports); +} + +// Producer side of m_write_reports. A ring this deep is only full when the device has stopped +// taking reports, and the write thread is about to drop the connection then. +void Wiimote::QueueWrite(TimePoint time, const u8* data, size_t size) +{ + ASSERT(size <= MAX_PAYLOAD); + WriteRecord record; + record.time = time; + record.size = static_cast(size); + std::copy_n(data, size, record.data); + retro_spsc_write_frames(&m_write_reports, &record, 1, sizeof(record)); + retro_eventcount_notify(&m_write_event); } void Wiimote::Shutdown() @@ -202,8 +225,7 @@ void Wiimote::WriteReport(Report rpt) const auto report_time = Core::IsCPUThread() ? core_timing.GetTargetHostTime(core_timing.GetTicks()) : Clock::now(); - m_write_reports.Emplace(report_time, std::move(rpt)); - m_write_event.Set(); + QueueWrite(report_time, rpt.data(), rpt.size()); } // to be called from CPU thread @@ -232,12 +254,8 @@ void Wiimote::ResetDataReporting() void Wiimote::ClearReadQueue() { - Report rpt; - - // The "Clear" function isn't thread-safe :/ - while (m_read_reports.Pop(rpt)) - { - } + // Consumer side: discards what the read thread has delivered so far. + retro_spsc_skip(&m_read_reports, retro_spsc_read_avail(&m_read_reports)); } void Wiimote::EventLinked() @@ -293,41 +311,47 @@ void Wiimote::InterruptDataOutput(const u8* data, const u32 size) bool Wiimote::Read() { - Report rpt(MAX_PAYLOAD); - auto const result = IORead(rpt.data()); + // Receive straight into the next ring record. With the ring full the core has stopped taking + // reports; receive into a scratch record and drop it. + ReadRecord scratch; + ReadRecord* record = &scratch; + void* span; + const bool reserved = retro_spsc_write_begin(&m_read_reports, &span) >= sizeof(ReadRecord); + if (reserved) + record = static_cast(span); + + auto const result = IORead(record->data); // Drop the report if not connected. - if (m_is_linked && result > 0) + const bool keep = m_is_linked && result > 0; + if (keep && m_balance_board_dump_port > 0 && m_index == WIIMOTE_BALANCE_BOARD) { - if (m_balance_board_dump_port > 0 && m_index == WIIMOTE_BALANCE_BOARD) - { - static sf::UdpSocket Socket; - (void)Socket.send(rpt.data(), rpt.size(), sf::IpAddress::LocalHost, - m_balance_board_dump_port); - } + static sf::UdpSocket Socket; + (void)Socket.send(record->data, MAX_PAYLOAD, sf::IpAddress::LocalHost, + m_balance_board_dump_port); + } - // Add it to queue - rpt.resize(result); - m_read_reports.Push(std::move(rpt)); + if (reserved) + { + record->size = static_cast(result); + retro_spsc_write_end(&m_read_reports, keep ? sizeof(ReadRecord) : 0); } return result != 0; } -bool Wiimote::Write(const TimedReport& timed_report) +bool Wiimote::Write(const WriteRecord& record) { - auto const& rpt = timed_report.report; - if (m_balance_board_dump_port > 0 && m_index == WIIMOTE_BALANCE_BOARD) { static sf::UdpSocket Socket; - (void)Socket.send((char*)rpt.data(), rpt.size(), sf::IpAddress::LocalHost, + (void)Socket.send(record.data, record.size, sf::IpAddress::LocalHost, m_balance_board_dump_port); } // Write the report at the proper time, mainly for speaker data, not that it will help much. - std::this_thread::sleep_until(timed_report.time); - int ret = IOWrite(rpt.data(), rpt.size()); + std::this_thread::sleep_until(record.time); + int ret = IOWrite(record.data, record.size); return ret != 0; } @@ -416,7 +440,13 @@ static bool IsDataReport(const Report& rpt) bool Wiimote::GetNextReport(Report* report) { - return m_read_reports.Pop(*report); + const void* span; + if (retro_spsc_read_begin(&m_read_reports, &span) < sizeof(ReadRecord)) + return false; + const ReadRecord* const record = static_cast(span); + report->assign(record->data, record->data + record->size); + retro_spsc_read_end(&m_read_reports, sizeof(ReadRecord)); + return true; } // Returns the next report that should be sent @@ -503,15 +533,13 @@ void Wiimote::Prepare() const auto now = Clock::now(); // Set reporting mode to non-continuous core buttons and turn on rumble. - Report mode_report = {WR_SET_REPORT | BT_OUTPUT, u8(OutputReportID::ReportMode), 1, - u8(InputReportID::ReportCore)}; - m_write_reports.Emplace(now, std::move(mode_report)); + const u8 mode_report[] = {WR_SET_REPORT | BT_OUTPUT, u8(OutputReportID::ReportMode), 1, + u8(InputReportID::ReportCore)}; + QueueWrite(now, mode_report, sizeof(mode_report)); // Request status and turn off rumble. - Report req_status_report = {WR_SET_REPORT | BT_OUTPUT, u8(OutputReportID::RequestStatus), 0}; - m_write_reports.Emplace(now + std::chrono::milliseconds{200}, std::move(req_status_report)); - - m_write_event.Set(); + const u8 req_status_report[] = {WR_SET_REPORT | BT_OUTPUT, u8(OutputReportID::RequestStatus), 0}; + QueueWrite(now + std::chrono::milliseconds{200}, req_status_report, sizeof(req_status_report)); } void Wiimote::EmuStop() @@ -778,7 +806,7 @@ void Wiimote::StopThread() return; m_run_thread.Clear(); - m_write_event.Set(); + retro_eventcount_notify(&m_write_event); // Note that the write thread stops the read thread. m_write_thread.join(); @@ -794,7 +822,7 @@ void Wiimote::ReadThreadFunc() { WARN_LOG_FMT(WIIMOTE, "Wiimote::Read failed on Wiimote {}.", m_index + 1); m_run_thread.Clear(); - m_write_event.Set(); + retro_eventcount_notify(&m_write_event); break; } } @@ -834,11 +862,12 @@ void Wiimote::WriteThreadFunc() { bool write_success = false; - if (!m_write_reports.Empty()) + if (retro_spsc_read_avail(&m_write_reports) >= sizeof(WriteRecord)) { // Send a normal report. - write_success = Write(m_write_reports.Front()); - m_write_reports.Pop(); + WriteRecord record; + retro_spsc_read(&m_write_reports, &record, sizeof(record)); + write_success = Write(record); } else if (Clock::now() - last_write_time >= WRITE_TEST_INTERVAL) { @@ -851,7 +880,13 @@ void Wiimote::WriteThreadFunc() else { // Nothing to do. Wait a while for a kick. - m_write_event.WaitFor(WRITE_TEST_INTERVAL); + const int key = retro_eventcount_prepare_wait(&m_write_event); + if (retro_spsc_read_avail(&m_write_reports) >= sizeof(WriteRecord) || !m_run_thread.IsSet()) + retro_eventcount_cancel_wait(&m_write_event); + else + retro_eventcount_commit_wait_timeout( + &m_write_event, key, + std::chrono::duration_cast(WRITE_TEST_INTERVAL).count()); continue; } diff --git a/Source/Core/Core/HW/WiimoteReal/WiimoteReal.h b/Source/Core/Core/HW/WiimoteReal/WiimoteReal.h index 1512d134ec..b01c91f34c 100644 --- a/Source/Core/Core/HW/WiimoteReal/WiimoteReal.h +++ b/Source/Core/Core/HW/WiimoteReal/WiimoteReal.h @@ -14,7 +14,6 @@ #include "Common/Config/Config.h" #include "Common/Event.h" #include "Common/Flag.h" -#include "Common/SPSCQueue.h" #include "Common/WorkQueueThread.h" #include "Core/HW/Wiimote.h" @@ -23,6 +22,9 @@ #include "Core/HW/WiimoteCommon/WiimoteReport.h" #include "Core/USBUtils.h" +#include +#include + class PointerWrap; namespace WiimoteReal @@ -115,14 +117,25 @@ protected: u8 m_bt_device_index = 0; private: - struct TimedReport + // Reports cross threads as fixed 32-byte records in lock-free rings: a record never straddles + // the end of a ring, so the read thread receives straight into one. + struct ReadRecord + { + u8 size; + u8 data[MAX_PAYLOAD]; + u8 pad[32 - 1 - MAX_PAYLOAD]; + }; + struct WriteRecord { TimePoint time; - Report report; + u8 size; + u8 data[MAX_PAYLOAD]; }; + static_assert(sizeof(ReadRecord) == 32 && sizeof(WriteRecord) == 32); + void QueueWrite(TimePoint time, const u8* data, size_t size); bool Read(); - bool Write(const TimedReport& timed_report); + bool Write(const WriteRecord& record); void StartThread(); void StopThread(); @@ -165,10 +178,10 @@ private: // Triggered when the thread has finished ConnectInternal. Common::Event m_thread_ready_event; - Common::SPSCQueue m_read_reports; - Common::SPSCQueue m_write_reports; + retro_spsc_t m_read_reports; + retro_spsc_t m_write_reports; // Kick the write thread. - Common::Event m_write_event; + retro_eventcount_t m_write_event; bool m_speaker_enabled_in_dolphin_config = false; int m_balance_board_dump_port = 0; diff --git a/Source/Core/Core/IOS/Network/KD/NetKDRequest.cpp b/Source/Core/Core/IOS/Network/KD/NetKDRequest.cpp index 484c9d190c..b0e54bd089 100644 --- a/Source/Core/Core/IOS/Network/KD/NetKDRequest.cpp +++ b/Source/Core/Core/IOS/Network/KD/NetKDRequest.cpp @@ -2,6 +2,7 @@ // SPDX-License-Identifier: GPL-2.0-or-later #include "Core/IOS/Network/KD/NetKDRequest.h" +#include "Common/Thread.h" #include #include diff --git a/Source/Core/Core/IOS/Network/KD/NetKDRequest.h b/Source/Core/Core/IOS/Network/KD/NetKDRequest.h index e41b554d5d..7919bda29f 100644 --- a/Source/Core/Core/IOS/Network/KD/NetKDRequest.h +++ b/Source/Core/Core/IOS/Network/KD/NetKDRequest.h @@ -7,6 +7,7 @@ #include #include #include +#include #include "Common/CommonTypes.h" #include "Common/Event.h" diff --git a/Source/Core/Core/IOS/USB/Bluetooth/BTReal.cpp b/Source/Core/Core/IOS/USB/Bluetooth/BTReal.cpp index 682d5553e7..03dfd79bb4 100644 --- a/Source/Core/Core/IOS/USB/Bluetooth/BTReal.cpp +++ b/Source/Core/Core/IOS/USB/Bluetooth/BTReal.cpp @@ -3,6 +3,8 @@ #include "Core/IOS/USB/Bluetooth/BTReal.h" +#include + #include #include #include @@ -302,7 +304,7 @@ void BluetoothRealDevice::TryToFillACLEndpoint() auto& core_timing = GetSystem().GetCoreTiming(); core_timing.Throttle(core_timing.GetTicks()); - const auto buffer = m_lib_usb_bt_adapter->ReceiveACLData(); + const auto buffer = m_lib_usb_bt_adapter->PeekACLData(); if (buffer.empty()) return; @@ -310,6 +312,7 @@ void BluetoothRealDevice::TryToFillACLEndpoint() m_acl_endpoint->FillBuffer(buffer.data(), buffer.size()); GetEmulationKernel().EnqueueIPCReply(m_acl_endpoint->ios_request, s32(buffer.size())); + m_lib_usb_bt_adapter->PopACLData(); m_acl_endpoint.reset(); } diff --git a/Source/Core/Core/IOS/USB/Bluetooth/LibUSBBluetoothAdapter.cpp b/Source/Core/Core/IOS/USB/Bluetooth/LibUSBBluetoothAdapter.cpp index cb8bd3e38e..4e6f9cb13d 100644 --- a/Source/Core/Core/IOS/USB/Bluetooth/LibUSBBluetoothAdapter.cpp +++ b/Source/Core/Core/IOS/USB/Bluetooth/LibUSBBluetoothAdapter.cpp @@ -2,6 +2,9 @@ // SPDX-License-Identifier: GPL-2.0-or-later #include "Core/IOS/USB/Bluetooth/LibUSBBluetoothAdapter.h" +#include "Common/Thread.h" + +#include #include #include @@ -86,6 +89,11 @@ bool LibUSBBluetoothAdapter::IsConfiguredBluetoothDevice(u16 vid, u16 pid) LibUSBBluetoothAdapter::LibUSBBluetoothAdapter() { + retro_atomic_int_init(&m_run_input_transfers, 1); + mpsc_stack_init(&m_transfers_to_free); + retro_spsc_init_mirrored(&m_hci_event_queue, INPUT_RING_SIZE); + retro_spsc_init_mirrored(&m_acl_data_queue, INPUT_RING_SIZE); + if (!m_context.IsValid()) return; @@ -175,9 +183,15 @@ LibUSBBluetoothAdapter::LibUSBBluetoothAdapter() LibUSBBluetoothAdapter::~LibUSBBluetoothAdapter() { - if (m_handle == nullptr) - return; + if (m_handle != nullptr) + Close(); + retro_spsc_free(&m_acl_data_queue); + retro_spsc_free(&m_hci_event_queue); +} + +void LibUSBBluetoothAdapter::Close() +{ // Wait for completion (or time out) of all HCI commands. while (AreCommandsPendingResponse()) { @@ -188,16 +202,19 @@ LibUSBBluetoothAdapter::~LibUSBBluetoothAdapter() m_output_worker.Shutdown(); // Stop the repeating input transfers. - if (std::lock_guard lg{m_transfers_mutex}; true) - m_run_input_transfers = false; + retro_atomic_store_release_int(&m_run_input_transfers, 0); - // Cancel all LibUSB transfers. - std::ranges::for_each(m_transfer_buffers | std::ranges::views::keys, libusb_cancel_transfer); - - // Wait for transfer callbacks and clean up all the buffers. + // Cancel all LibUSB transfers, wait for their callbacks and clean up all the buffers. A + // callback that read the run flag before it was cleared may resubmit its transfer after the + // cancel, so whatever is still in flight is cancelled again on every pass. while (!m_transfer_buffers.empty()) { - m_transfers_to_free.WaitForData(); + std::ranges::for_each(m_transfer_buffers | std::ranges::views::keys, libusb_cancel_transfer); + const int key = retro_eventcount_prepare_wait(&m_transfer_freed.ec); + if (!mpsc_stack_empty(&m_transfers_to_free)) + retro_eventcount_cancel_wait(&m_transfer_freed.ec); + else + retro_eventcount_commit_wait_timeout(&m_transfer_freed.ec, key, 10000); CleanCompletedTransfers(); } @@ -236,16 +253,29 @@ void LibUSBBluetoothAdapter::Update() CleanCompletedTransfers(); } +// Consumer side of an input ring: the length of the whole packet at its front, or 0. +static u16 PeekPacketLength(const retro_spsc_t* queue) +{ + const std::size_t avail = retro_spsc_read_avail(queue); + u16 length; + if (avail < sizeof(length)) + return 0; + retro_spsc_peek(queue, &length, sizeof(length)); + return avail >= sizeof(length) + length ? length : 0; +} + auto LibUSBBluetoothAdapter::ReceiveHCIEvent() -> BufferType { - if (m_hci_event_queue.Empty()) + const u16 length = PeekPacketLength(&m_hci_event_queue); + if (length == 0) { Update(); return {}; } - auto buffer = std::move(m_hci_event_queue.Front()); - m_hci_event_queue.Pop(); + BufferType buffer(length); + retro_spsc_skip(&m_hci_event_queue, sizeof(length)); + retro_spsc_read(&m_hci_event_queue, buffer.data(), length); // We only care about "COMMAND_COMPL" and "COMMAND_STATUS" events here. // The controller will reply with one of those for every command. @@ -264,11 +294,26 @@ auto LibUSBBluetoothAdapter::ReceiveHCIEvent() -> BufferType return buffer; } -auto LibUSBBluetoothAdapter::ReceiveACLData() -> BufferType +std::span LibUSBBluetoothAdapter::PeekACLData() { - BufferType buffer; - m_acl_data_queue.Pop(buffer); - return buffer; + const u16 length = PeekPacketLength(&m_acl_data_queue); + if (length == 0) + return {}; + + m_acl_peeked = sizeof(length) + length; + const void* span; + if (retro_spsc_read_begin(&m_acl_data_queue, &span) >= m_acl_peeked) + return {static_cast(span) + sizeof(length), length}; + + // Only a ring without a mirror splits a packet at its end. + m_acl_bounce.resize(m_acl_peeked); + retro_spsc_peek(&m_acl_data_queue, m_acl_bounce.data(), m_acl_peeked); + return {m_acl_bounce.data() + sizeof(length), length}; +} + +void LibUSBBluetoothAdapter::PopACLData() +{ + retro_spsc_skip(&m_acl_data_queue, std::exchange(m_acl_peeked, 0)); } bool LibUSBBluetoothAdapter::IsControllerReadyForCommand() const @@ -305,12 +350,13 @@ void LibUSBBluetoothAdapter::AcknowledgeCommand(std::span buffer) libusb_transfer* LibUSBBluetoothAdapter::AllocateTransfer(std::size_t buffer_size) { - auto& item = - *m_transfer_buffers.emplace(libusb_alloc_transfer(0), Common::UniqueBuffer{buffer_size}) - .first; + auto& item = *m_transfer_buffers + .emplace(libusb_alloc_transfer(0), + Common::UniqueBuffer{BUFFER_HEADER_SIZE + buffer_size}) + .first; auto* const transfer = item.first; - transfer->buffer = item.second.get(); + transfer->buffer = item.second.get() + BUFFER_HEADER_SIZE; return transfer; } @@ -321,11 +367,25 @@ void LibUSBBluetoothAdapter::FreeTransfer(libusb_transfer* transfer) m_transfer_buffers.erase(transfer); } +// Any thread. +void LibUSBBluetoothAdapter::QueueFree(libusb_transfer* transfer) +{ + FreeNode* const node = reinterpret_cast(transfer->buffer - BUFFER_HEADER_SIZE); + node->transfer = transfer; + mpsc_stack_push(&m_transfers_to_free, &node->link); + retro_eventcount_notify(&m_transfer_freed.ec); +} + void LibUSBBluetoothAdapter::CleanCompletedTransfers() { - libusb_transfer* transfer{}; - while (m_transfers_to_free.Pop(transfer)) - FreeTransfer(transfer); + mpsc_stack_node_t* node = mpsc_stack_drain(&m_transfers_to_free); + while (node) + { + // The node is freed with its transfer's buffer. + mpsc_stack_node_t* const next = node->next; + FreeTransfer(reinterpret_cast(node)->transfer); + node = next; + } } void LibUSBBluetoothAdapter::PushHCICommand(TimedTransfer transfer) @@ -365,8 +425,7 @@ void LibUSBBluetoothAdapter::SubmitTransfer(libusb_transfer* transfer) ERROR_LOG_FMT(IOS_WIIMOTE, "libusb_submit_transfer: {}", LibusbUtils::ErrorWrap(ret)); // Failed transers will not invoke callbacks so we must mark the buffer to be free'd here. - std::lock_guard lk{m_transfers_mutex}; - m_transfers_to_free.Emplace(transfer); + QueueFree(transfer); } void LibUSBBluetoothAdapter::ScheduleBulkTransfer(u8 endpoint, std::span data, @@ -520,9 +579,7 @@ bool LibUSBBluetoothAdapter::OpenDevice(const libusb_device_descriptor& device_d void LibUSBBluetoothAdapter::HandleOutputTransfer(libusb_transfer* tr) { HandleTransferError(tr); - - std::lock_guard lg{m_transfers_mutex}; - m_transfers_to_free.Emplace(tr); + QueueFree(tr); } void LibUSBBluetoothAdapter::HandleInputTransfer(libusb_transfer* tr) @@ -530,15 +587,16 @@ void LibUSBBluetoothAdapter::HandleInputTransfer(libusb_transfer* tr) if (tr->status == LIBUSB_TRANSFER_COMPLETED) { const bool is_hci_event = tr->endpoint == HCI_EVENT; - auto& queue = is_hci_event ? m_hci_event_queue : m_acl_data_queue; + retro_spsc_t* const queue = is_hci_event ? &m_hci_event_queue : &m_acl_data_queue; + const u16 length = static_cast(tr->actual_length); - if (queue.Size() < MAX_INPUT_QUEUE_SIZE) + if (retro_spsc_write_avail(queue) >= sizeof(length) + length) { - if (tr->actual_length != 0) + if (length != 0) { - BufferType buffer(tr->actual_length); - std::copy_n(tr->buffer, tr->actual_length, buffer.data()); - queue.Emplace(std::move(buffer)); + // The consumer only takes a packet once all of it is there. + retro_spsc_write(queue, &length, sizeof(length)); + retro_spsc_write(queue, tr->buffer, length); } m_showed_long_queue_drop = false; @@ -553,9 +611,7 @@ void LibUSBBluetoothAdapter::HandleInputTransfer(libusb_transfer* tr) HandleTransferError(tr); - std::lock_guard lg{m_transfers_mutex}; - - if (m_run_input_transfers) + if (retro_atomic_load_acquire_int(&m_run_input_transfers)) { // Continuously repeat this input transfer so long as we're still running. const auto ret = libusb_submit_transfer(tr); @@ -566,7 +622,7 @@ void LibUSBBluetoothAdapter::HandleInputTransfer(libusb_transfer* tr) LibusbUtils::ErrorWrap(ret)); } - m_transfers_to_free.Emplace(tr); + QueueFree(tr); } void LibUSBBluetoothAdapter::HandleTransferError(libusb_transfer* transfer) diff --git a/Source/Core/Core/IOS/USB/Bluetooth/LibUSBBluetoothAdapter.h b/Source/Core/Core/IOS/USB/Bluetooth/LibUSBBluetoothAdapter.h index 0a8f63bcc1..ce33596dba 100644 --- a/Source/Core/Core/IOS/USB/Bluetooth/LibUSBBluetoothAdapter.h +++ b/Source/Core/Core/IOS/USB/Bluetooth/LibUSBBluetoothAdapter.h @@ -7,6 +7,7 @@ #include #include #include +#include #include "Common/Buffer.h" #include "Common/Timer.h" @@ -15,6 +16,11 @@ #include "Core/IOS/USB/Bluetooth/hci.h" #include "Core/LibusbUtils.h" +#include +#include +#include +#include + struct libusb_device_handle; struct libusb_device_descriptor; struct libusb_transfer; @@ -43,7 +49,11 @@ public: // Return a packet at the front of the queue, or an empty buffer when nothing is queued. // ReceiveHCIEvent is expected to be called regularly to do some internal processing. [[nodiscard]] BufferType ReceiveHCIEvent(); - [[nodiscard]] BufferType ReceiveACLData(); + + // The ACL packet at the front of the queue, read in place; empty when nothing is queued. + // PopACLData releases it. + [[nodiscard]] std::span PeekACLData(); + void PopACLData(); // Schedule a transfer to submit at a specific time. void ScheduleBulkTransfer(u8 endpoint, std::span data, TimePoint target_time); @@ -61,28 +71,43 @@ public: bool IsWiiBTModule() const; private: - // Inputs will be dropped when queue is full. - static constexpr std::size_t MAX_INPUT_QUEUE_SIZE = 100; + // Bytes each input ring holds; inputs are dropped when it is full. + static constexpr std::size_t INPUT_RING_SIZE = 64 * 1024; static constexpr u8 INTERFACE = 0x00; std::string m_last_open_error; + // Callbacks on the libusb event thread notify this until that thread is joined, which happens + // when m_context goes, so it is declared first to be released last. + struct TransferFreedEvent + { + TransferFreedEvent() { retro_eventcount_init(&ec); } + ~TransferFreedEvent() { retro_eventcount_free(&ec); } + retro_eventcount_t ec; + } m_transfer_freed; + LibusbUtils::Context m_context; libusb_device_handle* m_handle = nullptr; - // Protects run-flag and Push'ing to the completed transfer queue. - std::mutex m_transfers_mutex; + retro_atomic_int_t m_run_input_transfers; - bool m_run_input_transfers = true; + // Callback and worker threads push transfers here, lock-free, when they are ready for clean up. + // The node lives in the front of the transfer's own buffer, so this allocates nothing. + struct FreeNode + { + mpsc_stack_node_t link; + libusb_transfer* transfer; + }; + static constexpr std::size_t BUFFER_HEADER_SIZE = (sizeof(FreeNode) + 15) & ~std::size_t{15}; + void QueueFree(libusb_transfer* transfer); + mpsc_stack_t m_transfers_to_free; - // callback/worker threads push transfers here that are ready for clean up. - // This avoids locking around m_transfer_buffers. - Common::WaitableSPSCQueue m_transfers_to_free; - - // Incoming packets Push'd from libusb callback thread. - Common::SPSCQueue m_hci_event_queue; - Common::SPSCQueue m_acl_data_queue; + // Incoming packets from the libusb callback thread, each a 16-bit length and the payload. + retro_spsc_t m_hci_event_queue; + retro_spsc_t m_acl_data_queue; + std::size_t m_acl_peeked = 0; + std::vector m_acl_bounce; // All transfers are alloc'd and kept here until complete. std::map> m_transfer_buffers; @@ -137,6 +162,7 @@ private: void AcknowledgeCommand(std::span buffer); libusb_transfer* AllocateTransfer(std::size_t buffer_size); + void Close(); void FreeTransfer(libusb_transfer*); void CleanCompletedTransfers(); diff --git a/Source/Core/Core/IOS/USB/Emulated/LogitechMic.cpp b/Source/Core/Core/IOS/USB/Emulated/LogitechMic.cpp index ff62b9f02e..ce014ba4c7 100644 --- a/Source/Core/Core/IOS/USB/Emulated/LogitechMic.cpp +++ b/Source/Core/Core/IOS/USB/Emulated/LogitechMic.cpp @@ -3,6 +3,8 @@ #include "Core/IOS/USB/Emulated/LogitechMic.h" +#include + #include #include diff --git a/Source/Core/DolphinLibretro/Boot.cpp b/Source/Core/DolphinLibretro/Boot.cpp index 51e8a77878..e5c8886e7d 100644 --- a/Source/Core/DolphinLibretro/Boot.cpp +++ b/Source/Core/DolphinLibretro/Boot.cpp @@ -6,6 +6,7 @@ #include #include "Common/CommonPaths.h" +#include "Common/Thread.h" #include "Core/CommonTitles.h" #include "Common/FileUtil.h" #include "Common/StringUtil.h" diff --git a/Source/Core/VideoCommon/AsyncRequests.cpp b/Source/Core/VideoCommon/AsyncRequests.cpp index 7348724e43..3d5fa11afa 100644 --- a/Source/Core/VideoCommon/AsyncRequests.cpp +++ b/Source/Core/VideoCommon/AsyncRequests.cpp @@ -4,43 +4,64 @@ #include "VideoCommon/AsyncRequests.h" #include "Core/System.h" - #include "VideoCommon/Fifo.h" #include "VideoCommon/VertexManagerBase.h" -#include "VideoCommon/VideoBackendBase.h" -#include "VideoCommon/VideoEvents.h" AsyncRequests AsyncRequests::s_singleton; -AsyncRequests::AsyncRequests() = default; +// Pending requests the ring holds before the CPU thread has to wait for the video thread. +static constexpr size_t QUEUE_REQUESTS = 1024; + +AsyncRequests::AsyncRequests() +{ + retro_waitable_spsc_init(&m_queue, QUEUE_REQUESTS * sizeof(Request*)); + retro_eventcount_init(&m_completed); +} + +AsyncRequests::~AsyncRequests() +{ + retro_eventcount_free(&m_completed); + retro_waitable_spsc_free(&m_queue); +} void AsyncRequests::PullEvents() { - if (m_queue.Empty()) + if (retro_spsc_read_avail(&m_queue.queue) < sizeof(Request*)) return; // This is only called if the queue isn't empty. // So just flush the pipeline to get accurate results. g_vertex_manager->Flush(); - while (!m_queue.Empty()) - { - std::invoke(std::move(m_queue.Front())); - m_queue.Pop(); - } + Request* request; + while (retro_waitable_spsc_read(&m_queue, &request, sizeof(request)) == sizeof(request)) + request->run(request); } -void AsyncRequests::QueueEvent(Event&& event) +void AsyncRequests::Submit(Request* request) { - m_queue.Push(std::move(event)); + // Only waits when the video thread is QUEUE_REQUESTS requests behind. + while (!retro_waitable_spsc_wait_writable(&m_queue, sizeof(request), 1000000)) + { + } + retro_waitable_spsc_write(&m_queue, &request, sizeof(request)); auto& system = Core::System::GetInstance(); system.GetFifo().RunGpu(); } -void AsyncRequests::WaitForEmptyQueue() +void AsyncRequests::WaitDone(retro_atomic_int_t* done) { - m_queue.WaitForEmpty(); + while (!retro_atomic_load_acquire_int(done)) + { + const int key = retro_eventcount_prepare_wait(&m_completed); + if (retro_atomic_load_acquire_int(done)) + { + retro_eventcount_cancel_wait(&m_completed); + break; + } + retro_eventcount_commit_wait(&m_completed, key); + } } void AsyncRequests::SetPassthrough(bool enable) diff --git a/Source/Core/VideoCommon/AsyncRequests.h b/Source/Core/VideoCommon/AsyncRequests.h index 2c3e84fd47..8f82dd75b3 100644 --- a/Source/Core/VideoCommon/AsyncRequests.h +++ b/Source/Core/VideoCommon/AsyncRequests.h @@ -5,25 +5,30 @@ #include #include -#include +#include +#include +#include -#include "Common/Functional.h" -#include "Common/SPSCQueue.h" +#include +#include +#include struct EfbPokeData; class PointerWrap; +// Requests from the CPU thread to the video thread. The CPU thread is the only producer and the +// video thread the only consumer, so the requests travel through a lock-free SPSC ring as +// pointers: a blocking request lives on the caller's stack and allocates nothing. class AsyncRequests { public: AsyncRequests(); + ~AsyncRequests(); // Called from the Video thread. void PullEvents(); // The following are called from the CPU thread. - void WaitForEmptyQueue(); - template F> void PushEvent(F&& callback) { @@ -33,7 +38,18 @@ public: return; } - QueueEvent(Event{std::forward(callback)}); + using Fn = std::decay_t; + struct Owned : Request + { + Fn fn; + static void Run(Request* r) + { + Owned* const self = static_cast(r); + std::invoke(self->fn); + delete self; + } + }; + Submit(new Owned{{&Owned::Run}, Fn(std::forward(callback))}); } template F> @@ -42,10 +58,33 @@ public: if (m_passthrough) return std::invoke(std::forward(callback)); - std::packaged_task task{std::forward(callback)}; - QueueEvent(Event{[&] { task(); }}); - - return task.get_future().get(); + using R = std::invoke_result_t; + using Fn = std::remove_reference_t; + struct Blocking : Request + { + Fn* fn; + retro_atomic_int_t done; + std::conditional_t, char, std::optional> result; + static void Run(Request* r) + { + Blocking* const self = static_cast(r); + if constexpr (std::is_void_v) + std::invoke(*self->fn); + else + self->result.emplace(std::invoke(*self->fn)); + // The caller owns this request again from here on. + retro_atomic_store_release_int(&self->done, 1); + retro_eventcount_notify(&s_singleton.m_completed); + } + }; + Blocking request; + request.run = &Blocking::Run; + request.fn = &callback; + retro_atomic_int_init(&request.done, 0); + Submit(&request); + WaitDone(&request.done); + if constexpr (!std::is_void_v) + return std::move(*request.result); } // Not thread-safe. Only set during initialization. @@ -54,13 +93,17 @@ public: static AsyncRequests* GetInstance() { return &s_singleton; } private: - using Event = Common::MoveOnlyFunction; + struct Request + { + void (*run)(Request*); + }; - void QueueEvent(Event&& event); + void Submit(Request* request); + void WaitDone(retro_atomic_int_t* done); static AsyncRequests s_singleton; - Common::WaitableSPSCQueue m_queue; - + retro_waitable_spsc_t m_queue; + retro_eventcount_t m_completed; bool m_passthrough = true; }; diff --git a/Source/Core/VideoCommon/PerformanceTracker.cpp b/Source/Core/VideoCommon/PerformanceTracker.cpp index 3d519865be..1739c7d06e 100644 --- a/Source/Core/VideoCommon/PerformanceTracker.cpp +++ b/Source/Core/VideoCommon/PerformanceTracker.cpp @@ -23,12 +23,18 @@ PerformanceTracker::PerformanceTracker(std::optional log_name, const std::optional
sample_window_duration) : m_log_name{std::move(log_name)}, m_sample_window_duration{sample_window_duration} { + retro_spsc_init(&m_raw_dts, MAX_DT_QUEUE_SIZE * sizeof(DT)); Reset(); } +PerformanceTracker::~PerformanceTracker() +{ + retro_spsc_free(&m_raw_dts); +} + void PerformanceTracker::Reset() { - m_raw_dts.Clear(); + retro_spsc_skip(&m_raw_dts, retro_spsc_read_avail(&m_raw_dts)); m_dt_queue.clear(); m_dt_total = DT::zero(); @@ -54,14 +60,18 @@ void PerformanceTracker::Count() } m_last_raw_dt = diff; - m_raw_dts.Push(diff); + // A full ring already holds the MAX_DT_QUEUE_SIZE samples UpdateStats keeps. + retro_spsc_write_frames(&m_raw_dts, &diff, 1, sizeof(diff)); } void PerformanceTracker::UpdateStats() { DT diff{}; - while (m_raw_dts.Pop(diff)) + while (retro_spsc_read_avail(&m_raw_dts) >= sizeof(diff)) + { + retro_spsc_read(&m_raw_dts, &diff, sizeof(diff)); HandleRawDt(diff); + } // Update Std Dev MathUtil::RunningVariance variance; diff --git a/Source/Core/VideoCommon/PerformanceTracker.h b/Source/Core/VideoCommon/PerformanceTracker.h index e210a8497c..c1074bf7cc 100644 --- a/Source/Core/VideoCommon/PerformanceTracker.h +++ b/Source/Core/VideoCommon/PerformanceTracker.h @@ -9,14 +9,15 @@ #include #include "Common/CommonTypes.h" -#include "Common/SPSCQueue.h" + +#include class PerformanceTracker { public: PerformanceTracker(std::optional log_name = std::nullopt, const std::optional
sample_window_duration = std::nullopt); - ~PerformanceTracker() = default; + ~PerformanceTracker(); PerformanceTracker(const PerformanceTracker&) = delete; PerformanceTracker& operator=(const PerformanceTracker&) = delete; @@ -58,7 +59,8 @@ private: // Push'd from Count() // and Pop'd from UpdateStats() - Common::SPSCQueue
m_raw_dts; + // Frame times from Count() to UpdateStats(), lock-free. + retro_spsc_t m_raw_dts; std::atomic
m_last_raw_dt = DT::zero(); // Amount of time to sample dt's over (defaults to config) diff --git a/Source/UnitTests/Common/SPSCQueueTest.cpp b/Source/UnitTests/Common/SPSCQueueTest.cpp deleted file mode 100644 index ddc57c8d1a..0000000000 --- a/Source/UnitTests/Common/SPSCQueueTest.cpp +++ /dev/null @@ -1,91 +0,0 @@ -// Copyright 2014 Dolphin Emulator Project -// SPDX-License-Identifier: GPL-2.0-or-later - -#include - -#include -#include - -#include "Common/CommonTypes.h" -#include "Common/SPSCQueue.h" - -TEST(SPSCQueue, Simple) -{ - Common::SPSCQueue q; - - EXPECT_EQ(0u, q.Size()); - EXPECT_TRUE(q.Empty()); - - q.Push(1); - EXPECT_EQ(1u, q.Size()); - EXPECT_FALSE(q.Empty()); - - u32 v; - q.Pop(v); - EXPECT_EQ(1u, v); - EXPECT_EQ(0u, q.Size()); - EXPECT_TRUE(q.Empty()); - - // Test the FIFO order. - for (u32 i = 0; i < 1000; ++i) - q.Push(i); - EXPECT_EQ(1000u, q.Size()); - for (u32 i = 0; i < 1000; ++i) - { - u32 v2 = 0; - q.Pop(v2); - EXPECT_EQ(i, v2); - } - EXPECT_TRUE(q.Empty()); - - for (u32 i = 0; i < 1000; ++i) - q.Push(i); - EXPECT_FALSE(q.Empty()); - q.Clear(); - EXPECT_TRUE(q.Empty()); -} - -TEST(SPSCQueue, MultiThreaded) -{ - struct Foo - { - std::shared_ptr ptr; - u32 i; - }; - - // A shared_ptr held by every element in the queue. - auto sptr = std::make_shared(0); - - auto queue_ptr = std::make_unique>(); - auto& q = *queue_ptr; - - constexpr u32 reps = 100000; - - auto inserter = [&] { - for (u32 i = 0; i != reps; ++i) - q.Push({sptr, i}); - - q.WaitForEmpty(); - EXPECT_EQ(sptr.use_count(), 1); - q.Push({sptr, 0}); - EXPECT_EQ(sptr.use_count(), 2); - }; - - auto popper = [&] { - for (u32 i = 0; i != reps; ++i) - { - q.WaitForData(); - EXPECT_EQ(i, q.Front().i); - q.Pop(); - } - }; - - std::thread popper_thread(popper); - std::thread inserter_thread(inserter); - - popper_thread.join(); - inserter_thread.join(); - - queue_ptr.reset(); - EXPECT_EQ(sptr.use_count(), 1); -}