CDROMAsyncReader: Use persistent thread

src/core/cdrom.cpp

@@ -568,10 +568,7 @@ void CDROM::Initialize()
 {
   s_state.disc_region = DiscRegion::NonPS1;
   s_state.manual_lid_control = MANUAL_LID_CONTROL_DISABLED;
-
-  if (g_settings.cdrom_readahead_sectors > 0)
-    CDROMAsyncReader::StartThread(g_settings.cdrom_readahead_sectors);
-
+  CDROMAsyncReader::SetReadaheadSectors(g_settings.cdrom_readahead_sectors);
   Reset();
 }
 
@@ -585,7 +582,6 @@ void CDROM::Shutdown()
   s_state.async_interrupt_event.Deactivate();
   s_state.command_second_response_event.Deactivate();
   s_state.command_event.Deactivate();
-  CDROMAsyncReader::StopThread();
   CDROMAsyncReader::RemoveMedia();
 }
 
@@ -1115,15 +1111,10 @@ TinyString CDROM::LBAToMSFString(CDImage::LBA lba)
 
 void CDROM::SetReadaheadSectors(u32 readahead_sectors)
 {
-  const bool want_thread = (readahead_sectors > 0);
-  if (want_thread == CDROMAsyncReader::IsUsingThread() && CDROMAsyncReader::GetReadaheadCount() == readahead_sectors)
+  if (CDROMAsyncReader::GetReadaheadCount() == readahead_sectors)
     return;
 
-  if (want_thread)
-    CDROMAsyncReader::StartThread(readahead_sectors);
-  else
-    CDROMAsyncReader::StopThread();
-
+  CDROMAsyncReader::SetReadaheadSectors(readahead_sectors);
   if (HasMedia())
     CDROMAsyncReader::QueueReadSector(s_state.requested_lba);
 }

src/core/cdrom_async_reader.cpp

@@ -2,8 +2,10 @@
 // SPDX-License-Identifier: CC-BY-NC-ND-4.0
 
 #include "cdrom_async_reader.h"
+#include "host.h"
 
 #include "common/assert.h"
+#include "common/error.h"
 #include "common/log.h"
 #include "common/threading.h"
 #include "common/timer.h"
@@ -12,6 +14,7 @@ LOG_CHANNEL(CDROMAsyncReader);
 
 namespace CDROMAsyncReader {
 
+static bool IsReadaheadEnabled();
 static void EmptyBuffers();
 static bool ReadSectorIntoBuffer(std::unique_lock<Threading::Mutex>& lock);
 static void ReadSectorNonThreaded(CDImage::LBA lba);
@@ -26,17 +29,19 @@ struct Locals
 
   std::atomic<CDImage::LBA> next_position{};
   std::atomic_bool next_position_set{false};
-  std::atomic_bool shutdown_flag{true};
+  std::atomic_bool shutdown_flag{false};
+  std::atomic_bool readahead_enabled{false};
 
   std::atomic_bool is_reading{false};
   std::atomic_bool can_readahead{false};
   std::atomic_bool seek_error{false};
 
-  std::vector<BufferSlot> buffers;
   std::atomic<u32> buffer_front{0};
   std::atomic<u32> buffer_back{0};
   std::atomic<u32> buffer_count{0};
 
+  std::vector<BufferSlot> buffers;
+
   Threading::Mutex mutex;
   Threading::ConditionVariable do_read_cv;
   Threading::ConditionVariable notify_read_complete_cv;
@@ -50,32 +55,32 @@ ALIGN_TO_CACHE_LINE static Locals s_locals;
 
 CDImage::LBA CDROMAsyncReader::GetLastReadSector()
 {
-  return s_locals.buffers[s_locals.buffer_front.load()].lba;
+  return s_locals.buffers[s_locals.buffer_front.load(std::memory_order_acquire)].lba;
 }
 
 const CDROMAsyncReader::SectorBuffer& CDROMAsyncReader::GetSectorBuffer()
 {
-  return s_locals.buffers[s_locals.buffer_front.load()].data;
+  return s_locals.buffers[s_locals.buffer_front.load(std::memory_order_acquire)].data;
 }
 
 const CDImage::SubChannelQ& CDROMAsyncReader::GetSectorSubQ()
 {
-  return s_locals.buffers[s_locals.buffer_front.load()].subq;
+  return s_locals.buffers[s_locals.buffer_front.load(std::memory_order_acquire)].subq;
 }
 
 u32 CDROMAsyncReader::GetBufferedSectorCount()
 {
-  return s_locals.buffer_count.load();
+  return s_locals.buffer_count.load(std::memory_order_acquire);
 }
 
 bool CDROMAsyncReader::HasBufferedSectors()
 {
-  return (s_locals.buffer_count.load() > 0);
+  return (s_locals.buffer_count.load(std::memory_order_acquire) > 0);
 }
 
 u32 CDROMAsyncReader::GetReadaheadCount()
 {
-  return static_cast<u32>(s_locals.buffers.size());
+  return s_locals.readahead_enabled.load(std::memory_order_relaxed) ? static_cast<u32>(s_locals.buffers.size()) : 0;
 }
 
 bool CDROMAsyncReader::HasMedia()
@@ -93,54 +98,47 @@ const std::string& CDROMAsyncReader::GetMediaPath()
   return s_locals.media->GetPath();
 }
 
