diff --git a/src/core/cdrom.cpp b/src/core/cdrom.cpp index 3586b7197..9535124e7 100644 --- a/src/core/cdrom.cpp +++ b/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); } diff --git a/src/core/cdrom_async_reader.cpp b/src/core/cdrom_async_reader.cpp index 3bc52f19f..8eef8d96d 100644 --- a/src/core/cdrom_async_reader.cpp +++ b/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& lock); static void ReadSectorNonThreaded(CDImage::LBA lba); @@ -26,17 +29,19 @@ struct Locals std::atomic 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 buffers; std::atomic buffer_front{0}; std::atomic buffer_back{0}; std::atomic buffer_count{0}; + std::vector 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(s_locals.buffers.size()); + return s_locals.readahead_enabled.load(std::memory_order_relaxed) ? static_cast(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 media) { - if (IsUsingThread()) - CancelReadahead(); - + CancelReadahead(); s_locals.media = std::move(media); } std::unique_ptr 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(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(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& 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(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& } 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(s_locals.buffers.size()) - s_locals.buffer_count.load()); - while (s_locals.buffer_count.load() < static_cast(s_locals.buffers.size())) + static_cast(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(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; } } diff --git a/src/core/cdrom_async_reader.h b/src/core/cdrom_async_reader.h index acc6194fd..0a12d40c6 100644 --- a/src/core/cdrom_async_reader.h +++ b/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 media); std::unique_ptr RemoveMedia(); diff --git a/src/core/core.cpp b/src/core/core.cpp index 96b7e8984..3416f3041 100644 --- a/src/core/core.cpp +++ b/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 = {};