VideoCommon: revamp asset loading system to: run with multiple worker threads, leverage a priority system so that assets being used recently are preferred, and avoid recursive mutex locking

Source/Core/VideoCommon/Assets/CustomAsset.cpp

@@ -6,8 +6,8 @@
 namespace VideoCommon
 {
 CustomAsset::CustomAsset(std::shared_ptr<CustomAssetLibrary> library,
-                         const CustomAssetLibrary::AssetID& asset_id)
-    : m_owning_library(std::move(library)), m_asset_id(asset_id)
+                         const CustomAssetLibrary::AssetID& asset_id, u64 session_id)
+    : m_owning_library(std::move(library)), m_asset_id(asset_id), m_session_id(session_id)
 {
 }
 
@@ -34,6 +34,11 @@ const CustomAssetLibrary::TimeType& CustomAsset::GetLastLoadedTime() const
   return m_last_loaded_time;
 }
 
+std::size_t CustomAsset::GetSessionId() const
+{
+  return m_session_id;
+}
+
 const CustomAssetLibrary::AssetID& CustomAsset::GetAssetId() const
 {
   return m_asset_id;

Source/Core/VideoCommon/Assets/CustomAsset.h

@@ -18,7 +18,7 @@ class CustomAsset
 {
 public:
   CustomAsset(std::shared_ptr<CustomAssetLibrary> library,
-              const CustomAssetLibrary::AssetID& asset_id);
+              const CustomAssetLibrary::AssetID& asset_id, u64 session_id);
   virtual ~CustomAsset() = default;
   CustomAsset(const CustomAsset&) = delete;
   CustomAsset(CustomAsset&&) = delete;
@@ -39,6 +39,9 @@ public:
   // Returns an id that uniquely identifies this asset
   const CustomAssetLibrary::AssetID& GetAssetId() const;
 
+  // Returns an id that is unique to this session
+  std::size_t GetSessionId() const;
+
   // A rough estimate of how much space this asset
   // will take in memroy
   std::size_t GetByteSizeInMemory() const;
@@ -49,6 +52,7 @@ protected:
 private:
   virtual CustomAssetLibrary::LoadInfo LoadImpl(const CustomAssetLibrary::AssetID& asset_id) = 0;
   CustomAssetLibrary::AssetID m_asset_id;
+  std::size_t m_session_id;
 
   mutable std::mutex m_info_lock;
   std::size_t m_bytes_loaded = 0;

Source/Core/VideoCommon/Assets/CustomAssetLoader.cpp

@@ -4,13 +4,17 @@
 #include "VideoCommon/Assets/CustomAssetLoader.h"
 
 #include "Common/MemoryUtil.h"
+#include "Common/Thread.h"
+
 #include "VideoCommon/Assets/CustomAssetLibrary.h"
+#include "VideoCommon/VideoEvents.h"
 
 namespace VideoCommon
 {
 void CustomAssetLoader::Init()
 {
-  m_asset_monitor_thread_shutdown.Clear();
+  m_frame_event =
+      AfterFrameEvent::Register([this](Core::System&) { OnFrameEnd(); }, "CustomAssetLoader");
 
   const size_t sys_mem = Common::MemPhysical();
   const size_t recommended_min_mem = 2 * size_t(1024 * 1024 * 1024);
@@ -18,65 +22,12 @@ void CustomAssetLoader::Init()
   m_max_memory_available =
       (sys_mem / 2 < recommended_min_mem) ? (sys_mem / 2) : (sys_mem - recommended_min_mem);
 
-  m_asset_monitor_thread = std::thread([this]() {
-    Common::SetCurrentThreadName("Asset monitor");
-    while (true)
-    {
-      if (m_asset_monitor_thread_shutdown.IsSet())
-      {
-        break;
-      }
-
-      std::this_thread::sleep_for(TIME_BETWEEN_ASSET_MONITOR_CHECKS);
-
-      std::lock_guard lk(m_asset_load_lock);
-      for (auto& [asset_id, asset_to_monitor] : m_assets_to_monitor)
-      {
-        if (auto ptr = asset_to_monitor.lock())
-        {
-          const auto write_time = ptr->GetLastWriteTime();
-          if (write_time > ptr->GetLastLoadedTime())
-          {
-            (void)ptr->Load();
-          }
-        }
-      }
-    }
-  });
-
-  m_asset_load_thread.Reset("Custom Asset Loader", [this](std::weak_ptr<CustomAsset> asset) {
-    if (auto ptr = asset.lock())
-    {
-      if (m_memory_exceeded)
-        return;
-
-      if (ptr->Load())
-      {
-        std::lock_guard lk(m_asset_load_lock);
-        const std::size_t asset_memory_size = ptr->GetByteSizeInMemory();
-        m_total_bytes_loaded += asset_memory_size;
-        m_assets_to_monitor.try_emplace(ptr->GetAssetId(), ptr);
-        if (m_total_bytes_loaded > m_max_memory_available)
-        {
-          ERROR_LOG_FMT(VIDEO,
-                        "Asset memory exceeded with asset '{}', future assets won't load until "
-                        "memory is available.",
-                        ptr->GetAssetId());
-          m_memory_exceeded = true;
-        }
-      }
-    }
-  });
+  ResizeWorkerThreads(2);
 }
 
 void CustomAssetLoader ::Shutdown()
 {
-  m_asset_load_thread.Shutdown(true);
-
-  m_asset_monitor_thread_shutdown.Set();
-  m_asset_monitor_thread.join();
-  m_assets_to_monitor.clear();
-  m_total_bytes_loaded = 0;
+  Reset(false);
 }
 
 std::shared_ptr<GameTextureAsset>
@@ -105,4 +56,242 @@ std::shared_ptr<MeshAsset> CustomAssetLoader::LoadMesh(const CustomAssetLibrary:
 {
   return LoadOrCreateAsset<MeshAsset>(asset_id, m_meshes, std::move(library));
 }
+
+void CustomAssetLoader::AssetReferenced(u64 asset_session_id)
+{
+  auto& asset_load_info = m_asset_load_info[asset_session_id];
+  asset_load_info.last_xfb_seen = m_xfbs_seen;
+}
+
+void CustomAssetLoader::Reset(bool restart_worker_threads)
+{
+  const std::size_t worker_thread_count = m_worker_threads.size();
+  StopWorkerThreads();
+
+  m_game_textures.clear();
+  m_pixel_shaders.clear();
+  m_materials.clear();
+  m_meshes.clear();
+
+  m_assetid_to_asset_index.clear();
+  m_asset_load_info.clear();
+
+  {
+    std::lock_guard<std::mutex> guard(m_reload_work_lock);
+    m_assetids_to_reload.clear();
+  }
+
+  {
+    std::lock_guard<std::mutex> guard(m_pending_work_lock);
+    m_pending_work_per_frame.clear();
+  }
+
+  {
+    std::lock_guard<std::mutex> guard(m_completed_work_lock);
+    m_completed_work.clear();
+  }
+
+  if (restart_worker_threads)
+  {
+    StartWorkerThreads(static_cast<u32>(worker_thread_count));
+  }
+}
+
+void CustomAssetLoader::ReloadAsset(const CustomAssetLibrary::AssetID& asset_id)
+{
+  std::lock_guard<std::mutex> guard(m_reload_work_lock);
+  m_assetids_to_reload.push_back(asset_id);
+}
+
+bool CustomAssetLoader::StartWorkerThreads(u32 num_worker_threads)
+{
+  if (num_worker_threads == 0)
+    return true;
+
+  for (u32 i = 0; i < num_worker_threads; i++)
+  {
+    m_worker_thread_start_result.store(false);
+
+    void* thread_param = nullptr;
+    std::thread thr(&CustomAssetLoader::WorkerThreadEntryPoint, this, thread_param);
+    m_init_event.Wait();
+
+    if (!m_worker_thread_start_result.load())
+    {
+      WARN_LOG_FMT(VIDEO, "Failed to start asset load worker thread.");
+      thr.join();
+      break;
+    }
+
+    m_worker_threads.push_back(std::move(thr));
+  }
+
+  return HasWorkerThreads();
+}
+
+bool CustomAssetLoader::ResizeWorkerThreads(u32 num_worker_threads)
+{
+  if (m_worker_threads.size() == num_worker_threads)
+    return true;
+
+  StopWorkerThreads();
+  return StartWorkerThreads(num_worker_threads);
+}
+
+bool CustomAssetLoader::HasWorkerThreads() const
+{
+  return !m_worker_threads.empty();
+}
+
+void CustomAssetLoader::StopWorkerThreads()
+{
+  if (!HasWorkerThreads())
+    return;
+
+  // Signal worker threads to stop, and wake all of them.
+  {
+    std::lock_guard<std::mutex> guard(m_pending_work_lock);
+    m_exit_flag.Set();
+    m_worker_thread_wake.notify_all();
+  }
+
+  // Wait for worker threads to exit.
+  for (std::thread& thr : m_worker_threads)
+    thr.join();
+  m_worker_threads.clear();
+  m_exit_flag.Clear();
+}
+
+void CustomAssetLoader::WorkerThreadEntryPoint(void* param)
+{
+  Common::SetCurrentThreadName("Asset Loader Worker");
+
+  m_worker_thread_start_result.store(true);
+  m_init_event.Set();
+
+  WorkerThreadRun();
+}
+
+void CustomAssetLoader::WorkerThreadRun()
+{
+  std::unique_lock<std::mutex> pending_lock(m_pending_work_lock);
+  while (!m_exit_flag.IsSet())
+  {
+    m_worker_thread_wake.wait(pending_lock);
+
+    while (!m_pending_work_per_frame.empty() && !m_exit_flag.IsSet())
+    {
+      m_busy_workers++;
+
+      auto pending_iter = m_pending_work_per_frame.begin();
+      auto item(std::move(pending_iter->second));
+      m_pending_work_per_frame.erase(pending_iter);
+
+      const auto item_shared = item.lock();
+      pending_lock.unlock();
+
+      if (item_shared && m_last_frame_total_loaded_memory < m_max_memory_available)
+      {
+        if (item_shared->Load())
+        {
+          // This asset could be double counted, but will be corected on the next frame
+          m_last_frame_total_loaded_memory += item_shared->GetByteSizeInMemory();
+        }
+
+        // Regardless of whether the load was successful or not
+        // mark it as complete
+        {
+          std::lock_guard<std::mutex> completed_guard(m_completed_work_lock);
+          m_completed_work.push_back(item_shared->GetSessionId());
+        }
+      }
+
+      pending_lock.lock();
+      m_busy_workers--;
+    }
+  }
+}
+
+void CustomAssetLoader::OnFrameEnd()
+{
+  std::vector<CustomAssetLibrary::AssetID> assetids_to_reload;
+  {
+    std::lock_guard<std::mutex> guard(m_reload_work_lock);
+    m_assetids_to_reload.swap(assetids_to_reload);
+  }
+  for (const CustomAssetLibrary::AssetID& asset_id : assetids_to_reload)
+  {
+    if (const auto asset_session_id_iter = m_assetid_to_asset_index.find(asset_id);
+        asset_session_id_iter != m_assetid_to_asset_index.end())
+    {
+      auto& asset_load_info = m_asset_load_info[asset_session_id_iter->second];
+      asset_load_info.xfb_load_request = m_xfbs_seen;
+    }
+  }
+
+  std::vector<std::size_t> completed_work;
+  {
+    std::lock_guard<std::mutex> guard(m_completed_work_lock);
+    m_completed_work.swap(completed_work);
+  }
+  for (std::size_t completed_index : completed_work)
+  {
+    auto& asset_load_info = m_asset_load_info[completed_index];
+
+    // If we had a load request and it wasn't from this frame, clear it
+    if (asset_load_info.xfb_load_request && asset_load_info.xfb_load_request != m_xfbs_seen)
+    {
+      asset_load_info.xfb_load_request = std::nullopt;
+    }
+  }
+
+  m_xfbs_seen++;
+
+  std::size_t total_bytes_loaded = 0;
+
+  // Build up the work prioritizing newest requested assets first
+  PendingWorkContainer new_pending_work;
+  for (const auto& asset_load_info : m_asset_load_info)
+  {
+    if (const auto asset = asset_load_info.asset.lock())
+    {
+      total_bytes_loaded += asset->GetByteSizeInMemory();
+      if (total_bytes_loaded > m_max_memory_available)
+      {
+        if (!m_memory_exceeded)
+        {
+          m_memory_exceeded = true;
+          ERROR_LOG_FMT(VIDEO,
+                        "Asset memory exceeded with asset '{}', future assets won't load until "
+                        "memory is available.",
+                        asset->GetAssetId());
+        }
+        break;
+      }
+      if (asset_load_info.xfb_load_request)
+      {
+        new_pending_work.emplace(asset_load_info.last_xfb_seen, asset_load_info.asset);
+      }
+    }
+  }
+
+  if (m_memory_exceeded && total_bytes_loaded < m_max_memory_available)
+  {
+    INFO_LOG_FMT(VIDEO, "Asset memory went below limit, new assets can begin loading.");
+    m_memory_exceeded = false;
+  }
+
+  m_last_frame_total_loaded_memory = total_bytes_loaded;
+
+  if (new_pending_work.empty())
+    return;
+
+  // Now notify our workers
+  {
+    std::lock_guard<std::mutex> guard(m_pending_work_lock);
+    std::swap(m_pending_work_per_frame, new_pending_work);
+    m_worker_thread_wake.notify_all();
+  }
+}
+
 }  // namespace VideoCommon

Source/Core/VideoCommon/Assets/CustomAssetLoader.h

@@ -4,19 +4,24 @@
 #pragma once
 
 #include <chrono>
+#include <deque>
 #include <map>
 #include <memory>
 #include <mutex>
+#include <optional>
 #include <thread>
+#include <vector>
 
+#include "Common/Event.h"
 #include "Common/Flag.h"
+#include "Common/HookableEvent.h"
 #include "Common/Logging/Log.h"
-#include "Common/WorkQueueThread.h"
 #include "VideoCommon/Assets/CustomAsset.h"
 #include "VideoCommon/Assets/MaterialAsset.h"
 #include "VideoCommon/Assets/MeshAsset.h"
 #include "VideoCommon/Assets/ShaderAsset.h"
 #include "VideoCommon/Assets/TextureAsset.h"
+#include "VideoCommon/Present.h"
 
 namespace VideoCommon
 {
@@ -52,6 +57,14 @@ public:
   std::shared_ptr<MeshAsset> LoadMesh(const CustomAssetLibrary::AssetID& asset_id,
                                       std::shared_ptr<CustomAssetLibrary> library);
 
+  // Notifies the asset system that an asset has been used
+  void AssetReferenced(u64 asset_session_id);
+
+  // Requests that an asset that exists be reloaded
+  void ReloadAsset(const CustomAssetLibrary::AssetID& asset_id);
+
+  void Reset(bool restart_worker_threads = true);
+
 private:
   // TODO C++20: use a 'derived_from' concept against 'CustomAsset' when available
   template <typename AssetType>
@@ -65,44 +78,88 @@ private:
     {
       auto shared = it->second.lock();
       if (shared)
+      {
+        auto& asset_load_info = m_asset_load_info[shared->GetSessionId()];
+        asset_load_info.last_xfb_seen = m_xfbs_seen;
+        asset_load_info.xfb_load_request = m_xfbs_seen;
         return shared;
       }
-    std::shared_ptr<AssetType> ptr(new AssetType(std::move(library), asset_id), [&](AssetType* a) {
-      {
-        std::lock_guard lk(m_asset_load_lock);
-        m_total_bytes_loaded -= a->GetByteSizeInMemory();
-        m_assets_to_monitor.erase(a->GetAssetId());
-        if (m_max_memory_available >= m_total_bytes_loaded && m_memory_exceeded)
-        {
-          INFO_LOG_FMT(VIDEO, "Asset memory went below limit, new assets can begin loading.");
-          m_memory_exceeded = false;
     }
+
+    const auto [index_it, index_inserted] = m_assetid_to_asset_index.try_emplace(asset_id, 0);
+    if (index_inserted)
+    {
+      index_it->second = m_asset_load_info.size();
     }
-      delete a;
-    });
+
+    auto ptr = std::make_shared<AssetType>(std::move(library), asset_id, index_it->second);
     it->second = ptr;
-    m_asset_load_thread.Push(it->second);
+
+    AssetLoadInfo* asset_load_info;
+    if (index_inserted)
+    {
+      m_asset_load_info.emplace_back();
+      asset_load_info = &m_asset_load_info.back();
+    }
+    else
+    {
+      asset_load_info = &m_asset_load_info[index_it->second];
+    }
+
+    asset_load_info->asset = ptr;
+    asset_load_info->last_xfb_seen = m_xfbs_seen;
+    asset_load_info->xfb_load_request = m_xfbs_seen;
     return ptr;
   }
 
-  static constexpr auto TIME_BETWEEN_ASSET_MONITOR_CHECKS = std::chrono::milliseconds{500};
+  bool StartWorkerThreads(u32 num_worker_threads);
+  bool ResizeWorkerThreads(u32 num_worker_threads);
+  bool HasWorkerThreads() const;
+  void StopWorkerThreads();
+
+  void WorkerThreadEntryPoint(void* param);
+  void WorkerThreadRun();
+
+  void OnFrameEnd();
+
+  Common::EventHook m_frame_event;
 
   std::map<CustomAssetLibrary::AssetID, std::weak_ptr<GameTextureAsset>> m_game_textures;
   std::map<CustomAssetLibrary::AssetID, std::weak_ptr<PixelShaderAsset>> m_pixel_shaders;
   std::map<CustomAssetLibrary::AssetID, std::weak_ptr<MaterialAsset>> m_materials;
   std::map<CustomAssetLibrary::AssetID, std::weak_ptr<MeshAsset>> m_meshes;
-  std::thread m_asset_monitor_thread;
-  Common::Flag m_asset_monitor_thread_shutdown;
 
-  std::size_t m_total_bytes_loaded = 0;
+  std::map<CustomAssetLibrary::AssetID, std::size_t> m_assetid_to_asset_index;
+
+  struct AssetLoadInfo
+  {
+    std::weak_ptr<CustomAsset> asset;
+    u64 last_xfb_seen = 0;
+    std::optional<u64> xfb_load_request;
+  };
+  std::vector<AssetLoadInfo> m_asset_load_info;
+  u64 m_xfbs_seen = 0;
+
   std::size_t m_max_memory_available = 0;
-  std::atomic_bool m_memory_exceeded = false;
+  bool m_memory_exceeded = false;
+  std::atomic_size_t m_last_frame_total_loaded_memory = 0;
 
-  std::map<CustomAssetLibrary::AssetID, std::weak_ptr<CustomAsset>> m_assets_to_monitor;
+  Common::Flag m_exit_flag;
+  Common::Event m_init_event;
 
-  // Use a recursive mutex to handle the scenario where an asset goes out of scope while
-  // iterating over the assets to monitor which calls the lock above in 'LoadOrCreateAsset'
-  std::recursive_mutex m_asset_load_lock;
-  Common::WorkQueueThread<std::weak_ptr<CustomAsset>> m_asset_load_thread;
+  std::vector<std::thread> m_worker_threads;
+  std::atomic_bool m_worker_thread_start_result{false};
+
+  using PendingWorkContainer = std::multimap<u64, std::weak_ptr<CustomAsset>, std::greater<>>;
+  PendingWorkContainer m_pending_work_per_frame;
+  std::mutex m_pending_work_lock;
+  std::condition_variable m_worker_thread_wake;
+  std::atomic_size_t m_busy_workers{0};
+
+  std::vector<std::size_t> m_completed_work;
+  std::mutex m_completed_work_lock;
+
+  std::vector<CustomAssetLibrary::AssetID> m_assetids_to_reload;
+  std::mutex m_reload_work_lock;
 };
 }  // namespace VideoCommon

Source/Core/VideoCommon/GraphicsModSystem/Runtime/CustomPipeline.cpp

@@ -182,6 +182,10 @@ void CustomPipeline::UpdatePixelData(
   {
     m_pixel_material.m_asset = loader.LoadMaterial(material_to_load, library);
   }
+  else
+  {
+    loader.AssetReferenced(m_pixel_material.m_asset->GetSessionId());
+  }
 
   const auto material_data = m_pixel_material.m_asset->GetData();
   if (!material_data)
@@ -217,6 +221,10 @@ void CustomPipeline::UpdatePixelData(
 
     m_last_generated_shader_code = ShaderCode{};
   }
+  else
+  {
+    loader.AssetReferenced(m_pixel_shader.m_asset->GetSessionId());
+  }
 
   const auto shader_data = m_pixel_shader.m_asset->GetData();
   if (!shader_data)

Source/Core/VideoCommon/TextureCacheBase.cpp

@@ -36,6 +36,7 @@
 #include "VideoCommon/AbstractFramebuffer.h"
 #include "VideoCommon/AbstractGfx.h"
 #include "VideoCommon/AbstractStagingTexture.h"
+#include "VideoCommon/Assets/CustomAssetLoader.h"
 #include "VideoCommon/Assets/CustomTextureData.h"
 #include "VideoCommon/BPMemory.h"
 #include "VideoCommon/FramebufferManager.h"
@@ -263,10 +264,13 @@ void TextureCacheBase::SetBackupConfig(const VideoConfig& config)
 
 bool TextureCacheBase::DidLinkedAssetsChange(const TCacheEntry& entry)
 {
+  auto& system = Core::System::GetInstance();
+  auto& loader = system.GetCustomAssetLoader();
   for (const auto& cached_asset : entry.linked_game_texture_assets)
   {
     if (cached_asset.m_asset)
     {
+      loader.AssetReferenced(cached_asset.m_asset->GetSessionId());
       if (cached_asset.m_asset->GetLastLoadedTime() > cached_asset.m_cached_write_time)
         return true;
     }