-bool CDROMAsyncReader::IsUsingThread()
-{
-  return s_locals.read_thread.Joinable();
-}
-
-void CDROMAsyncReader::StartThread(u32 readahead_count)
+bool CDROMAsyncReader::ProcessStartup(Error* error)
 {
-  if (IsUsingThread())
-    StopThread();
-
-  s_locals.buffers.clear();
-  s_locals.buffers.resize(readahead_count);
-  EmptyBuffers();
+  if (!s_locals.read_thread.Start(&CDROMAsyncReader::WorkerThreadEntryPoint))
+  {
+    Error::SetStringView(error, "Failed to start CDROMAsyncReader thread");
+    return false;
+  }
 
-  s_locals.shutdown_flag.store(false);
-  s_locals.read_thread.Start(&CDROMAsyncReader::WorkerThreadEntryPoint);
-  INFO_LOG("Read thread started with readahead of {} sectors", readahead_count);
+  return true;
 }
 
-void CDROMAsyncReader::StopThread()
+void CDROMAsyncReader::ProcessShutdown()
 {
-  if (!IsUsingThread())
-    return;
-
   {
-    std::unique_lock lock(s_locals.mutex);
-    s_locals.shutdown_flag.store(true);
+    std::lock_guard lock(s_locals.mutex);
+    s_locals.shutdown_flag.store(true, std::memory_order_release);
     s_locals.do_read_cv.notify_one();
   }
 
   s_locals.read_thread.Join();
+}
+
+void CDROMAsyncReader::SetReadaheadSectors(u32 readahead_sectors)
+{
+  CancelReadahead();
   EmptyBuffers();
   s_locals.buffers.clear();
+  s_locals.buffers.resize(std::max(readahead_sectors, 1u));
+  s_locals.readahead_enabled.store(readahead_sectors >= 1, std::memory_order_relaxed);
+  DEV_LOG("Readahead set to {} sectors", readahead_sectors);
 }
 
 void CDROMAsyncReader::SetMedia(std::unique_ptr<CDImage> media)
 {
-  if (IsUsingThread())
-    CancelReadahead();
-
+  CancelReadahead();
   s_locals.media = std::move(media);
 }
 
 std::unique_ptr<CDImage> CDROMAsyncReader::RemoveMedia()
 {
-  if (IsUsingThread())
-    CancelReadahead();
-
+  CancelReadahead();
   return std::move(s_locals.media);
 }
 
