Common: Introduce FlushThread class to flush data on a thread with a delay.

Source/Core/Common/CMakeLists.txt

@@ -70,6 +70,7 @@ add_library(common
   Flag.h
   FloatUtils.cpp
   FloatUtils.h
+  FlushThread.h
   FormatUtil.h
   FPURoundMode.h
   GekkoDisassembler.cpp

Source/Core/Common/FlushThread.h

@@ -0,0 +1,143 @@
+// Copyright 2025 Dolphin Emulator Project
+// SPDX-License-Identifier: GPL-2.0-or-later
+
+#pragma once
+
+#include <atomic>
+#include <functional>
+#include <semaphore>
+#include <thread>
+
+#include "Common/CommonTypes.h"
+#include "Common/Event.h"
+#include "Common/Thread.h"
+
+// This class allows flushing data writes in a delayed manner.
+// When SetDirty is called the provided function will be invoked on thread with configured delay.
+// Multiple SetDirty calls may produce just one flush, delay based on the last call.
+
+namespace Common
+{
+
+class FlushThread final
+{
+public:
+  FlushThread() = default;
+  explicit FlushThread(std::string name, std::function<void()> func)
+  {
+    Reset(std::move(name), std::move(func));
+  }
+
+  ~FlushThread() { Shutdown(); }
+
+  FlushThread(const FlushThread&) = delete;
+  FlushThread& operator=(const FlushThread&) = delete;
+
+  FlushThread(FlushThread&&) = delete;
+  FlushThread& operator=(FlushThread&&) = delete;
+
+  // May not take effect until clean.
+  void SetFlushDelay(DT delay) { m_flush_delay.store(delay, std::memory_order_relaxed); }
+
+  // (Re)Starts the thread with the provided flush function.
+  // Other state is unchanged.
+  void Reset(std::string name, std::function<void()> func)
+  {
+    Shutdown();
+
+    m_want_shutdown.store(false, std::memory_order_relaxed);
+    m_thread = std::thread{std::bind_front(&FlushThread::ThreadFunc, this), std::move(name),
+                           std::move(func)};
+  }
+
+  // Graceful immediate shutdown. Waits for final flush if necessary.
+  // Does nothing if thread isn't running.
+  void Shutdown()
+  {
+    if (!m_thread.joinable())
+      return;
+
+    WaitForCompletion();
+    m_want_shutdown.store(true, std::memory_order_relaxed);
+    m_event.Set();
+    m_thread.join();
+  }
+
+  void SetDirty()
+  {
+    m_dirty_count.fetch_add(1, std::memory_order_relaxed);
+    m_flush_deadline.store((Clock::now() + m_flush_delay.load(std::memory_order_relaxed)));
+    m_event.Set();
+  }
+
+  // Lets the worker immediately flush if necessary.
+  // Does nothing if thread isn't running.
+  void WaitForCompletion()
+  {
+    if (!m_thread.joinable())
+      return;
+
+    m_run_freely.release();
+
+    m_event.Set();
+    // Wait for m_dirty_count == 0.
+    while (auto old_count = m_dirty_count.load(std::memory_order_acquire))
+      m_dirty_count.wait(old_count, std::memory_order_acquire);
+
+    m_run_freely.acquire();
+  }
+
+private:
+  auto GetDeadline() const { return m_flush_deadline.load(std::memory_order_relaxed); }
+
+  void WaitUntilFlushIsWanted()
+  {
+    while (!m_run_freely.try_acquire_until(GetDeadline()))
+    {
+      if (Clock::now() >= GetDeadline())
+        return;
+    }
+    m_run_freely.release();
+  }
+
+  void ThreadFunc(const std::string& name, const std::function<void()>& flush_func)
+  {
+    Common::SetCurrentThreadName(name.c_str());
+
+    while (true)
+    {
+      m_event.Wait();
+
+      if (m_want_shutdown.load(std::memory_order_relaxed))
+        break;
+
+      WaitUntilFlushIsWanted();
+
+      const auto cleaning_count = m_dirty_count.load(std::memory_order_relaxed);
+      if (cleaning_count != 0)
+      {
+        flush_func();
+        m_dirty_count.fetch_sub(cleaning_count, std::memory_order_release);
+        m_dirty_count.notify_all();
+      }
+    }
+  }
+
+  // Incremented when a flush needs to happen.
+  // Decremented by worker-thread to signal completion.
+  std::atomic<u32> m_dirty_count{};
+
+  std::atomic<DT> m_flush_delay{};
+  std::atomic<TimePoint> m_flush_deadline{};
+
+  // Worker tries to acquire this for the flush delay.
+  // Releasing it lets the worker run without waiting.
+  std::counting_semaphore<> m_run_freely{0};
+
+  std::thread m_thread;
+  Common::Event m_event;
+
+  std::atomic_bool m_want_shutdown{};
+};
+
+}  // namespace Common

Source/Core/DolphinLib.props

@@ -60,6 +60,7 @@
     <ClInclude Include="Common\FixedSizeQueue.h" />
     <ClInclude Include="Common\Flag.h" />
     <ClInclude Include="Common\FloatUtils.h" />
+    <ClInclude Include="Common\FlushThread.h" />
     <ClInclude Include="Common\FormatUtil.h" />
     <ClInclude Include="Common\FPURoundMode.h" />
     <ClInclude Include="Common\GekkoDisassembler.h" />

Source/UnitTests/Common/CMakeLists.txt

@@ -13,6 +13,7 @@ add_dolphin_test(FileUtilTest FileUtilTest.cpp)
 add_dolphin_test(FixedSizeQueueTest FixedSizeQueueTest.cpp)
 add_dolphin_test(FlagTest FlagTest.cpp)
 add_dolphin_test(FloatUtilsTest FloatUtilsTest.cpp)
+add_dolphin_test(FlushThreadTest FlushThreadTest.cpp)
 add_dolphin_test(MathUtilTest MathUtilTest.cpp)
 add_dolphin_test(NandPathsTest NandPathsTest.cpp)
 add_dolphin_test(SettingsHandlerTest SettingsHandlerTest.cpp)

Source/UnitTests/Common/FlushThreadTest.cpp

@@ -0,0 +1,79 @@
+// Copyright 2025 Dolphin Emulator Project
+// SPDX-License-Identifier: GPL-2.0-or-later
+
+#include <gtest/gtest.h>
+
+#include <atomic>
+
+#include "Common/CommonTypes.h"
+#include "Common/FlushThread.h"
+
+TEST(FlushThread, Simple)
+{
+  Common::FlushThread ft;
+
+  std::atomic<int> value = 0;
+
+  ft.Reset("flush", [&] { ++value; });
+
+  // No flush on start.
+  EXPECT_EQ(value.load(), 0);
+
+  ft.SetDirty();
+  ft.WaitForCompletion();
+
+  // One flush.
+  EXPECT_EQ(value.load(), 1);
+
+  ft.Reset("flush", [&] { ++value; });
+
+  // No change after reset.
+  EXPECT_EQ(value.load(), 1);
+
+  ft.Shutdown();
+  ft.SetDirty();
+  ft.WaitForCompletion();
+
+  // No change because shutdown.
+  EXPECT_EQ(value.load(), 1);
+
+  ft.Reset("flush", [&] {
+    ++value;
+    value.notify_one();
+  });
+  ft.WaitForCompletion();
+
+  // Dirty state persits on reset.
+  EXPECT_EQ(value.load(), 2);
+
+  value = 0;
+
+  ft.SetFlushDelay(std::chrono::milliseconds{999999});
+  ft.SetDirty();
+  ft.SetDirty();
+  ft.SetDirty();
+
+  // Not using EXPECT_ here because the tests are technically racey.
+
+  // Probably no flush yet, because of the delay.
+  GTEST_LOG_(INFO) << "Ideally 0: " << value.load();
+
+  const auto start = std::chrono::steady_clock::now();
+  ft.WaitForCompletion();
+  const auto end = std::chrono::steady_clock::now();
+
+  GTEST_LOG_(INFO) << "Ideally 0: "
+                   << duration_cast<std::chrono::milliseconds>(end - start).count();
+
+  // At least one flush happened. Probably just one.
+  EXPECT_GT(value.load(), 0);
+  GTEST_LOG_(INFO) << "Ideally 1: " << value.load();
+
+  value = 0;
+
+  ft.SetDirty();
+  ft.Reset("flush", [] {});
+
+  // Reset first causes a shutdown, so we have an additional immediate flush.
+  EXPECT_EQ(value.load(), 1);
+}

Source/UnitTests/UnitTests.vcxproj

@@ -52,6 +52,7 @@
     <ClCompile Include="Common\FixedSizeQueueTest.cpp" />
     <ClCompile Include="Common\FlagTest.cpp" />
     <ClCompile Include="Common\FloatUtilsTest.cpp" />
+    <ClCompile Include="Common\FlushThreadTest.cpp" />
     <ClCompile Include="Common\MathUtilTest.cpp" />
     <ClCompile Include="Common\NandPathsTest.cpp" />
     <ClCompile Include="Common\SettingsHandlerTest.cpp" />
@@ -118,4 +119,4 @@
     <!--This is only executed via msbuild, VS test runner automatically does this-->
     <Exec Command="$(TargetPath)" />
   </Target>
-</Project>
+</Project>