From cd04ecb7f47d18438816d3f1f766860fc66da6a9 Mon Sep 17 00:00:00 2001 From: Stenzek Date: Sat, 12 Sep 2026 15:44:28 +1000 Subject: [PATCH] CDROMAsyncReader: Refactor to use batch reads --- src/core-tests/CMakeLists.txt | 1 + src/core-tests/cdrom_async_reader_tests.cpp | 443 +++++++++++++++++ src/core-tests/core-tests.vcxproj | 1 + src/core-tests/core-tests.vcxproj.filters | 1 + src/core/cdrom.cpp | 63 +-- src/core/cdrom_async_reader.cpp | 522 +++++++++++++------- src/core/cdrom_async_reader.h | 85 ++-- src/util-tests/cd_image_tests.cpp | 250 ++++++++++ 8 files changed, 1121 insertions(+), 245 deletions(-) create mode 100644 src/core-tests/cdrom_async_reader_tests.cpp diff --git a/src/core-tests/CMakeLists.txt b/src/core-tests/CMakeLists.txt index 3b02999cf..a32b59a8f 100644 --- a/src/core-tests/CMakeLists.txt +++ b/src/core-tests/CMakeLists.txt @@ -2,6 +2,7 @@ # SPDX-License-Identifier: CC-BY-NC-ND-4.0 + Packaging Restriction add_executable(core-tests + cdrom_async_reader_tests.cpp cheats_tests.cpp cpu_disasm_tests.cpp spu_tests.cpp diff --git a/src/core-tests/cdrom_async_reader_tests.cpp b/src/core-tests/cdrom_async_reader_tests.cpp new file mode 100644 index 000000000..2d2e4ee8b --- /dev/null +++ b/src/core-tests/cdrom_async_reader_tests.cpp @@ -0,0 +1,443 @@ +// SPDX-FileCopyrightText: 2026 Connor McLaughlin +// SPDX-License-Identifier: CC-BY-NC-ND-4.0 + +#include "core/cdrom_async_reader.h" + +#include "common/assert.h" + +#include "gtest/gtest.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace { +class TestCDImage final : public CDImage +{ +public: + struct ReadCall + { + LBA lba; + u32 count; + }; + + explicit TestCDImage(bool is_physical_device = true) : m_is_physical_device(is_physical_device) + { + Track track = {}; + track.track_number = 1; + track.start_lba = 0; + track.first_index = 0; + track.length = 1000; + track.mode = TrackMode::Audio; + track.submode = SubchannelMode::None; + track.control = SubChannelQ::Control(0); + m_tracks.push_back(track); + + Index index = {}; + index.file_sector_size = RAW_SECTOR_SIZE; + index.start_lba_on_disc = 0; + index.track_number = 1; + index.index_number = 1; + index.length = track.length; + index.mode = track.mode; + index.submode = track.submode; + index.control = track.control; + m_indices.push_back(index); + + m_lba_count = track.length; + AddLeadOutIndex(); + } + + void SetFirstFailedLBA(std::optional lba) { m_first_failed_lba = lba; } + + bool IsPhysicalDevice() const override { return m_is_physical_device; } + + void BlockNextRead() + { + std::unique_lock lock(m_mutex); + m_block_next_read = true; + m_read_blocked = false; + } + + void WaitUntilReadIsBlocked() + { + std::unique_lock lock(m_mutex); + m_cv.wait(lock, [this]() { return m_read_blocked; }); + } + + void UnblockRead() + { + std::unique_lock lock(m_mutex); + m_block_next_read = false; + m_cv.notify_all(); + } + + std::vector GetReadCalls() const + { + std::unique_lock lock(m_mutex); + return m_read_calls; + } + + u32 ReadSectorsFromIndex(std::span sectors, const Index& index, LBA lba_in_index, + SectorReadMode mode) override + { + EXPECT_EQ(mode, SectorReadMode::DataAndSubQ); + const LBA first_lba = index.start_lba_on_disc + lba_in_index; + { + std::unique_lock lock(m_mutex); + m_read_calls.push_back({first_lba, static_cast(sectors.size())}); + if (m_block_next_read) + { + m_read_blocked = true; + m_cv.notify_all(); + m_cv.wait(lock, [this]() { return !m_block_next_read; }); + } + } + + u32 count = static_cast(sectors.size()); + if (m_first_failed_lba.has_value()) + { + if (first_lba >= m_first_failed_lba.value()) + count = 0; + else if ((first_lba + count) > m_first_failed_lba.value()) + count = m_first_failed_lba.value() - first_lba; + } + + for (u32 i = 0; i < count; i++) + sectors[i].data.fill(static_cast(first_lba + i)); + return count; + } + +private: + mutable std::mutex m_mutex; + std::condition_variable m_cv; + std::vector m_read_calls; + std::optional m_first_failed_lba; + bool m_block_next_read = false; + bool m_read_blocked = false; + bool m_is_physical_device = true; +}; + +void ExpectAndRelease(CDROMAsyncReader& reader, CDImage::LBA expected_lba, bool expected_result = true) +{ + const CDROMAsyncReader::ReadResult& result = reader.WaitForReadToComplete(); + EXPECT_EQ(result.lba, expected_lba); + EXPECT_EQ(result.result, expected_result); + if (expected_result) + EXPECT_EQ(result.sector.data.front(), static_cast(expected_lba)); + reader.ReleaseSector(); +} +} // namespace + +TEST(CDROMAsyncReader, ReadsAndRefillsInBatches) +{ + auto image = std::make_unique(); + TestCDImage* image_ptr = image.get(); + + CDROMAsyncReader reader; + reader.SetMedia(std::move(image)); + reader.StartThread(8); + reader.QueueReadSector(100); + reader.WaitForIdle(); + + ASSERT_EQ(image_ptr->GetReadCalls().size(), 1u); + EXPECT_EQ(image_ptr->GetReadCalls()[0].lba, 100u); + EXPECT_EQ(image_ptr->GetReadCalls()[0].count, 8u); + + for (CDImage::LBA lba = 100; lba < 104; lba++) + { + ExpectAndRelease(reader, lba); + reader.QueueReadSector(lba + 1); + } + reader.WaitForIdle(); + + const std::vector calls = image_ptr->GetReadCalls(); + ASSERT_EQ(calls.size(), 2u); + EXPECT_EQ(calls[1].lba, 108u); + EXPECT_EQ(calls[1].count, 4u); +} + +TEST(CDROMAsyncReader, FileImagesReadAndRefillIncrementally) +{ + auto image = std::make_unique(false); + TestCDImage* image_ptr = image.get(); + + CDROMAsyncReader reader; + reader.SetMedia(std::move(image)); + reader.StartThread(8); + reader.QueueReadSector(100); + reader.WaitForIdle(); + + // File images publish each sector independently so a slow decompression cannot hold the entire forward window. + std::vector calls = image_ptr->GetReadCalls(); + ASSERT_EQ(calls.size(), 8u); + for (u32 i = 0; i < calls.size(); i++) + { + EXPECT_EQ(calls[i].lba, 100u + i); + EXPECT_EQ(calls[i].count, 1u); + } + + // Replenish immediately after consuming one sector instead of waiting for half the window to become empty. This + // preserves the old steady-state behavior when fast-forwarding while physical devices retain batched refills. + ExpectAndRelease(reader, 100); + reader.QueueReadSector(101); + reader.WaitForIdle(); + + calls = image_ptr->GetReadCalls(); + ASSERT_EQ(calls.size(), 9u); + EXPECT_EQ(calls.back().lba, 108u); + EXPECT_EQ(calls.back().count, 1u); +} + +TEST(CDROMAsyncReader, RetainsConsumedSectorsAfterRefill) +{ + auto image = std::make_unique(); + TestCDImage* image_ptr = image.get(); + + CDROMAsyncReader reader; + reader.SetMedia(std::move(image)); + reader.StartThread(8); + reader.QueueReadSector(5); + reader.WaitForIdle(); + + // Move four sectors forward, which triggers a refill of sectors 13 through 16. + for (CDImage::LBA lba = 5; lba < 9; lba++) + { + ExpectAndRelease(reader, lba); + reader.QueueReadSector(lba + 1); + } + reader.WaitForIdle(); + + ASSERT_EQ(image_ptr->GetReadCalls().size(), 2u); + EXPECT_EQ(image_ptr->GetReadCalls()[0].lba, 5u); + EXPECT_EQ(image_ptr->GetReadCalls()[0].count, 8u); + EXPECT_EQ(image_ptr->GetReadCalls()[1].lba, 13u); + EXPECT_EQ(image_ptr->GetReadCalls()[1].count, 4u); + + // Sector 5 is behind the current position but still resident in the history half of the cache. + reader.QueueReadSector(5); + ExpectAndRelease(reader, 5); + EXPECT_EQ(image_ptr->GetReadCalls().size(), 2u); +} + +TEST(CDROMAsyncReader, PreservesPartialBatchBeforeError) +{ + auto image = std::make_unique(); + TestCDImage* image_ptr = image.get(); + image_ptr->SetFirstFailedLBA(103); + + CDROMAsyncReader reader; + reader.SetMedia(std::move(image)); + reader.StartThread(8); + reader.QueueReadSector(100); + reader.WaitForIdle(); + + for (CDImage::LBA lba = 100; lba < 103; lba++) + { + ExpectAndRelease(reader, lba); + reader.QueueReadSector(lba + 1); + } + ExpectAndRelease(reader, 103, false); + reader.WaitForIdle(); + + // The queued error terminates this readahead run; consuming its successful prefix must not retry the same failure. + EXPECT_EQ(image_ptr->GetReadCalls().size(), 1u); +} + +TEST(CDROMAsyncReader, UncachedSubQOnlySkipsGeneratedDataRead) +{ + auto image = std::make_unique(); + TestCDImage* image_ptr = image.get(); + + CDROMAsyncReader reader; + reader.SetMedia(std::move(image)); + + CDImage::Sector sector; + sector.data.fill(0x5A); + ASSERT_TRUE(reader.ReadSectorUncached(100, §or, CDImage::SectorReadMode::SubQOnly)); + EXPECT_TRUE(image_ptr->GetReadCalls().empty()); + EXPECT_TRUE(sector.subq.IsCRCValid()); + EXPECT_TRUE(std::all_of(sector.data.begin(), sector.data.end(), [](u8 value) { return value == 0x5A; })); +} + +TEST(CDROMAsyncReader, CachedSectorRemainsAccessibleDuringBatchRead) +{ + auto image = std::make_unique(); + TestCDImage* image_ptr = image.get(); + + CDROMAsyncReader reader; + reader.SetMedia(std::move(image)); + reader.StartThread(4); + reader.QueueReadSector(100); + reader.WaitForIdle(); + + ExpectAndRelease(reader, 100); + reader.QueueReadSector(101); + ExpectAndRelease(reader, 101); + + // Advancing to sector 102 makes the four-sector ring half empty and starts a two-sector refill. Keep that backend + // call blocked while another thread exercises the same borrow path used by the emulation thread. + image_ptr->BlockNextRead(); + reader.QueueReadSector(102); + image_ptr->WaitUntilReadIsBlocked(); + + std::future> cached_result = + std::async(std::launch::async, [&reader]() -> std::pair { + const CDROMAsyncReader::ReadResult& result = reader.WaitForReadToComplete(); + const std::pair copy = {result.lba, result.result}; + reader.ReleaseSector(); + return copy; + }); + + const std::future_status cached_status = cached_result.wait_for(std::chrono::seconds(1)); + image_ptr->UnblockRead(); + ASSERT_EQ(cached_status, std::future_status::ready) << "Cached-sector access waited for the in-flight backend read"; + + const auto [lba, result] = cached_result.get(); + EXPECT_EQ(lba, 102u); + EXPECT_TRUE(result); +} + +TEST(CDROMAsyncReader, CachedSectorRemainsAccessibleDuringUncachedRead) +{ + auto image = std::make_unique(); + TestCDImage* image_ptr = image.get(); + + CDROMAsyncReader reader; + reader.SetMedia(std::move(image)); + reader.StartThread(4); + reader.QueueReadSector(100); + reader.WaitForIdle(); + + // An uncached read needs exclusive use of the stateful CDImage backend, but it must not retain the state mutex + // while waiting for the device. Cached-sector access should remain entirely independent of that slow operation. + image_ptr->BlockNextRead(); + CDImage::Sector uncached_sector; + std::future uncached_result = std::async( + std::launch::async, [&reader, &uncached_sector]() { return reader.ReadSectorUncached(200, &uncached_sector); }); + image_ptr->WaitUntilReadIsBlocked(); + + std::future> cached_result = + std::async(std::launch::async, [&reader]() -> std::pair { + const CDROMAsyncReader::ReadResult& result = reader.WaitForReadToComplete(); + const std::pair copy = {result.lba, result.result}; + reader.ReleaseSector(); + return copy; + }); + + const std::future_status cached_status = cached_result.wait_for(std::chrono::seconds(1)); + image_ptr->UnblockRead(); + ASSERT_EQ(cached_status, std::future_status::ready) << "Cached-sector access waited for the uncached backend read"; + + const auto [lba, result] = cached_result.get(); + EXPECT_EQ(lba, 100u); + EXPECT_TRUE(result); + EXPECT_TRUE(uncached_result.get()); + EXPECT_EQ(uncached_sector.data.front(), static_cast(200)); +} + +TEST(CDROMAsyncReader, BorrowedSectorRemainsStableDuringRefillPublication) +{ + auto image = std::make_unique(); + TestCDImage* image_ptr = image.get(); + + CDROMAsyncReader reader; + reader.SetMedia(std::move(image)); + reader.StartThread(4); + reader.QueueReadSector(100); + reader.WaitForIdle(); + + ExpectAndRelease(reader, 100); + reader.QueueReadSector(101); + ExpectAndRelease(reader, 101); + + image_ptr->BlockNextRead(); + reader.QueueReadSector(102); + image_ptr->WaitUntilReadIsBlocked(); + + // Keep the reference borrowed while the worker appends its completed batch. Publication may update atomic cache + // metadata, but must not relocate or overwrite the selected ring slot until the emulation thread releases it. + const CDROMAsyncReader::ReadResult& borrowed_result = reader.WaitForReadToComplete(); + ASSERT_EQ(borrowed_result.lba, 102u); + ASSERT_TRUE(borrowed_result.result); + ASSERT_EQ(borrowed_result.sector.data.front(), static_cast(102)); + + image_ptr->UnblockRead(); + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(1); + while (reader.GetBufferedSectorCount() < 4 && std::chrono::steady_clock::now() < deadline) + std::this_thread::yield(); + + const bool refill_published = (reader.GetBufferedSectorCount() == 4); + EXPECT_EQ(borrowed_result.lba, 102u); + EXPECT_TRUE(borrowed_result.result); + EXPECT_EQ(borrowed_result.sector.data.front(), static_cast(102)); + reader.ReleaseSector(); + + ASSERT_TRUE(refill_published) << "Worker did not publish the completed refill while a cached sector was borrowed"; + reader.WaitForIdle(); +} + +TEST(CDROMAsyncReader, BackwardHitDuringRefillDoesNotEvictCurrentSector) +{ + auto image = std::make_unique(); + TestCDImage* image_ptr = image.get(); + + CDROMAsyncReader reader; + reader.SetMedia(std::move(image)); + reader.StartThread(4); + reader.QueueReadSector(0); + reader.WaitForIdle(); + + // Fill both halves of the cache, leaving sectors 0 through 4 behind the current position and 5 through 7 ahead. + for (CDImage::LBA lba = 0; lba < 5; lba++) + { + ExpectAndRelease(reader, lba); + reader.QueueReadSector(lba + 1); + reader.WaitForIdle(); + } + + ExpectAndRelease(reader, 5); + image_ptr->BlockNextRead(); + reader.QueueReadSector(6); + image_ptr->WaitUntilReadIsBlocked(); + + // The new position needs slots which the in-flight batch intended to reclaim. The batch must be discarded rather + // than evicting sector 0 out from under the cache hit. + reader.QueueReadSector(0); + image_ptr->UnblockRead(); + reader.WaitForIdle(); + ExpectAndRelease(reader, 0); + + const std::vector calls = image_ptr->GetReadCalls(); + ASSERT_FALSE(calls.empty()); + EXPECT_EQ(calls.back().lba, 8u); +} + +TEST(CDROMAsyncReader, DiscardsStaleInFlightBatch) +{ + auto image = std::make_unique(); + TestCDImage* image_ptr = image.get(); + image_ptr->BlockNextRead(); + + CDROMAsyncReader reader; + reader.SetMedia(std::move(image)); + reader.StartThread(4); + reader.QueueReadSector(100); + image_ptr->WaitUntilReadIsBlocked(); + + // Queueing a new location changes the request generation while the old device/image request is still in flight. + reader.QueueReadSector(200); + image_ptr->UnblockRead(); + reader.WaitForIdle(); + + ExpectAndRelease(reader, 200); + const std::vector calls = image_ptr->GetReadCalls(); + ASSERT_EQ(calls.size(), 2u); + EXPECT_EQ(calls[0].lba, 100u); + EXPECT_EQ(calls[1].lba, 200u); +} diff --git a/src/core-tests/core-tests.vcxproj b/src/core-tests/core-tests.vcxproj index 4e555f4b0..bb457e081 100644 --- a/src/core-tests/core-tests.vcxproj +++ b/src/core-tests/core-tests.vcxproj @@ -3,6 +3,7 @@ + diff --git a/src/core-tests/core-tests.vcxproj.filters b/src/core-tests/core-tests.vcxproj.filters index fb8c94920..1e1ca9346 100644 --- a/src/core-tests/core-tests.vcxproj.filters +++ b/src/core-tests/core-tests.vcxproj.filters @@ -6,6 +6,7 @@ + diff --git a/src/core/cdrom.cpp b/src/core/cdrom.cpp index c982a48cd..4f0dfe5f7 100644 --- a/src/core/cdrom.cpp +++ b/src/core/cdrom.cpp @@ -1822,8 +1822,10 @@ CDImage::LBA CDROM::GetNextSectorToBeRead() if (!IsReadingOrPlaying() && !IsSeeking()) return s_state.current_lba; - s_reader.WaitForReadToComplete(); - return s_reader.GetLastReadSector(); + const CDROMAsyncReader::ReadResult& read_result = s_reader.WaitForReadToComplete(); + const CDImage::LBA read_lba = read_result.lba; + s_reader.ReleaseSector(); + return read_lba; } void CDROM::BeginCommand(Command command) @@ -3071,9 +3073,8 @@ void CDROM::UpdateSubQPosition(bool update_logical) if (update_logical) { - CDImage::SubChannelQ real_subq = {}; - CDROMAsyncReader::SectorBuffer raw_sector; - if (!s_reader.ReadSectorUncached(new_subq_lba, &real_subq, &raw_sector)) + CDImage::Sector sector = {}; + if (!s_reader.ReadSectorUncached(new_subq_lba, §or)) { ERROR_LOG("Failed to read subq for sector {} for subq position", new_subq_lba); } @@ -3081,11 +3082,11 @@ void CDROM::UpdateSubQPosition(bool update_logical) { s_state.last_subq_needs_update = false; - const CDImage::SubChannelQ& subq = GetSectorSubQ(new_subq_lba, real_subq); + const CDImage::SubChannelQ& subq = GetSectorSubQ(new_subq_lba, sector.subq); if (subq.IsCRCValid()) s_state.last_subq = subq; - ProcessDataSectorHeader(raw_sector.data()); + ProcessDataSectorHeader(sector.data.data()); } } } @@ -3108,11 +3109,11 @@ void CDROM::EnsureLastSubQValid() s_state.last_subq_needs_update = false; - CDImage::SubChannelQ real_subq = {}; - if (!s_reader.ReadSectorUncached(s_state.current_subq_lba, &real_subq, nullptr)) + CDImage::Sector sector = {}; + if (!s_reader.ReadSectorUncached(s_state.current_subq_lba, §or, CDImage::SectorReadMode::SubQOnly)) ERROR_LOG("Failed to read subq for sector {} for subq position", s_state.current_subq_lba); - const CDImage::SubChannelQ& subq = GetSectorSubQ(s_state.current_subq_lba, real_subq); + const CDImage::SubChannelQ& subq = GetSectorSubQ(s_state.current_subq_lba, sector.subq); if (subq.IsCRCValid()) s_state.last_subq = subq; } @@ -3131,24 +3132,25 @@ bool CDROM::CompleteSeek() const bool logical = (s_state.drive_state == DriveState::SeekingLogical); ClearDriveState(); - bool seek_okay = s_reader.WaitForReadToComplete(); + const CDROMAsyncReader::ReadResult& read_result = s_reader.WaitForReadToComplete(); + bool seek_okay = read_result.result; - s_state.current_subq_lba = s_reader.GetLastReadSector(); + s_state.current_subq_lba = read_result.lba; s_state.last_subq_needs_update = false; s_state.subq_lba_update_tick = System::GetGlobalTickCounter(); s_state.subq_lba_update_carry = 0; if (seek_okay) { - const CDImage::SubChannelQ& subq = GetSectorSubQ(s_reader.GetLastReadSector(), s_reader.GetSectorSubQ()); - s_state.current_lba = s_reader.GetLastReadSector(); + const CDImage::SubChannelQ& subq = GetSectorSubQ(read_result.lba, read_result.sector.subq); + s_state.current_lba = read_result.lba; if (subq.IsCRCValid()) { // seek and update sub-q for ReadP command s_state.last_subq = subq; s_state.last_subq_needs_update = false; - const auto [seek_mm, seek_ss, seek_ff] = CDImage::Position::FromLBA(s_reader.GetLastReadSector()).ToBCD(); + const auto [seek_mm, seek_ss, seek_ff] = CDImage::Position::FromLBA(read_result.lba).ToBCD(); seek_okay = (subq.absolute_minute_bcd == seek_mm && subq.absolute_second_bcd == seek_ss && subq.absolute_frame_bcd == seek_ff); if (seek_okay) @@ -3157,7 +3159,7 @@ bool CDROM::CompleteSeek() { if (logical) { - ProcessDataSectorHeader(s_reader.GetSectorBuffer().data()); + ProcessDataSectorHeader(read_result.sector.data.data()); seek_okay = (s_state.last_sector_header.minute == seek_mm && s_state.last_sector_header.second == seek_ss && s_state.last_sector_header.frame == seek_ff); @@ -3192,13 +3194,14 @@ bool CDROM::CompleteSeek() if (subq.track_number_bcd == CDImage::LEAD_OUT_TRACK_NUMBER) { - WARNING_LOG("Invalid seek to lead-out area (LBA {})", s_reader.GetLastReadSector()); + WARNING_LOG("Invalid seek to lead-out area (LBA {})", read_result.lba); seek_okay = false; } } } } + s_reader.ReleaseSector(); return seek_okay; } @@ -3208,8 +3211,7 @@ void CDROM::DoSeekComplete() const bool seek_okay = CompleteSeek(); if (seek_okay) { - DEV_LOG("{} seek to [{}] complete{}", logical ? "Logical" : "Physical", - LBAToMSFString(s_reader.GetLastReadSector()), + DEV_LOG("{} seek to [{}] complete{}", logical ? "Logical" : "Physical", LBAToMSFString(s_state.current_subq_lba), s_state.read_after_seek ? ", now reading" : (s_state.play_after_seek ? ", now playing" : "")); // seek complete, transition to play/read if requested @@ -3231,8 +3233,7 @@ void CDROM::DoSeekComplete() } else { - WARNING_LOG("{} seek to [{}] failed", logical ? "Logical" : "Physical", - LBAToMSFString(s_reader.GetLastReadSector())); + WARNING_LOG("{} seek to [{}] failed", logical ? "Logical" : "Physical", LBAToMSFString(s_state.current_subq_lba)); s_state.secondary_status.ClearActiveBits(); SendAsyncErrorResponse(STAT_SEEK_ERROR, 0x04); s_state.last_sector_header_valid = false; @@ -3381,8 +3382,10 @@ void CDROM::StopMotor() void CDROM::DoSectorRead() { // TODO: Queue the next read here and swap the buffer. - if (!s_reader.WaitForReadToComplete()) [[unlikely]] + const CDROMAsyncReader::ReadResult& read_result = s_reader.WaitForReadToComplete(); + if (!read_result.result) [[unlikely]] { + s_reader.ReleaseSector(); Host::AddIconOSDMessage( OSDMessageType::Error, "DiscReadError", ICON_EMOJI_WARNING, TRANSLATE_STR("CDROM", "Disc Read Error"), TRANSLATE_STR( @@ -3391,7 +3394,7 @@ void CDROM::DoSectorRead() return; } - s_state.current_lba = s_reader.GetLastReadSector(); + s_state.current_lba = read_result.lba; s_state.current_subq_lba = s_state.current_lba; s_state.last_subq_needs_update = false; s_state.subq_lba_update_tick = System::GetGlobalTickCounter(); @@ -3399,7 +3402,7 @@ void CDROM::DoSectorRead() s_state.secondary_status.SetReadingBits(s_state.drive_state == DriveState::Playing); - const CDImage::SubChannelQ& subq = GetSectorSubQ(s_state.current_lba, s_reader.GetSectorSubQ()); + const CDImage::SubChannelQ& subq = GetSectorSubQ(s_state.current_lba, read_result.sector.subq); const bool subq_valid = subq.IsCRCValid(); if (subq_valid) { @@ -3440,7 +3443,8 @@ void CDROM::DoSectorRead() if (subq.track_number_bcd == CDImage::LEAD_OUT_TRACK_NUMBER) { - DEV_LOG("Read reached lead-out area of disc at LBA {}, stopping", s_reader.GetLastReadSector()); + DEV_LOG("Read reached lead-out area of disc at LBA {}, stopping", read_result.lba); + s_reader.ReleaseSector(); StopReadingWithDataEnd(); StopMotor(); return; @@ -3449,7 +3453,7 @@ void CDROM::DoSectorRead() const bool is_data_sector = subq.IsData(); if (is_data_sector) { - ProcessDataSectorHeader(s_reader.GetSectorBuffer().data()); + ProcessDataSectorHeader(read_result.sector.data.data()); } else if (s_state.mode.auto_pause) { @@ -3458,6 +3462,7 @@ void CDROM::DoSectorRead() DEV_COLOR_LOG(StrongRed, "Auto pause at the start of track {:02x} ({} LBA {})", subq.track_number_bcd, LBAToMSFString(s_state.current_lba), s_state.current_lba); s_state.cdda_auto_pause_pending = false; + s_reader.ReleaseSector(); StopReadingWithDataEnd(); return; } @@ -3482,18 +3487,19 @@ void CDROM::DoSectorRead() u32 next_sector = s_state.current_lba + 1u; if (is_data_sector && s_state.drive_state == DriveState::Reading) { - ProcessDataSector(s_reader.GetSectorBuffer().data(), subq); + ProcessDataSector(read_result.sector.data.data(), subq); } else if (!is_data_sector && (s_state.drive_state == DriveState::Playing || (s_state.drive_state == DriveState::Reading && s_state.mode.cdda))) { - ProcessCDDASector(s_reader.GetSectorBuffer().data(), subq, subq_valid); + ProcessCDDASector(read_result.sector.data.data(), subq, subq_valid); if (s_state.fast_forward_rate != 0) next_sector = s_state.current_lba + SignExtend32(s_state.fast_forward_rate); } else if (s_state.drive_state != DriveState::Reading && s_state.drive_state != DriveState::Playing) { + s_reader.ReleaseSector(); Panic("Not reading or playing"); } else @@ -3503,6 +3509,7 @@ void CDROM::DoSectorRead() } s_state.requested_lba = next_sector; + s_reader.ReleaseSector(); s_reader.QueueReadSector(s_state.requested_lba); } diff --git a/src/core/cdrom_async_reader.cpp b/src/core/cdrom_async_reader.cpp index fa8c3cf6d..dfb3a4eb0 100644 --- a/src/core/cdrom_async_reader.cpp +++ b/src/core/cdrom_async_reader.cpp @@ -1,10 +1,13 @@ -// SPDX-FileCopyrightText: 2019-2024 Connor McLaughlin +// SPDX-FileCopyrightText: 2019-2026 Connor McLaughlin // SPDX-License-Identifier: CC-BY-NC-ND-4.0 #include "cdrom_async_reader.h" #include "common/assert.h" #include "common/log.h" #include "common/timer.h" + +#include + LOG_CHANNEL(CDROMAsyncReader); CDROMAsyncReader::CDROMAsyncReader() = default; @@ -14,18 +17,49 @@ CDROMAsyncReader::~CDROMAsyncReader() StopThread(); } +u32 CDROMAsyncReader::GetBufferedSectorCount() const +{ + return m_buffered_sector_count.load(std::memory_order_acquire); +} + +bool CDROMAsyncReader::HasBufferedSectors() const +{ + return (m_buffered_sector_count.load(std::memory_order_acquire) > 0); +} + +u32 CDROMAsyncReader::GetReadaheadCount() const +{ + std::unique_lock lock(m_mutex); + return m_readahead_count; +} + void CDROMAsyncReader::StartThread(u32 readahead_count) { + Assert(readahead_count > 0); if (IsUsingThread()) StopThread(); - m_buffers.clear(); - m_buffers.resize(readahead_count); - EmptyBuffers(); + { + std::unique_lock lock(m_mutex); + Assert(!m_sector_borrowed.load(std::memory_order_acquire)); + Assert(readahead_count <= (std::numeric_limits::max() / 2)); + m_readahead_count = readahead_count; + m_buffers.clear(); + // Retain up to one readahead window behind the current sector in addition to the forward window. + m_buffers.resize(static_cast(readahead_count) * 2); + // Allocate the worker's staging storage before the thread starts. Batch reads never allocate while holding the + // state mutex, keeping contended sections bounded to ring bookkeeping and buffer copies. + m_read_buffer.clear(); + m_read_buffer.resize(readahead_count); + EmptyBuffersLocked(); + m_next_position.reset(); + m_can_readahead = false; + m_shutdown_flag = false; + } - m_shutdown_flag.store(false); m_read_thread = std::thread(&CDROMAsyncReader::WorkerThreadEntryPoint, this); - INFO_LOG("Read thread started with readahead of {} sectors", readahead_count); + INFO_LOG("Read thread started with {} sectors of readahead and {} sectors of history", readahead_count, + readahead_count); } void CDROMAsyncReader::StopThread() @@ -35,36 +69,46 @@ void CDROMAsyncReader::StopThread() { std::unique_lock lock(m_mutex); - m_shutdown_flag.store(true); + Assert(!m_sector_borrowed.load(std::memory_order_acquire)); + m_request_generation++; + m_next_position.reset(); + m_can_readahead = false; + m_shutdown_flag = true; m_do_read_cv.notify_one(); + m_notify_read_complete_cv.notify_all(); } m_read_thread.join(); - EmptyBuffers(); + + std::unique_lock lock(m_mutex); + EmptyBuffersLocked(); m_buffers.clear(); + m_read_buffer.clear(); + m_readahead_count = 0; } void CDROMAsyncReader::SetMedia(std::unique_ptr media) { - if (IsUsingThread()) - CancelReadahead(); - + std::unique_lock lock(m_mutex); + Assert(!m_sector_borrowed.load(std::memory_order_acquire)); + CancelReadaheadLocked(lock); m_media = std::move(media); } std::unique_ptr CDROMAsyncReader::RemoveMedia() { - if (IsUsingThread()) - CancelReadahead(); - + std::unique_lock lock(m_mutex); + Assert(!m_sector_borrowed.load(std::memory_order_acquire)); + CancelReadaheadLocked(lock); return std::move(m_media); } bool CDROMAsyncReader::Precache(ProgressCallback* callback, Error* error) { - WaitForIdle(); - std::unique_lock lock(m_mutex); + Assert(!m_sector_borrowed.load(std::memory_order_acquire)); + CancelReadaheadLocked(lock); + if (!m_media) return false; else if (m_media->IsPrecached()) @@ -73,18 +117,10 @@ bool CDROMAsyncReader::Precache(ProgressCallback* callback, Error* error) const CDImage::PrecacheResult res = m_media->Precache(callback, error); if (res == CDImage::PrecacheResult::Unsupported) { - // fall back to copy precaching + // Fall back to copy precaching. std::unique_ptr memory_image = CDImage::CreateMemoryImage(m_media.get(), callback, error); if (memory_image) { - const CDImage::LBA lba = m_media->GetPositionOnDisc(); - if (!memory_image->Seek(lba)) [[unlikely]] - { - ERROR_LOG("Failed to seek to LBA {} in memory image", lba); - return false; - } - - m_media.reset(); m_media = std::move(memory_image); return true; } @@ -105,71 +141,92 @@ void CDROMAsyncReader::QueueReadSector(CDImage::LBA lba) return; } - const u32 buffer_count = m_buffer_count.load(); - if (buffer_count > 0) + std::unique_lock lock(m_mutex); + Assert(m_media && !m_buffers.empty()); + Assert(!m_sector_borrowed.load(std::memory_order_acquire)); + + if (m_buffer_count > 0) { - // 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 = m_buffer_front.load(); - if (m_buffers[buffer_front].lba == lba) + const u32 current_buffer = GetCurrentBufferLocked(); + + // Don't re-read the same sector if it was the last one we read. + // The CDC code does this when seeking->reading. + if (m_buffers[current_buffer].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(m_buffers.size()); - if (m_buffer_count > 1 && m_buffers[next_buffer].lba == lba) + // Search the whole cache, including sectors behind the current position. Forward hits avoid seeks when the + // emulated drive skips sectors, while history hits avoid physical reads when pause/position simulation moves + // the drive back to a recently consumed sector. + for (u32 offset = 0; offset < m_buffer_count; offset++) { - // great, don't need a seek, but still kick the thread to start reading ahead again - DEBUG_LOG("Readahead buffer hit for sector {}", lba); - m_buffer_front.store(next_buffer); - m_buffer_count.fetch_sub(1); - m_can_readahead.store(true); - m_do_read_cv.notify_one(); - return; + const u32 buffer_index = (m_buffer_front + offset) % static_cast(m_buffers.size()); + if (m_buffers[buffer_index].lba == lba) + { + [[maybe_unused]] const bool history_hit = (offset < m_buffer_current_offset); + m_buffer_current_offset = offset; + const u32 forward_count = GetBufferedSectorCountLocked(); + UpdatePublishedCacheStateLocked(); + DEBUG_LOG("Sector cache {} hit for LBA {} ({} cached behind, {} buffered forward)", + history_hit ? "history" : "readahead", lba, m_buffer_current_offset, forward_count); + + // Physical devices refill half a window at a time to avoid a command (and potentially a full rotation) per + // sector. Image files refill immediately and publish each sector as it completes, which keeps decompression + // latency off the emulation thread and gives the worker a chance to observe a new request between sectors. + const bool batch_readahead = m_media->IsPhysicalDevice(); + const u32 refill_threshold = m_readahead_count / 2; + const u32 last_buffer = (m_buffer_front + m_buffer_count - 1) % static_cast(m_buffers.size()); + const bool error_queued = !m_buffers[last_buffer].result; + const bool refill_needed = + batch_readahead ? (forward_count <= refill_threshold) : (forward_count < m_readahead_count); + m_can_readahead = (m_buffers[buffer_index].result && !error_queued && refill_needed); + if (m_can_readahead) + m_do_read_cv.notify_one(); + else if (error_queued) + TRACE_LOG("Not refilling after sector {} because an error is already queued at LBA {}", lba, + m_buffers[last_buffer].lba); + return; + } } } - // we need to toss away our readahead and start fresh - DEBUG_LOG("Readahead buffer miss, queueing seek to {}", lba); - std::unique_lock lock(m_mutex); - m_next_position_set.store(true); + // We need to toss away our readahead and start fresh. + DEBUG_LOG("Sector cache miss, queueing read at {} (discarding {} cached sectors)", lba, m_buffer_count); + m_request_generation++; m_next_position = lba; + m_can_readahead = false; + // Invalidate the lock-free fast path before returning. The worker will initialize the ring for this request, but a + // caller must not borrow a sector left over from the previous position in the meantime. + EmptyBuffersLocked(); m_do_read_cv.notify_one(); + m_notify_read_complete_cv.notify_all(); } -bool CDROMAsyncReader::ReadSectorUncached(CDImage::LBA lba, CDImage::SubChannelQ* subq, SectorBuffer* data) +bool CDROMAsyncReader::ReadSectorUncached(CDImage::LBA lba, CDImage::Sector* sector, CDImage::SectorReadMode mode) { - if (!IsUsingThread()) - return InternalReadSectorUncached(lba, subq, data); - std::unique_lock lock(m_mutex); + Assert(!m_sector_borrowed.load(std::memory_order_acquire)); - // wait until the read thread is idle - m_notify_read_complete_cv.wait(lock, [this]() { return !m_is_reading.load(); }); + // Reserve exclusive access to the stateful CDImage backend. Keep the reservation in m_is_reading, but release the + // state mutex so cache publication and cached reads are not blocked by the I/O itself. + m_notify_read_complete_cv.wait(lock, [this]() { return !m_is_reading; }); + m_is_reading = true; + lock.unlock(); - // read while the lock is held so it has to wait - const CDImage::LBA prev_lba = m_media->GetPositionOnDisc(); - const bool result = InternalReadSectorUncached(lba, subq, data); - if (!m_media->Seek(prev_lba)) [[unlikely]] - { - ERROR_LOG("Failed to re-seek to cached position {}", prev_lba); - m_can_readahead.store(false); - } + const bool result = InternalReadSectorUncached(lba, sector, mode); + lock.lock(); + m_is_reading = false; + m_notify_read_complete_cv.notify_all(); return result; } -bool CDROMAsyncReader::InternalReadSectorUncached(CDImage::LBA lba, CDImage::SubChannelQ* subq, SectorBuffer* data) +bool CDROMAsyncReader::InternalReadSectorUncached(CDImage::LBA lba, CDImage::Sector* sector, + CDImage::SectorReadMode mode) { - if (m_media->GetPositionOnDisc() != lba && !m_media->Seek(lba)) [[unlikely]] - { - WARNING_LOG("Seek to LBA {} failed", lba); - return false; - } - - if (!m_media->ReadRawSector(data, subq)) [[unlikely]] + if (!m_media || m_media->ReadSectors(lba, std::span(sector, 1), mode) != 1) [[unlikely]] { WARNING_LOG("Read of LBA {} failed", lba); return false; @@ -178,34 +235,51 @@ bool CDROMAsyncReader::InternalReadSectorUncached(CDImage::LBA lba, CDImage::Sub return true; } -bool CDROMAsyncReader::WaitForReadToComplete() +const CDROMAsyncReader::ReadResult& CDROMAsyncReader::WaitForReadToComplete() { - // Safe without locking with memory_order_seq_cst. - if (!m_next_position_set.load() && m_buffer_count.load() > 0) - { - TRACE_LOG("Returning sector {}", m_buffers[m_buffer_front.load()].lba); - return m_buffers[m_buffer_front.load()].result; - } - Timer wait_timer; - DEBUG_LOG("Sector read pending, waiting"); - std::unique_lock lock(m_mutex); - m_notify_read_complete_cv.wait( - lock, [this]() { return (m_buffer_count.load() > 0 || m_seek_error.load()) && !m_next_position_set.load(); }); - if (m_seek_error.load()) [[unlikely]] + // Publication stores the slot index before the forward count with release semantics. Once the count is non-zero, + // the current result is fully initialized and the worker will not overwrite that selected slot. QueueReadSector() + // is not legal while a result is borrowed, so pinning it here keeps the returned reference valid without locking. + if (m_buffered_sector_count.load(std::memory_order_acquire) > 0) { - m_seek_error.store(false); - return false; + const bool was_borrowed = m_sector_borrowed.exchange(true, std::memory_order_acq_rel); + Assert(!was_borrowed); + m_borrowed_buffer = m_published_buffer.load(std::memory_order_relaxed); + + const ReadResult& result = m_buffers[m_borrowed_buffer]; + TRACE_LOG("Borrowing cached sector {} ({} cached behind, {} buffered forward)", result.lba, + m_cached_behind_count.load(std::memory_order_relaxed), + m_buffered_sector_count.load(std::memory_order_relaxed)); + return result; } - const u32 front = m_buffer_front.load(); + std::unique_lock lock(m_mutex); + Assert(!m_sector_borrowed.load(std::memory_order_acquire)); + + if (m_buffer_count == 0 || m_next_position.has_value()) + DEBUG_LOG("Sector read pending, waiting"); + + m_notify_read_complete_cv.wait(lock, [this]() { return (m_buffer_count > 0 && !m_next_position.has_value()); }); + + m_borrowed_buffer = GetCurrentBufferLocked(); + m_sector_borrowed.store(true, std::memory_order_release); + const ReadResult& result = m_buffers[m_borrowed_buffer]; const double wait_time = wait_timer.GetTimeMilliseconds(); if (wait_time > 1.0f) [[unlikely]] - WARNING_LOG("Had to wait {:.2f} msec for LBA {}", wait_time, m_buffers[front].lba); + WARNING_LOG("Had to wait {:.2f} msec for LBA {}", wait_time, result.lba); - TRACE_LOG("Returning sector {} after waiting", m_buffers[front].lba); - return m_buffers[front].result; + TRACE_LOG("Borrowing sector {} ({} cached behind, {} buffered forward)", result.lba, m_buffer_current_offset, + GetBufferedSectorCountLocked()); + return result; +} + +void CDROMAsyncReader::ReleaseSector() +{ + Assert(m_sector_borrowed.load(std::memory_order_acquire)); + TRACE_LOG("Releasing sector {}", m_buffers[m_borrowed_buffer].lba); + m_sector_borrowed.store(false, std::memory_order_release); } void CDROMAsyncReader::WaitForIdle() @@ -214,96 +288,191 @@ void CDROMAsyncReader::WaitForIdle() return; std::unique_lock lock(m_mutex); - m_notify_read_complete_cv.wait(lock, [this]() { return (!m_is_reading.load() && !m_next_position_set.load()); }); + Assert(!m_sector_borrowed.load(std::memory_order_acquire)); + m_notify_read_complete_cv.wait( + lock, [this]() { return (!m_is_reading && !m_next_position.has_value() && !m_can_readahead); }); +} + +u32 CDROMAsyncReader::GetCurrentBufferLocked() const +{ + Assert(m_buffer_count > 0 && m_buffer_current_offset < m_buffer_count); + return (m_buffer_front + m_buffer_current_offset) % static_cast(m_buffers.size()); +} + +u32 CDROMAsyncReader::GetBufferedSectorCountLocked() const +{ + Assert(m_buffer_current_offset <= m_buffer_count); + return m_buffer_count - m_buffer_current_offset; +} + +void CDROMAsyncReader::UpdatePublishedCacheStateLocked() +{ + if (m_buffer_count > 0) + { + m_published_buffer.store(GetCurrentBufferLocked(), std::memory_order_relaxed); + m_cached_behind_count.store(m_buffer_current_offset, std::memory_order_relaxed); + // This release publishes both the selected index and all writes to the selected ReadResult. + m_buffered_sector_count.store(GetBufferedSectorCountLocked(), std::memory_order_release); + } + else + { + // Clear availability first. The other published values are irrelevant until a later non-zero release store. + m_buffered_sector_count.store(0, std::memory_order_release); + m_published_buffer.store(0, std::memory_order_relaxed); + m_cached_behind_count.store(0, std::memory_order_relaxed); + } } -void CDROMAsyncReader::EmptyBuffers() +void CDROMAsyncReader::EmptyBuffersLocked() { - m_buffer_front.store(0); - m_buffer_back.store(0); - m_buffer_count.store(0); + Assert(!m_sector_borrowed.load(std::memory_order_acquire)); + m_buffer_front = 0; + m_buffer_back = 0; + m_buffer_count = 0; + m_buffer_current_offset = 0; + UpdatePublishedCacheStateLocked(); } -bool CDROMAsyncReader::ReadSectorIntoBuffer(std::unique_lock& lock) +bool CDROMAsyncReader::ReadSectorBatch(std::unique_lock& lock) { Timer timer; - const u32 slot = m_buffer_back.load(); - m_buffer_back.store((slot + 1) % static_cast(m_buffers.size())); + // An uncached read or precache operation may have reserved the backend while leaving the state mutex unlocked. + m_notify_read_complete_cv.wait( + lock, [this]() { return (!m_is_reading || m_shutdown_flag || m_next_position.has_value()); }); + if (m_shutdown_flag || m_next_position.has_value()) + return false; + + const u32 forward_count = GetBufferedSectorCountLocked(); + Assert(forward_count < m_readahead_count); + const u32 forward_slots = m_readahead_count - forward_count; + const u32 sectors_to_read = m_media->IsPhysicalDevice() ? forward_slots : 1; + Assert(sectors_to_read > 0); - BufferSlot& buffer = m_buffers[slot]; - buffer.lba = m_media->GetPositionOnDisc(); - m_is_reading.store(true); + const u64 generation = m_request_generation; + const CDImage::LBA first_lba = m_next_read_lba; + m_is_reading = true; + + // Never hold the state mutex while accessing the image. Physical-device reads can take a full rotation, while + // the emulation thread must remain able to borrow and release sectors which are already in the ring. lock.unlock(); - TRACE_LOG("Reading LBA {}...", buffer.lba); + TRACE_LOG("Reading {} sectors starting at LBA {}...", sectors_to_read, first_lba); + const u32 sectors_read = m_media->ReadSectors( + first_lba, std::span(m_read_buffer).first(sectors_to_read), CDImage::SectorReadMode::DataAndSubQ); + + lock.lock(); + m_is_reading = false; + m_notify_read_complete_cv.notify_all(); - buffer.result = m_media->ReadRawSector(buffer.data.data(), &buffer.subq); - if (buffer.result) [[likely]] + // A newer request arrived while the lock was released. Its handler will reset the ring. + if (generation != m_request_generation || m_shutdown_flag || m_next_position.has_value()) { - const double read_time = timer.GetTimeMilliseconds(); - if (read_time > 1.0f) [[unlikely]] - DEV_LOG("Read LBA {} took {:.2f} msec", buffer.lba, read_time); + DEBUG_LOG("Discarding stale batch at LBA {} (request generation {} is now {})", first_lba, generation, + m_request_generation); + return false; } - else + + Assert(sectors_read <= sectors_to_read); + + // Make room only after the read completes, so history remains available to the emulation thread during slow + // physical I/O. If it moved farther back while the mutex was released, the batch may no longer fit without + // evicting its current sector; discard that now-unneeded readahead instead. + const u32 results_to_queue = sectors_read + ((sectors_read < sectors_to_read) ? 1u : 0u); + const u32 free_slots = static_cast(m_buffers.size()) - m_buffer_count; + const u32 history_to_evict = (results_to_queue > free_slots) ? (results_to_queue - free_slots) : 0; + if (history_to_evict > m_buffer_current_offset) + { + DEBUG_LOG("Discarding completed batch at LBA {} because the current cache position moved backward", first_lba); + return false; + } + else if (history_to_evict > 0) { - ERROR_LOG("Read of LBA {} failed", buffer.lba); + DEBUG_LOG("Evicting {} oldest cached sectors to append batch at LBA {}", history_to_evict, first_lba); + m_buffer_front = (m_buffer_front + history_to_evict) % static_cast(m_buffers.size()); + m_buffer_count -= history_to_evict; + m_buffer_current_offset -= history_to_evict; } - lock.lock(); - m_is_reading.store(false); - m_buffer_count.fetch_add(1); + for (u32 i = 0; i < sectors_read; i++) + { + ReadResult& result = m_buffers[m_buffer_back]; + result.lba = first_lba + i; + result.sector = std::move(m_read_buffer[i]); + result.result = true; + m_buffer_back = (m_buffer_back + 1) % static_cast(m_buffers.size()); + m_buffer_count++; + } + + m_next_read_lba = first_lba + sectors_read; + if (sectors_read < sectors_to_read) [[unlikely]] + { + // Preserve the successful prefix, then queue a failure at the exact LBA which could not be read. Consumers can + // process every completed sector before receiving the error, matching CDImage's partial-read contract. + ReadResult& error_result = m_buffers[m_buffer_back]; + error_result.lba = m_next_read_lba; + error_result.sector = {}; + error_result.result = false; + m_buffer_back = (m_buffer_back + 1) % static_cast(m_buffers.size()); + m_buffer_count++; + ERROR_LOG("Batch read at LBA {} returned {} of {} sectors; first failed LBA is {}", first_lba, sectors_read, + sectors_to_read, error_result.lba); + } + + UpdatePublishedCacheStateLocked(); m_notify_read_complete_cv.notify_all(); - return true; + + const double read_time = timer.GetTimeMilliseconds(); + if (read_time > 1.0f) [[unlikely]] + DEV_LOG("Read {} of {} sectors at LBA {} in {:.2f} msec", sectors_read, sectors_to_read, first_lba, read_time); + + return (sectors_read == sectors_to_read); } void CDROMAsyncReader::ReadSectorNonThreaded(CDImage::LBA lba) { Timer timer; + Assert(!m_sector_borrowed.load(std::memory_order_acquire)); - m_buffers.resize(1); - m_seek_error.store(false); - EmptyBuffers(); - - if (m_media->GetPositionOnDisc() != lba && !m_media->Seek(lba)) - { - WARNING_LOG("Seek to LBA {} failed", lba); - m_seek_error.store(true); - return; - } - - BufferSlot& buffer = m_buffers.front(); - buffer.lba = m_media->GetPositionOnDisc(); - - TRACE_LOG("Reading LBA {}...", buffer.lba); + // No worker exists in this mode, so read into a local result without holding the state mutex and publish it only + // after the backend operation has completed. + ReadResult new_result; + new_result.lba = lba; - buffer.result = m_media->ReadRawSector(buffer.data.data(), &buffer.subq); - if (buffer.result) [[likely]] + TRACE_LOG("Reading LBA {}...", new_result.lba); + new_result.result = InternalReadSectorUncached(lba, &new_result.sector, CDImage::SectorReadMode::DataAndSubQ); + if (new_result.result) [[likely]] { const double read_time = timer.GetTimeMilliseconds(); if (read_time > 1.0f) [[unlikely]] - DEV_LOG("Read LBA {} took {:.2f} msec", buffer.lba, read_time); + DEV_LOG("Read LBA {} took {:.2f} msec", new_result.lba, read_time); } else { - ERROR_LOG("Read of LBA {} failed", buffer.lba); + ERROR_LOG("Read of LBA {} failed", new_result.lba); } - m_buffer_count.fetch_add(1); + std::unique_lock lock(m_mutex); + Assert(!m_sector_borrowed.load(std::memory_order_acquire)); + m_buffers.resize(1); + EmptyBuffersLocked(); + m_buffers.front() = std::move(new_result); + m_buffer_count = 1; + m_buffer_back = 0; + UpdatePublishedCacheStateLocked(); } -void CDROMAsyncReader::CancelReadahead() +void CDROMAsyncReader::CancelReadaheadLocked(std::unique_lock& lock) { - DEV_LOG("Cancelling readahead"); - - std::unique_lock lock(m_mutex); + DEV_LOG("Cancelling readahead ({} sectors cached, generation {})", m_buffer_count, m_request_generation); - // wait until the read thread is idle - m_notify_read_complete_cv.wait(lock, [this]() { return !m_is_reading.load(); }); + m_request_generation++; + m_next_position.reset(); + m_can_readahead = false; - // prevent it from doing any more when it re-acquires the lock - m_can_readahead.store(false); - EmptyBuffers(); + // Wait until an in-flight read observes the generation change and becomes idle. + m_notify_read_complete_cv.wait(lock, [this]() { return !m_is_reading; }); + EmptyBuffersLocked(); } void CDROMAsyncReader::WorkerThreadEntryPoint() @@ -312,69 +481,56 @@ void CDROMAsyncReader::WorkerThreadEntryPoint() for (;;) { - m_do_read_cv.wait( - lock, [this]() { return (m_shutdown_flag.load() || m_next_position_set.load() || m_can_readahead.load()); }); - if (m_shutdown_flag.load()) + m_do_read_cv.wait(lock, [this]() { return (m_shutdown_flag || m_next_position.has_value() || m_can_readahead); }); + if (m_shutdown_flag) break; for (;;) { - if (m_next_position_set.load()) + if (m_next_position.has_value()) { - // discard buffers, we're seeking to a new location - const CDImage::LBA seek_location = m_next_position.load(); - EmptyBuffers(); - m_next_position_set.store(false); - m_seek_error.store(false); - m_is_reading.store(true); - lock.unlock(); - - // seek without lock held in case it takes time - DEBUG_LOG("Seeking to LBA {}...", seek_location); - const bool seek_result = (m_media->GetPositionOnDisc() == seek_location || m_media->Seek(seek_location)); - - lock.lock(); - m_is_reading.store(false); - - // did another request come in? abort if so - if (m_next_position_set.load()) - continue; - - // did we fail the seek? - if (!seek_result) [[unlikely]] - { - // add the error result, and don't try to read ahead - WARNING_LOG("Seek to LBA {} failed", seek_location); - m_seek_error.store(true); - m_notify_read_complete_cv.notify_all(); - break; - } - - // go go read ahead! - m_can_readahead.store(true); + // Discard buffers and start reading at the new location. CDImage reads use explicit LBAs, so this does not + // need a separate blocking seek operation. + const CDImage::LBA read_location = m_next_position.value(); + EmptyBuffersLocked(); + m_next_position.reset(); + m_next_read_lba = read_location; + m_can_readahead = true; } - if (!m_can_readahead.load()) + if (!m_can_readahead) break; - // readahead time! read as many sectors as we have space for - DEBUG_LOG("Reading ahead {} sectors...", static_cast(m_buffers.size()) - m_buffer_count.load()); - while (m_buffer_count.load() < static_cast(m_buffers.size())) + // Physical devices fill the available forward space in one request, while image files publish each sector as + // it completes. The other half of the ring retains recently consumed sectors and is evicted only when a + // completed read needs the space. + DEBUG_LOG("Reading ahead {} sectors from LBA {} ({} cached behind, {} buffered forward)", + m_readahead_count - GetBufferedSectorCountLocked(), m_next_read_lba, m_buffer_current_offset, + GetBufferedSectorCountLocked()); + while (m_can_readahead && GetBufferedSectorCountLocked() < m_readahead_count) { - if (m_next_position_set.load()) + if (m_next_position.has_value()) { - // a seek request came in while we're reading, so bail out + // A new request came in while we were reading, so bail out and service it. break; } - // stop reading if we hit the end or get an error - if (!ReadSectorIntoBuffer(lock)) + // Stop reading if we hit the end or get an error. A partial batch remains queued ahead of its error result. + if (!ReadSectorBatch(lock)) break; } - // readahead buffer is full or errored at this point - m_can_readahead.store(false); + if (m_next_position.has_value()) + continue; + + // The forward window is full, the request moved backward, or a read errored at this point. + m_can_readahead = false; + m_notify_read_complete_cv.notify_all(); break; } } + + m_is_reading = false; + m_can_readahead = false; + m_notify_read_complete_cv.notify_all(); } diff --git a/src/core/cdrom_async_reader.h b/src/core/cdrom_async_reader.h index b3d297cf5..52dfa3fec 100644 --- a/src/core/cdrom_async_reader.h +++ b/src/core/cdrom_async_reader.h @@ -2,11 +2,12 @@ // SPDX-License-Identifier: CC-BY-NC-ND-4.0 #pragma once -#include "util/cd_image.h" #include "types.h" -#include +#include "util/cd_image.h" #include #include +#include +#include #include class ProgressCallback; @@ -14,25 +15,19 @@ class ProgressCallback; class CDROMAsyncReader { public: - using SectorBuffer = std::array; - - struct BufferSlot + struct ReadResult { - CDImage::LBA lba; - SectorBuffer data; - CDImage::SubChannelQ subq; - bool result; + CDImage::LBA lba = 0; + CDImage::Sector sector; + bool result = false; }; CDROMAsyncReader(); ~CDROMAsyncReader(); - CDImage::LBA GetLastReadSector() const { return m_buffers[m_buffer_front.load()].lba; } - const SectorBuffer& GetSectorBuffer() const { return m_buffers[m_buffer_front.load()].data; } - const CDImage::SubChannelQ& GetSectorSubQ() const { return m_buffers[m_buffer_front.load()].subq; } - u32 GetBufferedSectorCount() const { return m_buffer_count.load(); } - bool HasBufferedSectors() const { return (m_buffer_count.load() > 0); } - u32 GetReadaheadCount() const { return static_cast(m_buffers.size()); } + u32 GetBufferedSectorCount() const; + bool HasBufferedSectors() const; + u32 GetReadaheadCount() const; bool HasMedia() const { return static_cast(m_media); } const CDImage* GetMedia() const { return m_media.get(); } @@ -51,38 +46,60 @@ public: void QueueReadSector(CDImage::LBA lba); - bool WaitForReadToComplete(); + /// Returns a borrowed result which remains valid until ReleaseSector() is called. + const ReadResult& WaitForReadToComplete(); + void ReleaseSector(); void WaitForIdle(); /// Bypasses the sector cache and reads directly from the image. - bool ReadSectorUncached(CDImage::LBA lba, CDImage::SubChannelQ* subq, SectorBuffer* data); + bool ReadSectorUncached(CDImage::LBA lba, CDImage::Sector* sector, + CDImage::SectorReadMode mode = CDImage::SectorReadMode::DataAndSubQ); private: - void EmptyBuffers(); - bool ReadSectorIntoBuffer(std::unique_lock& lock); + u32 GetCurrentBufferLocked() const; + u32 GetBufferedSectorCountLocked() const; + void UpdatePublishedCacheStateLocked(); + void EmptyBuffersLocked(); + bool ReadSectorBatch(std::unique_lock& lock); void ReadSectorNonThreaded(CDImage::LBA lba); - bool InternalReadSectorUncached(CDImage::LBA lba, CDImage::SubChannelQ* subq, SectorBuffer* data); - void CancelReadahead(); + bool InternalReadSectorUncached(CDImage::LBA lba, CDImage::Sector* sector, CDImage::SectorReadMode mode); + void CancelReadaheadLocked(std::unique_lock& lock); void WorkerThreadEntryPoint(); std::unique_ptr m_media; - std::mutex m_mutex; + // Protects ring topology, request transitions, and condition-variable predicates. It is never held while accessing + // the CDImage backend; m_is_reading reserves exclusive backend access across those unlocked operations. + mutable std::mutex m_mutex; std::thread m_read_thread; std::condition_variable m_do_read_cv; std::condition_variable m_notify_read_complete_cv; - std::atomic m_next_position{}; - std::atomic_bool m_next_position_set{false}; - std::atomic_bool m_shutdown_flag{true}; - - std::atomic_bool m_is_reading{false}; - std::atomic_bool m_can_readahead{false}; - std::atomic_bool m_seek_error{false}; - - std::vector m_buffers; - std::atomic m_buffer_front{0}; - std::atomic m_buffer_back{0}; - std::atomic m_buffer_count{0}; + std::optional m_next_position; + CDImage::LBA m_next_read_lba = 0; + u64 m_request_generation = 0; + bool m_shutdown_flag = true; + bool m_is_reading = false; + bool m_can_readahead = false; + + // Published while m_mutex is held. An acquire load of m_buffered_sector_count makes the selected slot contents and + // m_published_buffer visible to the lock-free cached read path. + std::atomic m_published_buffer{0}; + std::atomic m_buffered_sector_count{0}; + std::atomic m_cached_behind_count{0}; + std::atomic_bool m_sector_borrowed{false}; + + // Core-thread-owned pin for the returned reference. The worker can append or evict history while borrowed, but + // preserves the physical slot selected by m_published_buffer until the core queues another position. + u32 m_borrowed_buffer = 0; + + std::vector m_buffers; + // Worker-owned staging storage keeps the ring immutable while a batch read is in flight. + std::vector m_read_buffer; + u32 m_readahead_count = 0; + u32 m_buffer_front = 0; + u32 m_buffer_back = 0; + u32 m_buffer_count = 0; + u32 m_buffer_current_offset = 0; }; diff --git a/src/util-tests/cd_image_tests.cpp b/src/util-tests/cd_image_tests.cpp index 0433434d0..4c0061621 100644 --- a/src/util-tests/cd_image_tests.cpp +++ b/src/util-tests/cd_image_tests.cpp @@ -12,11 +12,14 @@ #include #include +#include #include #include #include #include #include +#include +#include namespace { @@ -29,6 +32,8 @@ public: { } + explicit TempFile(std::string path) : m_path(std::move(path)) {} + ~TempFile() { if (!m_path.empty()) @@ -79,6 +84,110 @@ void ExpectSyncAndHeader(const std::array& sector, EXPECT_EQ(sector[15], mode); } +class SplitIndexCDImage final : public CDImage +{ +public: + struct ReadCall + { + LBA lba; + u32 count; + SectorReadMode mode; + }; + + SplitIndexCDImage() + { + Track track = {}; + track.track_number = 1; + track.first_index = 0; + track.length = 5; + track.mode = TrackMode::Audio; + track.control = SubChannelQ::Control(0); + m_tracks.push_back(track); + + Index first = {}; + first.file_sector_size = RAW_SECTOR_SIZE; + first.track_number = 1; + first.index_number = 1; + first.length = 2; + first.mode = TrackMode::Audio; + first.control = track.control; + m_indices.push_back(first); + + Index second = first; + second.start_lba_on_disc = 2; + second.start_lba_in_track = 2; + second.index_number = 2; + second.length = 3; + m_indices.push_back(second); + + m_lba_count = track.length; + AddLeadOutIndex(); + } + + u32 ReadSectorsFromIndex(std::span sectors, const Index& index, LBA lba_in_index, + SectorReadMode mode) override + { + EXPECT_NE(mode, SectorReadMode::SubQOnly); + const LBA first_lba = index.start_lba_on_disc + lba_in_index; + m_read_calls.push_back({first_lba, static_cast(sectors.size()), mode}); + for (u32 i = 0; i < sectors.size(); i++) + sectors[i].data.fill(static_cast(first_lba + i)); + return static_cast(sectors.size()); + } + + std::vector m_read_calls; +}; + +class PatchParentCDImage final : public CDImage +{ +public: + struct ReadCall + { + LBA lba; + u32 count; + }; + + PatchParentCDImage() + { + Track track = {}; + track.track_number = 1; + track.first_index = 0; + track.length = 4; + track.mode = TrackMode::Audio; + track.control = SubChannelQ::Control(0); + m_tracks.push_back(track); + + Index index = {}; + index.file_sector_size = RAW_SECTOR_SIZE; + index.track_number = 1; + index.index_number = 1; + index.length = track.length; + index.mode = track.mode; + index.control = track.control; + m_indices.push_back(index); + + m_lba_count = track.length; + AddLeadOutIndex(); + } + + u32 ReadSectorsFromIndex(std::span sectors, const Index& index, LBA lba_in_index, + SectorReadMode mode) override + { + const LBA first_lba = index.start_lba_on_disc + lba_in_index; + m_read_calls.push_back({first_lba, static_cast(sectors.size())}); + if (m_fail_reads) + return 0; + + EXPECT_NE(mode, SectorReadMode::SubQOnly); + for (u32 i = 0; i < sectors.size(); i++) + sectors[i].data.fill(static_cast(first_lba + i)); + return static_cast(sectors.size()); + } + + std::vector m_read_calls; + bool m_fail_reads = false; +}; + } // namespace TEST(CDImage, ConvertMode1ToRaw) @@ -216,6 +325,147 @@ TEST(CDImage, BatchReadUsesExplicitLBA) 0u); } +TEST(CDImage, BatchReadSplitsAtIndexBoundaries) +{ + SplitIndexCDImage image; + std::array sectors; + ASSERT_EQ(image.ReadSectors(1, sectors, CDImage::SectorReadMode::DataAndSubQ), sectors.size()); + + ASSERT_EQ(image.m_read_calls.size(), 2u); + EXPECT_EQ(image.m_read_calls[0].lba, 1u); + EXPECT_EQ(image.m_read_calls[0].count, 1u); + EXPECT_EQ(image.m_read_calls[1].lba, 2u); + EXPECT_EQ(image.m_read_calls[1].count, 3u); + + for (u32 i = 0; i < sectors.size(); i++) + { + EXPECT_EQ(sectors[i].data.front(), i + 1); + EXPECT_TRUE(sectors[i].subq.IsCRCValid()); + EXPECT_EQ(sectors[i].subq.index_number_bcd, BinaryToBCD(static_cast(i == 0 ? 1 : 2))); + } +} + +TEST(CDImage, GeneratedSubQOnlySkipsBackendRead) +{ + SplitIndexCDImage image; + CDImage::Sector sector; + sector.data.fill(0x5A); + + ASSERT_EQ(image.ReadSectors(1, std::span(§or, 1), CDImage::SectorReadMode::SubQOnly), 1u); + EXPECT_TRUE(image.m_read_calls.empty()); + EXPECT_TRUE(sector.subq.IsCRCValid()); + EXPECT_TRUE(std::all_of(sector.data.begin(), sector.data.end(), [](u8 value) { return value == 0x5A; })); +} + +TEST(CDImage, DataOnlyDoesNotGenerateSubQ) +{ + SplitIndexCDImage image; + CDImage::Sector sector; + sector.subq.data.fill(0x5A); + + ASSERT_EQ(image.ReadSectors(1, std::span(§or, 1), CDImage::SectorReadMode::DataOnly), 1u); + ASSERT_EQ(image.m_read_calls.size(), 1u); + EXPECT_EQ(image.m_read_calls.front().mode, CDImage::SectorReadMode::DataOnly); + EXPECT_EQ(sector.data.front(), 1u); + EXPECT_TRUE(std::all_of(sector.subq.data.begin(), sector.subq.data.end(), [](u8 value) { return value == 0x5A; })); +} + +TEST(CDImage, CCDSubQOnlyDoesNotDependOnImageData) +{ + TempFile ccd("duckstation_cd_image_subq_only", "ccd"); + TempFile img(Path::ReplaceExtension(ccd.GetPath(), "img")); + TempFile sub(Path::ReplaceExtension(ccd.GetPath(), "sub")); + + // The image deliberately contains no readable sector. A valid SUB record must still be independently readable. + ASSERT_TRUE(img.Write(std::span())); + SplitIndexCDImage subq_generator; + CDImage::SubChannelQ expected_subq; + ASSERT_TRUE(subq_generator.GenerateSubChannelQ(&expected_subq, 1)); + std::array subcode = {}; + std::memcpy(subcode.data() + CDImage::SUBCHANNEL_BYTES_PER_FRAME, expected_subq.data.data(), + expected_subq.data.size()); + ASSERT_TRUE(sub.Write(subcode)); + + static constexpr std::string_view ccd_data = "[CloneCD]\n" + "Version=3\n" + "[Disc]\n" + "TocEntries=3\n" + "[Entry 0]\n" + "Point=0xA0\n" + "[Entry 1]\n" + "Point=1\n" + "ADR=1\n" + "Control=4\n" + "PLBA=0\n" + "[Entry 2]\n" + "Point=0xA2\n" + "PLBA=1\n" + "[TRACK 1]\n" + "MODE=1\n" + "INDEX 1=0\n"; + ASSERT_TRUE(ccd.WriteString(ccd_data)); + + Error error; + std::unique_ptr image = CDImage::OpenCCDImage(ccd.GetPath().c_str(), &error); + ASSERT_TRUE(image) << error.GetDescription(); + + CDImage::Sector sector; + EXPECT_EQ(image->ReadSectors(2 * CDImage::FRAMES_PER_SECOND, std::span(§or, 1), + CDImage::SectorReadMode::DataAndSubQ), + 0u); + + sector.data.fill(0x5A); + ASSERT_EQ(image->ReadSectors(2 * CDImage::FRAMES_PER_SECOND, std::span(§or, 1), + CDImage::SectorReadMode::SubQOnly), + 1u); + EXPECT_EQ(sector.subq.data, expected_subq.data); + EXPECT_TRUE(std::all_of(sector.data.begin(), sector.data.end(), [](u8 value) { return value == 0x5A; })); +} + +TEST(CDImage, PPFReplacementSectorsAreAuthoritative) +{ + auto parent = std::make_unique(); + PatchParentCDImage* parent_ptr = parent.get(); + + // A minimal PPF1 patch replacing byte 7 of sector 1. Loading materializes the entire source sector once. + std::vector patch(56, 0); + std::memcpy(patch.data(), "PPF1", 4); + const u32 patch_offset = CDImage::RAW_SECTOR_SIZE + 7; + const u8 patch_size = 1; + const u8 patch_value = 0xCC; + patch.insert(patch.end(), reinterpret_cast(&patch_offset), + reinterpret_cast(&patch_offset) + sizeof(patch_offset)); + patch.push_back(patch_size); + patch.push_back(patch_value); + + TempFile ppf("duckstation_cd_image_authoritative", "ppf"); + ASSERT_TRUE(ppf.Write(patch)); + + Error error; + std::unique_ptr image = CDImage::OverlayPPFPatch(ppf.GetPath().c_str(), std::move(parent), &error); + ASSERT_TRUE(image) << error.GetDescription(); + + // Once materialized, a replacement no longer depends on the parent sector remaining readable. + parent_ptr->m_fail_reads = true; + CDImage::Sector replaced_sector; + ASSERT_EQ( + image->ReadSectors(1, std::span(&replaced_sector, 1), CDImage::SectorReadMode::DataAndSubQ), 1u); + EXPECT_EQ(replaced_sector.data[0], 1u); + EXPECT_EQ(replaced_sector.data[7], patch_value); + + // A batch is split around replacement slots, so the parent sees only the unpatched gaps. + parent_ptr->m_fail_reads = false; + parent_ptr->m_read_calls.clear(); + std::array sectors; + ASSERT_EQ(image->ReadSectors(0, sectors, CDImage::SectorReadMode::DataAndSubQ), sectors.size()); + ASSERT_EQ(parent_ptr->m_read_calls.size(), 2u); + EXPECT_EQ(parent_ptr->m_read_calls[0].lba, 0u); + EXPECT_EQ(parent_ptr->m_read_calls[0].count, 1u); + EXPECT_EQ(parent_ptr->m_read_calls[1].lba, 2u); + EXPECT_EQ(parent_ptr->m_read_calls[1].count, 2u); + EXPECT_EQ(sectors[1].data[7], patch_value); +} + TEST(CDImage, Iso2048DetectedAsMode1) { TempFile iso("duckstation_cd_image_mode1", "iso");