@@ -183,55 +181,62 @@ bool CDROMAsyncReader::Precache(ProgressCallback* callback, Error* error)
 
 void CDROMAsyncReader::QueueReadSector(CDImage::LBA lba)
 {
-  if (!IsUsingThread())
+  if (!IsReadaheadEnabled())
   {
     ReadSectorNonThreaded(lba);
     return;
   }
 
-  const u32 buffer_count = s_locals.buffer_count.load();
-  if (buffer_count > 0)
+  std::unique_lock lock(s_locals.mutex);
+
+  // If there's already a seek pending, the existing buffers belong to the
+  // old position and must not satisfy this newer request.
+  if (!s_locals.next_position_set.load(std::memory_order_relaxed))
   {
-    // don't re-read the same sector if it was the last one we read
-    // the CDC code does this when seeking->reading
-    const u32 buffer_front = s_locals.buffer_front.load();
-    if (s_locals.buffers[buffer_front].lba == lba)
+    const u32 buffer_count = s_locals.buffer_count.load(std::memory_order_acquire);
+    if (buffer_count > 0)
     {
-      DEBUG_LOG("Skipping re-reading same sector {}", lba);
-      return;
-    }
+      // don't re-read the same sector if it was the last one we read
+      // the CDC code does this when seeking->reading
+      const u32 buffer_front = s_locals.buffer_front.load(std::memory_order_acquire);
+      if (s_locals.buffers[buffer_front].lba == lba)
+      {
+        DEBUG_LOG("Skipping re-reading same sector {}", lba);
+        return;
+      }
 
-    // did we readahead to the correct sector?
-    const u32 next_buffer = (buffer_front + 1) % static_cast<u32>(s_locals.buffers.size());
-    if (s_locals.buffer_count > 1 && s_locals.buffers[next_buffer].lba == lba)
-    {
-      // great, don't need a seek, but still kick the thread to start reading ahead again
-      DEBUG_LOG("Readahead buffer hit for sector {}", lba);
-      s_locals.buffer_front.store(next_buffer);
-      s_locals.buffer_count.fetch_sub(1);
-      s_locals.can_readahead.store(true);
-      s_locals.do_read_cv.notify_one();
-      return;
+      // did we readahead to the correct sector?
+      const u32 next_buffer = (buffer_front + 1) % static_cast<u32>(s_locals.buffers.size());
+      if (buffer_count > 1 && s_locals.buffers[next_buffer].lba == lba)
+      {
+        // great, don't need a seek, but still kick the thread to start reading ahead again
+        DEBUG_LOG("Readahead buffer hit for sector {}", lba);
+        s_locals.buffer_front.store(next_buffer, std::memory_order_release);
+        s_locals.buffer_count.fetch_sub(1, std::memory_order_release);
+        s_locals.can_readahead.store(true, std::memory_order_release);
+        s_locals.do_read_cv.notify_one();
+        return;
+      }
     }
   }
 
-  // we need to toss away our readahead and start fresh
+  // We either missed, or another seek was already queued.
+  // Replace the pending position so the newest request wins.
   DEBUG_LOG("Readahead buffer miss, queueing seek to {}", lba);
-  std::unique_lock lock(s_locals.mutex);
-  s_locals.next_position_set.store(true);
-  s_locals.next_position = lba;
+  s_locals.next_position.store(lba, std::memory_order_relaxed);
+  s_locals.next_position_set.store(true, std::memory_order_release);
   s_locals.do_read_cv.notify_one();
 }
 
 bool CDROMAsyncReader::ReadSectorUncached(CDImage::LBA lba, CDImage::SubChannelQ* subq, SectorBuffer* data)
 {
-  if (!IsUsingThread())
+  if (!IsReadaheadEnabled())
     return InternalReadSectorUncached(lba, subq, data);
 
   std::unique_lock lock(s_locals.mutex);
 
   // wait until the read thread is idle
-  s_locals.notify_read_complete_cv.wait(lock, []() { return !s_locals.is_reading.load(); });
+  s_locals.notify_read_complete_cv.wait(lock, []() { return !s_locals.is_reading.load(std::memory_order_acquire); });
 
   // read while the lock is held so it has to wait
   const CDImage::LBA prev_lba = s_locals.media->GetPositionOnDisc();
@@ -239,7 +244,7 @@ bool CDROMAsyncReader::ReadSectorUncached(CDImage::LBA lba, CDImage::SubChannelQ
   if (!s_locals.media->Seek(prev_lba)) [[unlikely]]
   {
     ERROR_LOG("Failed to re-seek to cached position {}", prev_lba);
-    s_locals.can_readahead.store(false);
+    s_locals.can_readahead.store(false, std::memory_order_release);
   }
 
   return result;
@@ -264,11 +269,12 @@ bool CDROMAsyncReader::InternalReadSectorUncached(CDImage::LBA lba, CDImage::Sub
 
 bool CDROMAsyncReader::WaitForReadToComplete()
 {
-  // Safe without locking with memory_order_seq_cst.
-  if (!s_locals.next_position_set.load() && s_locals.buffer_count.load() > 0)
+  if (!s_locals.next_position_set.load(std::memory_order_acquire) &&
+      s_locals.buffer_count.load(std::memory_order_acquire) > 0)
   {
-    TRACE_LOG("Returning sector {}", s_locals.buffers[s_locals.buffer_front.load()].lba);
-    return s_locals.buffers[s_locals.buffer_front.load()].result;
+    const u32 buffer_index = s_locals.buffer_front.load(std::memory_order_acquire);
+    TRACE_LOG("Returning sector {}", s_locals.buffers[buffer_index].lba);
+    return s_locals.buffers[buffer_index].result;
   }
 
   Timer wait_timer;
@@ -276,15 +282,17 @@ bool CDROMAsyncReader::WaitForReadToComplete()
 
   std::unique_lock lock(s_locals.mutex);
   s_locals.notify_read_complete_cv.wait(lock, []() {
-    return (s_locals.buffer_count.load() > 0 || s_locals.seek_error.load()) && !s_locals.next_position_set.load();
+    return (s_locals.buffer_count.load(std::memory_order_acquire) > 0 ||
+            s_locals.seek_error.load(std::memory_order_acquire)) &&
+           !s_locals.next_position_set.load(std::memory_order_acquire);
   });
-  if (s_locals.seek_error.load()) [[unlikely]]
+  if (s_locals.seek_error.load(std::memory_order_acquire)) [[unlikely]]
   {
-    s_locals.seek_error.store(false);
+    s_locals.seek_error.store(false, std::memory_order_release);
     return false;
   }
 
-  const u32 front = s_locals.buffer_front.load();
+  const u32 front = s_locals.buffer_front.load(std::memory_order_acquire);
   const double wait_time = wait_timer.GetTimeMilliseconds();
   if (wait_time > 1.0f) [[unlikely]]
     WARNING_LOG("Had to wait {:.2f} msec for LBA {}", wait_time, s_locals.buffers[front].lba);
@@ -295,36 +303,46 @@ bool CDROMAsyncReader::WaitForReadToComplete()
 
 void CDROMAsyncReader::WaitForIdle()
 {
-  if (!IsUsingThread())
+  if (!IsReadaheadEnabled())
     return;
 
   std::unique_lock lock(s_locals.mutex);
-  s_locals.notify_read_complete_cv.wait(
-    lock, []() { return (!s_locals.is_reading.load() && !s_locals.next_position_set.load()); });
+  s_locals.notify_read_complete_cv.wait(lock, []() {
+    return (!s_locals.is_reading.load(std::memory_order_acquire) &&
+            !s_locals.next_position_set.load(std::memory_order_acquire));
+  });
+}
+
+bool CDROMAsyncReader::IsReadaheadEnabled()
+{
+  // NOTE: Not called on worker thread.
+  DebugAssert(Host::IsOnCoreThread());
+  return s_locals.readahead_enabled.load(std::memory_order_relaxed);
 }
 
 void CDROMAsyncReader::EmptyBuffers()
 {
-  s_locals.buffer_front.store(0);
-  s_locals.buffer_back.store(0);
-  s_locals.buffer_count.store(0);
+  s_locals.buffer_front.store(0, std::memory_order_release);
+  s_locals.buffer_back.store(0, std::memory_order_release);
+  s_locals.buffer_count.store(0, std::memory_order_release);
 }
 
 bool CDROMAsyncReader::ReadSectorIntoBuffer(std::unique_lock<Threading::Mutex>& lock)
 {
   Timer timer;
 
-  const u32 slot = s_locals.buffer_back.load();
+  const u32 slot = s_locals.buffer_back.load(std::memory_order_acquire);
   s_locals.buffer_back.store((slot + 1) % static_cast<u32>(s_locals.buffers.size()));
 
   BufferSlot& buffer = s_locals.buffers[slot];
   buffer.lba = s_locals.media->GetPositionOnDisc();
-  s_locals.is_reading.store(true);
+  s_locals.is_reading.store(true, std::memory_order_release);
   lock.unlock();
 
   TRACE_LOG("Reading LBA {}...", buffer.lba);
 
-  buffer.result = s_locals.media->ReadRawSector(buffer.data.data(), &buffer.subq);
+  const bool read_result = s_locals.media->ReadRawSector(buffer.data.data(), &buffer.subq);
+  buffer.result = read_result;
   if (buffer.result) [[likely]]
   {
     const double read_time = timer.GetTimeMilliseconds();
@@ -337,24 +355,24 @@ bool CDROMAsyncReader::ReadSectorIntoBuffer(std::unique_lock<Threading::Mutex>&
   }
 
   lock.lock();
-  s_locals.is_reading.store(false);
-  s_locals.buffer_count.fetch_add(1);
+  s_locals.is_reading.store(false, std::memory_order_release);
+  s_locals.buffer_count.fetch_add(1, std::memory_order_release);
   s_locals.notify_read_complete_cv.notify_all();
-  return true;
+  return read_result;
 }
 
 void CDROMAsyncReader::ReadSectorNonThreaded(CDImage::LBA lba)
 {
   Timer timer;
 
-  s_locals.buffers.resize(1);
-  s_locals.seek_error.store(false);
+  Assert(!s_locals.buffers.empty());
+  s_locals.seek_error.store(false, std::memory_order_release);
   EmptyBuffers();
 
   if (s_locals.media->GetPositionOnDisc() != lba && !s_locals.media->Seek(lba))
   {
     WARNING_LOG("Seek to LBA {} failed", lba);
-    s_locals.seek_error.store(true);
+    s_locals.seek_error.store(true, std::memory_order_release);
     return;
   }
 
@@ -375,7 +393,7 @@ void CDROMAsyncReader::ReadSectorNonThreaded(CDImage::LBA lba)
     ERROR_LOG("Read of LBA {} failed", buffer.lba);
   }
 
-  s_locals.buffer_count.fetch_add(1);
+  s_locals.buffer_count.fetch_add(1, std::memory_order_release);
 }
 
 void CDROMAsyncReader::CancelReadahead()
@@ -385,35 +403,42 @@ void CDROMAsyncReader::CancelReadahead()
   std::unique_lock lock(s_locals.mutex);
 
   // wait until the read thread is idle
-  s_locals.notify_read_complete_cv.wait(lock, []() { return !s_locals.is_reading.load(); });
+  s_locals.notify_read_complete_cv.wait(lock, []() {
+    return !s_locals.is_reading.load(std::memory_order_acquire) &&
+           !s_locals.next_position_set.load(std::memory_order_acquire);
+  });
 
   // prevent it from doing any more when it re-acquires the lock
-  s_locals.can_readahead.store(false);
+  s_locals.can_readahead.store(false, std::memory_order_release);
   EmptyBuffers();
 }
 
 void CDROMAsyncReader::WorkerThreadEntryPoint()
 {
+  Threading::SetNameOfCurrentThread("CDROM Async Reader");
+
   std::unique_lock lock(s_locals.mutex);
 
   for (;;)
   {
     s_locals.do_read_cv.wait(lock, []() {
-      return (s_locals.shutdown_flag.load() || s_locals.next_position_set.load() || s_locals.can_readahead.load());
+      return (s_locals.shutdown_flag.load(std::memory_order_acquire) ||
+              s_locals.next_position_set.load(std::memory_order_acquire) ||
+              s_locals.can_readahead.load(std::memory_order_acquire));
     });
-    if (s_locals.shutdown_flag.load())
+    if (s_locals.shutdown_flag.load(std::memory_order_relaxed))
       break;
 
     for (;;)
     {
-      if (s_locals.next_position_set.load())
+      if (s_locals.next_position_set.load(std::memory_order_acquire))
       {
         // discard buffers, we're seeking to a new location
-        const CDImage::LBA seek_location = s_locals.next_position.load();
+        const CDImage::LBA seek_location = s_locals.next_position.load(std::memory_order_relaxed);
         EmptyBuffers();
-        s_locals.next_position_set.store(false);
-        s_locals.seek_error.store(false);
-        s_locals.is_reading.store(true);
+        s_locals.next_position_set.store(false, std::memory_order_release);
+        s_locals.seek_error.store(false, std::memory_order_release);
+        s_locals.is_reading.store(true, std::memory_order_release);
         lock.unlock();
 
         // seek without lock held in case it takes time
@@ -422,10 +447,10 @@ void CDROMAsyncReader::WorkerThreadEntryPoint()
           (s_locals.media->GetPositionOnDisc() == seek_location || s_locals.media->Seek(seek_location));
 
         lock.lock();
-        s_locals.is_reading.store(false);
+        s_locals.is_reading.store(false, std::memory_order_release);
 
         // did another request come in? abort if so
-        if (s_locals.next_position_set.load())
+        if (s_locals.next_position_set.load(std::memory_order_acquire))
           continue;
 
         // did we fail the seek?
@@ -433,24 +458,24 @@ void CDROMAsyncReader::WorkerThreadEntryPoint()
         {
           // add the error result, and don't try to read ahead
           WARNING_LOG("Seek to LBA {} failed", seek_location);
-          s_locals.seek_error.store(true);
+          s_locals.seek_error.store(true, std::memory_order_release);
           s_locals.notify_read_complete_cv.notify_all();
           break;
         }
 
         // go go read ahead!
-        s_locals.can_readahead.store(true);
+        s_locals.can_readahead.store(true, std::memory_order_release);
       }
 
-      if (!s_locals.can_readahead.load())
+      if (!s_locals.can_readahead.load(std::memory_order_acquire))
         break;
 
       // readahead time! read as many sectors as we have space for
       DEBUG_LOG("Reading ahead {} sectors...",
-                static_cast<u32>(s_locals.buffers.size()) - s_locals.buffer_count.load());
-      while (s_locals.buffer_count.load() < static_cast<u32>(s_locals.buffers.size()))
+                static_cast<u32>(s_locals.buffers.size()) - s_locals.buffer_count.load(std::memory_order_acquire));
+      while (s_locals.buffer_count.load(std::memory_order_acquire) < static_cast<u32>(s_locals.buffers.size()))
       {
-        if (s_locals.next_position_set.load())
+        if (s_locals.next_position_set.load(std::memory_order_acquire))
         {
           // a seek request came in while we're reading, so bail out
           break;
@@ -462,7 +487,7 @@ void CDROMAsyncReader::WorkerThreadEntryPoint()
       }
 
       // readahead buffer is full or errored at this point
-      s_locals.can_readahead.store(false);
+      s_locals.can_readahead.store(false, std::memory_order_release);
       break;
     }
   }

src/core/cdrom_async_reader.h

@@ -36,11 +36,10 @@ bool HasMedia();
 CDImage* GetMedia();
 const std::string& GetMediaPath();
 
-// TODO: FIXME: Make global shutdown
-bool IsUsingThread();
-void StartThread(u32 readahead_count = 8);
-void StopThread();
+bool ProcessStartup(Error* error);
+void ProcessShutdown();
 
+void SetReadaheadSectors(u32 readahead_sectors);
 void SetMedia(std::unique_ptr<CDImage> media);
 std::unique_ptr<CDImage> RemoveMedia();
 

src/core/core.cpp

@@ -2,9 +2,10 @@
 // SPDX-License-Identifier: CC-BY-NC-ND-4.0
 
 #include "core.h"
-#include "cheats.h"
 #include "achievements.h"
 #include "achievements_private.h"
+#include "cdrom_async_reader.h"
+#include "cheats.h"
 #include "core_private.h"
 #include "discord_presence.h"
 #include "gdb_server.h"
@@ -744,8 +745,9 @@ bool Core::CoreThreadInitialize(bool disable_worker_threads, Error* error)
 
   LogStartupInformation();
 
-  if (!VideoThread::ProcessStartup(error)) [[unlikely]]
+  if (!VideoThread::ProcessStartup(error) || !CDROMAsyncReader::ProcessStartup(error)) [[unlikely]]
   {
+    VideoThread::ProcessShutdown();
     s_locals.async_task_queue.SetWorkerCount(0, 0);
     s_locals.core_thread_handle = {};
 #ifdef _WIN32
@@ -783,6 +785,7 @@ void Core::CoreThreadShutdown()
   HTTPDownloader::Shutdown();
   HTTPCache::Shutdown();
 
+  CDROMAsyncReader::ProcessShutdown();
   VideoThread::ProcessShutdown();
 
   s_locals.core_thread_handle = {};