CHD: decode ahead per stream, so CD audio and data do not collide

A CHD was read through one decoded hunk and a thread that decoded one
hunk ahead. A game that plays CD audio while it streams data reads the
two from different places a sector at a time, so each threw out the
other's hunk and aimed the thread away from it, and the hunks were
decoded on the emulation thread: a stutter a GDI never shows.

Each stream (data tracks, audio tracks) now reads through its own
libretro-common vfs_prefetch, whose thread decodes the hunks ahead on an
image handle of its own, one window per hunk; sectors are lent from the
windows in place, and a read command's Prefetch() sets the data
stream's thread going before the first sector is asked for. The core
defines HAVE_THREADS where it builds threads, so the prefetcher is not
compiled to stubs. tools/chd/chd_streams_test, run from tools/chd/run.sh
and now in CI, reads audio and data at once and fails when the reading
thread does the decoding.
This commit is contained in:
libretroadmin committed 2026-10-26 14:49:16 +01:00
1 parent 8ad6b82d41
commit f62d495d2d
12 files changed
+900 -290

No files matched your search

+50
View File
@@ -0,0 +1,50 @@
name: CHD reading
# tools/chd/run.sh: every frame of CHDs made with each CD codec, a GD-ROM
# image and a parent chain against the files they came from; and CD audio
# and data read at once through chd.cpp, where the read-ahead threads must
# decode the hunks and neither stream throw out the other's (ASan + UBSan,
# TSan).
on:
push:
paths:
- 'core/imgread/**'
- 'core/deps/coreio/**'
- 'core/archive/**'
- 'core/libretro-common/formats/**'
- 'core/libretro-common/encodings/**'
- 'core/libretro-common/vfs/**'
- 'core/libretro-common/streams/**'
- 'core/libretro-common/rthreads/**'
- 'tools/chd/**'
- '.github/workflows/chd.yml'
pull_request:
paths:
- 'core/imgread/**'
- 'core/deps/coreio/**'
- 'core/archive/**'
- 'core/libretro-common/formats/**'
- 'core/libretro-common/encodings/**'
- 'core/libretro-common/vfs/**'
- 'core/libretro-common/streams/**'
- 'core/libretro-common/rthreads/**'
- 'tools/chd/**'
- '.github/workflows/chd.yml'
workflow_dispatch:
permissions:
contents: read
jobs:
chd:
runs-on: ubuntu-24.04
timeout-minutes: 30
steps:
- uses: actions/checkout@v4
- name: Install chdman
run: |
sudo apt-get update
sudo apt-get install -y --no-install-recommends build-essential mame-tools python3
- name: CHD reads, and CD audio and data at once
run: sh tools/chd/run.sh
+1
View File
@@ -1168,6 +1168,7 @@ endif
ifeq ($(NO_THREADS),1)
CORE_DEFINES += -DTARGET_NO_THREADS
else
CORE_DEFINES += -DHAVE_THREADS
NEED_PTHREAD=1
endif
+1
View File
@@ -438,6 +438,7 @@ SOURCES_C += $(LIBRETRO_COMM_DIR)/memmap/memalign.c \
$(LIBRETRO_COMM_DIR)/file/file_path.c \
$(LIBRETRO_COMM_DIR)/file/file_path_io.c \
$(LIBRETRO_COMM_DIR)/vfs/vfs_implementation.c \
$(LIBRETRO_COMM_DIR)/vfs/vfs_prefetch.c \
$(LIBRETRO_COMM_DIR)/streams/file_stream.c \
$(LIBRETRO_COMM_DIR)/streams/file_stream_transforms.c \
$(LIBRETRO_COMM_DIR)/encodings/encoding_utf.c \
+98 -67
View File
@@ -1,11 +1,21 @@
#include "common.h"
#include "chd_image.h"
#include "hunk_prefetch.h"
#include <vfs/vfs_prefetch.h>
/* tracks are padded to a multiple of this many frames */
const uint32_t CD_TRACK_PADDING = 4;
/* Which read-ahead a track reads through. CD audio and data are read
* from different places on the disc at once (music playing while a
* game streams), so each has its own, and neither throws out the
* hunks the other is about to use. */
enum { CHD_STREAM_DATA, CHD_STREAM_AUDIO, CHD_STREAMS };
/* Hunks each stream's thread decodes ahead of its reader. */
static const unsigned chd_stream_slots[CHD_STREAMS] = { 8, 4 };
struct CHDDisc : Disc
{
chd_image_t* chd;
@@ -14,16 +24,30 @@ struct CHDDisc : Disc
u32 hunkbytes;
u32 sph;
u32 total_hunks = 0;
/* Read-ahead on a worker thread, with a handle of its own. */
chd_image_t* chd2 = nullptr;
HunkPrefetch prefetch;
static const u32 NO_HUNK = HunkPrefetch::NO_HUNK;
u32 total_hunks = 0;
static bool prefetch_read(void *opaque, u32 hunk, u8 *dst)
/* A stream's read-ahead: libretro-common's prefetch thread, which
* decodes the hunks ahead of the reader on an image handle of its
* own. One window is one hunk. */
struct Stream
{
return chd_image_read_hunk(((CHDDisc *)opaque)->chd2, hunk, dst);
CHDDisc* disc;
chd_image_t* chd;
struct vfs_prefetch* p;
} streams[CHD_STREAMS];
static const u32 NO_HUNK = 0xFFFFFFFFu;
static int64_t stream_fetch(void *user, uint64_t off, uint8_t *buf, size_t len)
{
Stream *st = (Stream *)user;
u32 hunk = (u32)(off / st->disc->hunkbytes);
if (hunk >= st->disc->total_hunks)
return 0;
if (len < st->disc->hunkbytes || !chd_image_read_hunk(st->chd, hunk, buf))
return -1;
return (int64_t)st->disc->hunkbytes;
}
/* Decompress @hunk into hunk_mem on this thread. A hunk that fails to
@@ -40,66 +64,75 @@ struct CHDDisc : Disc
}
}
/* Return a pointer to the decompressed hunk, preferring the prefetched copy. */
u8* hunk_get(u32 hunk)
/* Where the sector at @fad_offs is decoded: in the stream's window
* when its thread got there first, else decoded here. Valid until
* the next read on this disc. */
const u8* sector_get(unsigned stream, u32 fad_offs)
{
if (hunk == old_hunk)
return hunk_mem; /* still the current hunk */
if (prefetch.Running())
{
/* Takes the prefetched hunk if there is one and aims the worker
* at the next; on a miss (first read or seek) the worker gets
* going on the next hunk while this thread decodes this one. */
bool hit = prefetch.Fetch(hunk);
hunk_mem = prefetch.Buffer();
old_hunk = hit ? hunk : NO_HUNK;
if (hit)
return hunk_mem;
}
hunk_read_sync(hunk);
return hunk_mem;
const u32 hunk = fad_offs / sph;
const u32 in = (fad_offs % sph) * (2352 + 96);
const u8* ptr;
if (streams[stream].p
&& vfs_prefetch_view(streams[stream].p, (uint64_t)hunk * hunkbytes + in, &ptr) > 0)
return ptr;
if (hunk != old_hunk)
hunk_read_sync(hunk);
return hunk_mem + in;
}
void start_prefetch(const char* file)
/* The stream's thread starts on @hunk now, ahead of the reads. */
void stream_aim(unsigned stream, u32 hunk)
{
chd2 = chd_image_open(file, NULL, 0);
if (chd2 == nullptr)
return;
if (!prefetch.Start(prefetch_read, this, hunkbytes, total_hunks))
{
chd_image_close(chd2);
chd2 = nullptr;
return;
}
/* The prefetcher's buffers take over from the one TryOpen made. */
delete [] hunk_mem;
hunk_mem = prefetch.Buffer();
old_hunk = NO_HUNK;
if (streams[stream].p)
vfs_prefetch_seek(streams[stream].p, (uint64_t)hunk * hunkbytes);
}
void stop_prefetch()
void start_streams(const char* file)
{
if (prefetch.Running())
for (unsigned i = 0; i < CHD_STREAMS; i++)
{
prefetch.Stop();
hunk_mem = nullptr;
old_hunk = NO_HUNK;
Stream& st = streams[i];
st.disc = this;
if (!(st.chd = chd_image_open(file, NULL, 0)))
continue;
if (!(st.p = vfs_prefetch_new(stream_fetch, &st, hunkbytes,
chd_stream_slots[i], 0)))
{
chd_image_close(st.chd);
st.chd = nullptr;
}
}
}
void stop_streams()
{
for (unsigned i = 0; i < CHD_STREAMS; i++)
{
vfs_prefetch_free(streams[i].p);
streams[i].p = nullptr;
chd_image_close(streams[i].chd);
streams[i].chd = nullptr;
}
chd_image_close(chd2);
chd2 = nullptr;
}
CHDDisc()
{
chd=0;
hunk_mem=0;
for (unsigned i = 0; i < CHD_STREAMS; i++)
{
streams[i].disc = this;
streams[i].chd = nullptr;
streams[i].p = nullptr;
}
}
bool TryOpen(const char* file);
~CHDDisc()
{
stop_prefetch();
stop_streams();
if (hunk_mem)
delete [] hunk_mem;
chd_image_close(chd);
@@ -112,31 +145,28 @@ struct CHDTrack : TrackFile
u32 StartFAD;
u32 Offset;
u32 fmt;
unsigned stream;
bool swap_bytes;
CHDTrack(CHDDisc* disc, u32 StartFAD,u32 Offset, u32 fmt, bool swap_bytes)
CHDTrack(CHDDisc* disc, u32 StartFAD,u32 Offset, u32 fmt, unsigned stream, bool swap_bytes)
{
this->disc=disc;
this->StartFAD=StartFAD;
this->Offset=Offset;
this->fmt=fmt;
this->stream=stream;
this->swap_bytes = swap_bytes;
}
virtual void Read(u32 FAD,u8* dst,SectorFormat* sector_type,u8* subcode,SubcodeFormat* subcode_type)
{
u32 fad_offs = FAD + Offset;
u32 hunk=(fad_offs)/disc->sph;
u8* hmem = disc->hunk_get(hunk);
u32 hunk_ofs=fad_offs%disc->sph;
const u8* from = disc->sector_get(stream, FAD + Offset);
if (swap_bytes)
{
/* The other byte order, made on the way across: 16 bits at a
* time, which the compiler does sixteen bytes at a time. (It
* was copied, and then swapped where it lay a byte at a time.) */
const u8 *from = hmem + hunk_ofs * (2352 + 96);
for (u32 i = 0; i + 1 < fmt; i += 2)
{
u16 w;
@@ -146,7 +176,7 @@ struct CHDTrack : TrackFile
}
}
else
memcpy(dst,hmem+hunk_ofs*(2352+96),fmt);
memcpy(dst,from,fmt);
*sector_type=Format();
@@ -162,29 +192,28 @@ struct CHDTrack : TrackFile
}
/* The compressed bytes of every hunk the sectors span, so neither
* this thread nor the read-ahead worker stalls on a cold page. Both
* handles are on the same file, so one hint serves both. */
* this thread nor the read-ahead stalls on a cold page (all handles
* are on the same file, so one hint serves them all), and the
* stream's thread set decoding from the first one. */
virtual void Prefetch(u32 FAD, u32 count)
{
u32 first = (FAD + Offset) / disc->sph;
u32 last = (FAD + Offset + count - 1) / disc->sph;
if (count)
chd_image_prefetch(disc->chd, first, last - first + 1);
if (!count)
return;
chd_image_prefetch(disc->chd, first, last - first + 1);
disc->stream_aim(stream, first);
}
/* The sector inside the decoded hunk; a track whose bytes have to
* be swapped is read through a buffer. */
virtual const u8* View(u32 FAD, SectorFormat* sector_type)
{
u32 fad_offs = FAD + Offset;
u8* hmem;
if (swap_bytes)
return NULL;
hmem = disc->hunk_get(fad_offs / disc->sph);
*sector_type = Format();
return hmem + (fad_offs % disc->sph) * (2352 + 96);
return disc->sector_get(stream, FAD + Offset);
}
};
@@ -275,7 +304,9 @@ bool CHDDisc::TryOpen(const char* file)
last_pregap = pregap;
/* Music is kept in the other byte order in a version 5 image (and
* in any of a CD, which is read from version 5 only). */
t.file = new CHDTrack(this, t.StartFAD, Offset + pregap_kept - t.StartFAD, sector_size, t.CTRL == 0 && chd_image_version(chd) >= 5);
t.file = new CHDTrack(this, t.StartFAD, Offset + pregap_kept - t.StartFAD, sector_size,
t.CTRL == 0 ? CHD_STREAM_AUDIO : CHD_STREAM_DATA,
t.CTRL == 0 && chd_image_version(chd) >= 5);
int padded = (frames + CD_TRACK_PADDING - 1) / CD_TRACK_PADDING;
Offset += padded * CD_TRACK_PADDING;
@@ -334,7 +365,7 @@ bool CHDDisc::TryOpen(const char* file)
}
}
start_prefetch(file);
start_streams(file);
return true;
}
-151
View File
@@ -1,151 +0,0 @@
#pragma once
/* Read-ahead for hunk-compressed disc images.
*
* A worker thread decodes the hunk the reader is expected to want next,
* so a sequential read finds it already decoded. Decoding is deterministic,
* so the reader gets the same bytes whichever thread produced them; only
* the CPU time moves.
*
* The two threads share three hunk buffers through a cTripleBuffer and
* never wait for one another: the reader either finds the hunk it asked
* for or decodes it itself, and the worker sleeps on an eventcount until
* the reader aims it at a new hunk. No lock is taken anywhere. */
#include "types.h"
#include "lockfree.h"
struct HunkPrefetch
{
/* Decode @hunk into @dst on the worker thread. false: could not. */
typedef bool (*ReadFn)(void *opaque, u32 hunk, u8 *dst);
static const u32 NO_HUNK = 0xFFFFFFFFu;
HunkPrefetch() : thread(NULL)
{
buf[0].mem = buf[1].mem = buf[2].mem = NULL;
}
~HunkPrefetch() { Stop(); }
bool Running() const { return thread != NULL; }
bool Start(ReadFn read_fn, void *read_opaque, u32 hunkbytes, u32 hunk_count)
{
if (thread != NULL)
return false;
for (int i = 0; i < 3; i++)
{
buf[i].mem = new u8[hunkbytes];
buf[i].hunk = NO_HUNK;
}
tri.Reset();
read = read_fn;
opaque = read_opaque;
total_hunks = hunk_count;
retro_atomic_int_init(&request, (int)NO_HUNK);
retro_atomic_int_init(&ready, (int)NO_HUNK);
retro_atomic_int_init(&stop, 0);
thread = sthread_create(worker_entry, this);
if (thread == NULL)
free_buffers();
return thread != NULL;
}
void Stop()
{
if (thread == NULL)
return;
retro_atomic_store_release_int(&stop, 1);
ec.Notify();
sthread_join(thread);
thread = NULL;
free_buffers();
}
/* Reader: the buffer the reader owns right now. Fetch() may change
* which one that is. */
u8* Buffer() { return buf[tri.Front()].mem; }
/* Reader: true when Buffer() now holds @hunk, decoded by the worker.
* false when it does not, and what Buffer() holds is then unknown;
* the reader decodes into it itself. Either way the worker is aimed
* at the hunk after @hunk. */
bool Fetch(u32 hunk)
{
bool hit = false;
/* ready only says a take is worth trying; the buffer's own tag,
* read once the buffer is ours, is what counts. */
if ((u32)retro_atomic_load_acquire_int(&ready) == hunk && tri.Take())
hit = buf[tri.Front()].hunk == hunk;
retro_atomic_store_release_int(&request, (int)(hunk + 1));
ec.Notify();
return hit;
}
private:
struct
{
u8 *mem;
u32 hunk; /* written by the worker before it publishes the buffer */
} buf[3];
cTripleBuffer tri;
retro_atomic_int_t request; /* hunk the reader wants decoded ahead */
retro_atomic_int_t ready; /* hunk the worker published last */
retro_atomic_int_t stop;
cEventCount ec;
sthread_t *thread;
ReadFn read;
void *opaque;
u32 total_hunks;
void free_buffers()
{
for (int i = 0; i < 3; i++)
{
delete [] buf[i].mem;
buf[i].mem = NULL;
}
}
static void worker_entry(void *p) { ((HunkPrefetch *)p)->worker(); }
void worker()
{
/* The last request answered, decoded or not: it is not tried
* again until the reader asks for something else. */
u32 done = NO_HUNK;
for (;;)
{
u32 req;
for (;;)
{
int key;
if (retro_atomic_load_acquire_int(&stop))
return;
req = (u32)retro_atomic_load_acquire_int(&request);
if (req != NO_HUNK && req != done)
break;
key = ec.Prepare();
req = (u32)retro_atomic_load_acquire_int(&request);
if (retro_atomic_load_acquire_int(&stop)
|| (req != NO_HUNK && req != done))
ec.Cancel();
else
ec.Commit(key);
}
if (req < total_hunks && read(opaque, req, buf[tri.Back()].mem))
{
buf[tri.Back()].hunk = req;
retro_atomic_store_release_int(&ready, (int)req);
tri.Publish();
}
done = req;
}
}
};
@@ -0,0 +1,102 @@
/* Copyright (C) 2026 The RetroArch team
*
* ---------------------------------------------------------------------------------------
* The following license statement only applies to this file (vfs_prefetch.h).
* ---------------------------------------------------------------------------------------
*
* Permission is hereby granted, free of charge,
* to any person obtaining a copy of this software and associated documentation files (the "Software"),
* to deal in the Software without restriction, including without limitation the rights to
* use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software,
* and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
* INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
* IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
* WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#ifndef __LIBRETRO_SDK_VFS_PREFETCH_H
#define __LIBRETRO_SDK_VFS_PREFETCH_H
#include <stdint.h>
#include <stddef.h>
#include <retro_common_api.h>
RETRO_BEGIN_DECLS
/* A background prefetcher for a file read over a network: a thread of
* its own fetches the windows ahead of the reader on a connection of
* its own, so a run of sequential reads is served from memory and the
* reading thread never waits for the network. The reader's side is
* lock-free: a read copies from a ready window or reports a miss (and
* the caller reads the network itself, as it would without this), and
* a seek out of the fetched range retargets the thread.
*
* Only built with HAVE_THREADS; without it vfs_prefetch_new() returns
* NULL and everything else is a no-op, so callers keep one code path. */
struct vfs_prefetch;
/* Read @len octets at @off into @buf; the octets read (short at end of
* file), -1 on failure. Runs on the prefetch thread, on that thread's
* own connection. */
typedef int64_t (*vfs_prefetch_fetch_t)(void *user, uint64_t off, uint8_t *buf, size_t len);
/**
* vfs_prefetch_new:
* @window : octets per window (a client's read-ahead size fits)
* @slots : windows kept ahead, 2 or more; memory is window x slots
* @start : the offset to begin at (0 for a fresh file)
*
* Starts the thread. NULL when threads are unavailable or memory is.
**/
struct vfs_prefetch *vfs_prefetch_new(vfs_prefetch_fetch_t fetch, void *user,
size_t window, unsigned slots, uint64_t start);
/**
* vfs_prefetch_read:
*
* Copy up to @len octets at @off from the fetched windows. Returns the
* octets copied (fewer than @len when the range runs past what is
* ready, 0 when a ready window says the file ends there), or -1 when
* nothing at @off is ready: the caller then reads the network itself.
* Consumed windows are recycled as the reader moves past them.
**/
int64_t vfs_prefetch_read(struct vfs_prefetch *p, uint64_t off, uint8_t *buf, size_t len);
/**
* vfs_prefetch_view:
*
* vfs_prefetch_read() without the copy: points @ptr at the octets at
* @off in a ready window and returns how many follow there, to the end
* of that window; 0 at the end of the file, -1 on a miss, which
* retargets the thread as a read's miss does. Only the reader recycles
* windows, so @ptr stays valid until its next read, view or seek.
**/
int64_t vfs_prefetch_view(struct vfs_prefetch *p, uint64_t off, const uint8_t **ptr);
/**
* vfs_prefetch_seek:
*
* The reader will continue at @off: when that is outside what is
* fetched or being fetched, the thread starts over there.
**/
void vfs_prefetch_seek(struct vfs_prefetch *p, uint64_t off);
/**
* vfs_prefetch_free:
*
* Stops the thread (waiting for a fetch in flight to end) and frees
* everything. @user is the caller's to release afterwards.
**/
void vfs_prefetch_free(struct vfs_prefetch *p);
RETRO_END_DECLS
#endif
+412
View File
@@ -0,0 +1,412 @@
/* Copyright (C) 2026 The RetroArch team
*
* ---------------------------------------------------------------------------------------
* The following license statement only applies to this file (vfs_prefetch.c).
* ---------------------------------------------------------------------------------------
*
* Permission is hereby granted, free of charge,
* to any person obtaining a copy of this software and associated documentation files (the "Software"),
* to deal in the Software without restriction, including without limitation the rights to
* use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software,
* and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
* INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
* IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
* WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include <stdlib.h>
#include <string.h>
#include <vfs/vfs_prefetch.h>
#ifdef HAVE_THREADS
#include <retro_atomic.h>
#include <rthreads/rthreads.h>
#include <rthreads/retro_eventcount.h>
/* Slot states. The thread moves EMPTY -> FILLING -> READY; only the
* reader moves READY -> EMPTY (a window it has passed, or one from a
* stream it left), so every write the thread makes to a slot comes
* after the reader's CAS and every read the reader makes comes after
* the thread's release: no field is ever touched from both sides at
* once, and neither side holds a lock. */
#define SLOT_EMPTY 0
#define SLOT_FILLING 1
#define SLOT_READY 2
struct vfs_prefetch_slot
{
uint8_t *buf;
uint64_t off;
size_t len;
int gen; /* the stream this window belongs to */
int eof; /* the file ends inside this window */
retro_atomic_int_t state;
};
struct vfs_prefetch
{
vfs_prefetch_fetch_t fetch;
void *user;
size_t window;
unsigned nslots;
struct vfs_prefetch_slot *slot;
/* The stream: a generation and the window index it starts at
* (windows are aligned to their size, so an index is enough and it
* fits an atomic int: 2^31 windows of 1 MiB). The reader stores the
* target index then bumps gen (release); the thread loads gen
* (acquire) and then the index. Windows carry the gen they were
* fetched under, so the reader ignores stale ones and the thread
* recycles them when it notices the bump. */
retro_atomic_int_t gen;
retro_atomic_int_t target_idx;
/* the window index (offset / window) the thread fetches next, so the
* reader can tell "not yet fetched" from "elsewhere" without a lock */
retro_atomic_int_t next_idx;
/* the gen next_idx belongs to: until the thread has taken a
* retarget, the stream is headed for target_idx instead */
retro_atomic_int_t seen_gen;
retro_atomic_int_t stop;
retro_eventcount_t wake; /* the thread parks here */
sthread_t *thread;
int thread_gen;
uint64_t next_off;
int at_eof;
};
static void vfs_prefetch_thread(void *data)
{
struct vfs_prefetch *p = (struct vfs_prefetch*)data;
for (;;)
{
unsigned i;
int gen;
struct vfs_prefetch_slot *s = NULL;
int64_t n;
if (retro_atomic_load_acquire_int(&p->stop))
return;
/* a retarget: start over there; the reader recycles the windows
* of the old stream itself as it finds them */
gen = retro_atomic_load_acquire_int(&p->gen);
if (gen != p->thread_gen)
{
p->thread_gen = gen;
p->next_off = (uint64_t)retro_atomic_load_acquire_int(&p->target_idx) * p->window;
p->at_eof = 0;
retro_atomic_store_release_int(&p->next_idx, (int)(p->next_off / p->window));
retro_atomic_store_release_int(&p->seen_gen, gen);
}
/* work: an empty slot while the stream has not ended */
if (!p->at_eof)
{
for (i = 0; i < p->nslots; i++)
{
if (retro_atomic_cas_int(&p->slot[i].state, SLOT_EMPTY, SLOT_FILLING))
{
s = &p->slot[i];
break;
}
}
}
if (!s)
{
/* nothing to do until the reader frees a window or seeks */
int key = retro_eventcount_prepare_wait(&p->wake);
if (retro_atomic_load_acquire_int(&p->stop)
|| retro_atomic_load_acquire_int(&p->gen) != p->thread_gen)
{
retro_eventcount_cancel_wait(&p->wake);
continue;
}
if (!p->at_eof)
{
for (i = 0; i < p->nslots; i++)
if (retro_atomic_load_acquire_int(&p->slot[i].state) == SLOT_EMPTY)
break;
if (i < p->nslots)
{
retro_eventcount_cancel_wait(&p->wake);
continue;
}
}
retro_eventcount_commit_wait(&p->wake, key);
continue;
}
s->off = p->next_off;
s->gen = p->thread_gen;
n = p->fetch(p->user, s->off, s->buf, p->window);
if (n < 0)
{
/* a failed fetch: give the slot back; the reader's own read
* will report the failure, and this thread tries again later */
retro_atomic_store_release_int(&s->state, SLOT_EMPTY);
p->at_eof = 1; /* stop hammering; a seek restarts */
continue;
}
s->len = (size_t)n;
s->eof = (size_t)n < p->window;
if (s->eof)
p->at_eof = 1;
p->next_off += (uint64_t)n;
retro_atomic_store_release_int(&p->next_idx, (int)(p->next_off / p->window));
retro_atomic_store_release_int(&s->state, SLOT_READY);
}
}
struct vfs_prefetch *vfs_prefetch_new(vfs_prefetch_fetch_t fetch, void *user,
size_t window, unsigned slots, uint64_t start)
{
struct vfs_prefetch *p;
unsigned i;
if (!fetch || !window || slots < 2)
return NULL;
if (!(p = (struct vfs_prefetch*)calloc(1, sizeof(*p))))
return NULL;
p->fetch = fetch;
p->user = user;
p->window = window;
p->nslots = slots;
if (!(p->slot = (struct vfs_prefetch_slot*)calloc(slots, sizeof(*p->slot))))
{
free(p);
return NULL;
}
for (i = 0; i < slots; i++)
{
if (!(p->slot[i].buf = (uint8_t*)malloc(window)))
{
while (i--)
free(p->slot[i].buf);
free(p->slot);
free(p);
return NULL;
}
retro_atomic_store_release_int(&p->slot[i].state, SLOT_EMPTY);
}
retro_atomic_store_release_int(&p->target_idx, (int)(start / window));
/* gen 1 with the target set: the thread sees a retarget and starts there */
p->thread_gen = 0;
retro_atomic_store_release_int(&p->gen, 1);
retro_atomic_store_release_int(&p->next_idx, (int)(start / window));
retro_atomic_store_release_int(&p->seen_gen, 0);
retro_atomic_store_release_int(&p->stop, 0);
if (!retro_eventcount_init(&p->wake))
{
for (i = 0; i < slots; i++)
free(p->slot[i].buf);
free(p->slot);
free(p);
return NULL;
}
if (!(p->thread = sthread_create(vfs_prefetch_thread, p)))
{
retro_eventcount_free(&p->wake);
for (i = 0; i < slots; i++)
free(p->slot[i].buf);
free(p->slot);
free(p);
return NULL;
}
return p;
}
/* Recycles the windows the reader has moved past, and any left from a
* stream it seeked away from; returns the current generation. */
static int vfs_prefetch_recycle(struct vfs_prefetch *p, uint64_t off)
{
int gen = retro_atomic_load_acquire_int(&p->gen);
int freed = 0;
unsigned i;
for (i = 0; i < p->nslots; i++)
{
struct vfs_prefetch_slot *s = &p->slot[i];
if (retro_atomic_load_acquire_int(&s->state) == SLOT_READY
&& (s->gen != gen || (!s->eof && s->off + s->len <= off)))
{
if (retro_atomic_cas_int(&s->state, SLOT_READY, SLOT_EMPTY))
freed = 1;
}
}
if (freed)
retro_eventcount_notify(&p->wake);
return gen;
}
int64_t vfs_prefetch_view(struct vfs_prefetch *p, uint64_t off, const uint8_t **ptr)
{
int gen;
unsigned i;
if (!p || !ptr)
return -1;
gen = vfs_prefetch_recycle(p, off);
for (i = 0; i < p->nslots; i++)
{
struct vfs_prefetch_slot *s = &p->slot[i];
if (retro_atomic_load_acquire_int(&s->state) != SLOT_READY
|| s->gen != gen)
continue;
if (off >= s->off && off < s->off + s->len)
{
*ptr = s->buf + (off - s->off);
return (int64_t)(s->off + s->len - off);
}
if (s->eof && off >= s->off + s->len)
return 0;
}
vfs_prefetch_seek(p, off);
return -1;
}
int64_t vfs_prefetch_read(struct vfs_prefetch *p, uint64_t off, uint8_t *buf, size_t len)
{
int gen;
unsigned i;
size_t done = 0;
if (!p)
return -1;
gen = vfs_prefetch_recycle(p, off);
/* copy from ready windows covering @off, across as many as touch */
while (done < len)
{
struct vfs_prefetch_slot *hit = NULL;
uint64_t at = off + done;
for (i = 0; i < p->nslots; i++)
{
struct vfs_prefetch_slot *s = &p->slot[i];
if (retro_atomic_load_acquire_int(&s->state) == SLOT_READY
&& s->gen == gen && at >= s->off && at < s->off + s->len)
{
hit = s;
break;
}
/* a ready window that says the file ends before @at */
if (retro_atomic_load_acquire_int(&s->state) == SLOT_READY
&& s->gen == gen && s->eof && at >= s->off + s->len)
return (int64_t)done;
}
if (!hit)
{
/* a miss: if the thread is behind or elsewhere, it starts over
* here rather than fetching what the reader pulls itself */
if (!done)
vfs_prefetch_seek(p, off);
break;
}
{
size_t take = (size_t)(hit->off + hit->len - at);
if (take > len - done)
take = len - done;
memcpy(buf + done, hit->buf + (at - hit->off), take);
done += take;
if (hit->eof && at + take >= hit->off + hit->len)
return (int64_t)done;
}
}
return done ? (int64_t)done : -1;
}
void vfs_prefetch_seek(struct vfs_prefetch *p, uint64_t off)
{
int gen;
unsigned i;
int next;
if (!p)
return;
gen = retro_atomic_load_acquire_int(&p->gen);
/* covered by a ready window: nothing to do */
for (i = 0; i < p->nslots; i++)
{
struct vfs_prefetch_slot *s = &p->slot[i];
if (retro_atomic_load_acquire_int(&s->state) == SLOT_READY
&& s->gen == gen && off >= s->off && off < s->off + s->len)
return;
}
/* about to be fetched (the window being filled or the next one),
* judged by where the stream is headed: a retarget the thread has
* not taken yet replaces where it was */
next = retro_atomic_load_acquire_int(&p->seen_gen) == gen
? retro_atomic_load_acquire_int(&p->next_idx)
: retro_atomic_load_acquire_int(&p->target_idx);
if (off / p->window + 1 >= (uint64_t)next && off / p->window <= (uint64_t)next + 1)
return;
retro_atomic_store_release_int(&p->target_idx, (int)(off / p->window));
retro_atomic_store_release_int(&p->gen, gen + 1);
/* the old stream's ready windows go back now, so the thread has
* room to start; ones still filling come back through read() */
for (i = 0; i < p->nslots; i++)
{
struct vfs_prefetch_slot *s = &p->slot[i];
if (retro_atomic_load_acquire_int(&s->state) == SLOT_READY)
retro_atomic_cas_int(&s->state, SLOT_READY, SLOT_EMPTY);
}
retro_eventcount_notify(&p->wake);
}
void vfs_prefetch_free(struct vfs_prefetch *p)
{
unsigned i;
if (!p)
return;
retro_atomic_store_release_int(&p->stop, 1);
retro_eventcount_notify(&p->wake);
sthread_join(p->thread);
retro_eventcount_free(&p->wake);
for (i = 0; i < p->nslots; i++)
free(p->slot[i].buf);
free(p->slot);
free(p);
}
#else /* !HAVE_THREADS */
struct vfs_prefetch *vfs_prefetch_new(vfs_prefetch_fetch_t fetch, void *user,
size_t window, unsigned slots, uint64_t start)
{
(void)fetch; (void)user; (void)window; (void)slots; (void)start;
return NULL;
}
int64_t vfs_prefetch_read(struct vfs_prefetch *p, uint64_t off, uint8_t *buf, size_t len)
{
(void)p; (void)off; (void)buf; (void)len;
return -1;
}
int64_t vfs_prefetch_view(struct vfs_prefetch *p, uint64_t off, const uint8_t **ptr)
{
(void)p; (void)off; (void)ptr;
return -1;
}
void vfs_prefetch_seek(struct vfs_prefetch *p, uint64_t off)
{
(void)p; (void)off;
}
void vfs_prefetch_free(struct vfs_prefetch *p)
{
(void)p;
}
#endif
+2
View File
@@ -56,6 +56,8 @@ COREFLAGS += -DNDEBUG -DNO_VERIFY
ifeq ($(NO_THREADS),1)
COREFLAGS += -DTARGET_NO_THREADS
else
COREFLAGS += -DHAVE_THREADS
endif
ifeq ($(HAVE_GENERIC_JIT),1)
+186
View File
@@ -0,0 +1,186 @@
/* CD audio and data read from a CHD at once, as a game plays music while
* it streams from the disc.
*
* chd_streams_test <image.chd> <image.bin>
*
* The image is the two-track CD tools/chd/make_fixture.py writes (data,
* then audio). The audio is read a sector at a time, steadily, and the
* data in bursts announced with Prefetch() as the GD-ROM's read command
* does, between pauses that stand in for emulation. The pauses are
* scaled to how fast this build decodes, so the reads are paced like the
* drive's, well below the decoder, under a sanitizer as without. Every
* sector must match the BIN it came from, and the hunks must be decoded
* by the read-ahead threads: the reading thread may decode only where a
* stream starts. chd_image_read_hunk is wrapped (run.sh links with
* --wrap) to count the decodes done on the reading thread.
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <rthreads/rthreads.h>
#include <retro_timers.h>
#include <features/features_cpu.h>
#include "types.h"
#include "imgread/common.h"
#include "imgread/chd_image.h"
Disc* chd_parse(const char* file);
u8 q_subchannel[96];
void GenericLog(LogTypes::LOG_LEVELS, LogTypes::LOG_TYPE, const char*, int,
const char* fmt, ...)
{
(void)fmt;
}
static uintptr_t reader_thread;
static long reader_decodes;
static long decodes;
extern "C" bool __real_chd_image_read_hunk(chd_image_t *img, uint32_t hunk, uint8_t *dst);
extern "C" bool __wrap_chd_image_read_hunk(chd_image_t *img, uint32_t hunk, uint8_t *dst)
{
if (sthread_get_current_thread_id() == reader_thread)
reader_decodes++;
__sync_fetch_and_add(&decodes, 1);
return __real_chd_image_read_hunk(img, hunk, dst);
}
#define BURSTS 12
#define BURST_SECTORS 96
#define AUDIO_STEPS 40 /* audio sectors between two bursts */
static retro_time_t pace_us; /* emulation between two sector reads */
static retro_time_t owed_us;
static void pace(retro_time_t us)
{
owed_us += us;
if (owed_us >= 1000)
{
retro_sleep((unsigned)(owed_us / 1000));
owed_us %= 1000;
}
}
/* Microseconds this build takes to decode a hunk, best of a few. */
static retro_time_t decode_us(const char *path)
{
chd_image_t *img = chd_image_open(path, NULL, 0);
retro_time_t best = 0;
u8 *buf;
u32 h;
if (!img)
return 1000;
buf = (u8 *)malloc(chd_image_hunk_bytes(img));
for (h = 0; h < 16 && h < chd_image_hunk_count(img); h++)
{
retro_time_t t0 = cpu_features_get_time_usec(), t;
__real_chd_image_read_hunk(img, h, buf);
t = cpu_features_get_time_usec() - t0;
if (!best || t < best)
best = t;
}
free(buf);
chd_image_close(img);
return best ? best : 1;
}
int main(int argc, char **argv)
{
FILE *f;
u8 *bin;
long bin_len;
Disc *disc;
u8 sector[2352];
SectorFormat fmt;
SubcodeFormat sub;
u32 data_fad, audio_fad, data_lba, audio_lba, data_end, audio_end;
int bad = 0, b, i;
long expected;
if (argc < 3)
{
fprintf(stderr, "usage: %s image.chd image.bin\n", argv[0]);
return 2;
}
if (!(f = fopen(argv[2], "rb")))
return 2;
fseek(f, 0, SEEK_END);
bin_len = ftell(f);
fseek(f, 0, SEEK_SET);
bin = (u8 *)malloc(bin_len);
if (fread(bin, 1, bin_len, f) != (size_t)bin_len)
return 2;
fclose(f);
/* a hunk holds 8 sectors: read one every half hunk's decode time,
* at least 50 us, a quarter of the speed the decoder keeps up with */
pace_us = decode_us(argv[1]) / 2;
if (pace_us < 50)
pace_us = 50;
reader_thread = sthread_get_current_thread_id();
if (!(disc = chd_parse(argv[1])) || disc->tracks.size() != 2
|| disc->tracks[1].CTRL != 0)
{
printf("FAIL: open as a data track and an audio track\n");
return 1;
}
/* LBA in the BIN = FAD - 150 */
data_fad = disc->tracks[0].StartFAD;
data_end = disc->tracks[0].EndFAD;
audio_fad = disc->tracks[1].StartFAD;
audio_end = disc->tracks[1].EndFAD;
/* let the threads settle after the open */
retro_sleep(50);
reader_decodes = 0;
for (b = 0; b < BURSTS; b++)
{
for (i = 0; i < AUDIO_STEPS && audio_fad <= audio_end; i++, audio_fad++)
{
audio_lba = audio_fad - 150;
if (!disc->ReadSector(audio_fad, sector, &fmt, NULL, &sub)
|| memcmp(sector, bin + (size_t)audio_lba * 2352, 2352))
bad++;
pace(pace_us);
}
/* the drive seeks before the first sector comes */
disc->Prefetch(data_fad, BURST_SECTORS);
pace(pace_us * 16);
for (i = 0; i < BURST_SECTORS && data_fad <= data_end; i++, data_fad++)
{
data_lba = data_fad - 150;
if (!disc->ReadSector(data_fad, sector, &fmt, NULL, &sub)
|| memcmp(sector, bin + (size_t)data_lba * 2352, 2352))
bad++;
pace(pace_us);
}
}
/* the threads stop with the disc, and their count with them */
delete disc;
/* A decode here as each stream starts, and room for a thread now and
* then late; reads that throw out each other's hunks decode here at
* least twice a burst. */
expected = 2 + BURSTS / 2;
printf(" %ld hunks decoded, %ld of them on the reading thread "
"(at most %ld wanted)\n", decodes, reader_decodes, expected);
if (bad)
printf("FAIL: %d sectors do not match the BIN\n", bad);
if (reader_decodes > expected)
printf("FAIL: audio and data reads decode on the reading thread\n");
free(bin);
if (bad || reader_decodes > expected)
return 1;
printf("chd_streams_test: PASS\n");
return 0;
}
+47
View File
@@ -13,6 +13,11 @@
# The preprocessor flags are read back out of the core's own build, so the
# test decodes with exactly the codec set the core is built with.
#
# chd_streams_test then reads CD audio and data from the CD at once, as a
# game playing music while it streams does, through the core's chd.cpp,
# under ASan + UBSan and under TSan: the read-ahead threads must decode the
# hunks, and neither stream throw out the other's.
#
# Needs chdman (MAME tools) and python3. Fails rather than skips when either
# is missing: a CHD check that quietly does nothing is how a codec stops
# working without anyone noticing.
@@ -64,6 +69,39 @@ $CC -O1 -g $VDEFS -fsanitize=address,undefined \
-lm -lpthread
done
CXX=${CXX:-c++}
STREAM_SRC="core/imgread/chd_image.c core/deps/coreio/coreio.c
core/archive/archive.c $L/formats/chd/rchd.c
$L/encodings/encoding_huffman.c $L/encodings/encoding_rzstd.c
$L/encodings/encoding_deflate.c $L/encodings/encoding_crc32.c
$L/formats/7z/r7z_lzma.c $L/formats/7z/r7z_archive.c
$L/formats/7z/r7z_lzma2.c $L/formats/7z/r7z_lzma_stream.c
$L/formats/7z/r7z_bcj2.c $L/formats/7z/r7z_filters.c
$L/formats/zip/rzip_archive.c $L/formats/rar/rrar_archive.c
$L/formats/rar/rrar_ppmd7.c $L/rthreads/rthreads.c
$L/rthreads/retro_eventcount.c $L/formats/flac/rflac.c
$L/features/features_cpu.c $L/streams/file_stream.c
$L/vfs/vfs_implementation.c $L/vfs/vfs_prefetch.c $L/file/file_path.c
$L/file/file_path_io.c $L/file/retro_dirent.c $L/compat/compat_strl.c
$L/compat/fopen_utf8.c $L/compat/compat_strcasestr.c
$L/encodings/encoding_utf.c $L/string/stdstring.c $L/string/rstrtod.c
$L/time/rtime.c $L/memmap/memalign.c"
for san in address,undefined thread; do
v=$(echo $san | tr -d ,)
mkdir -p "$WORK/o_$v"
for s in $STREAM_SRC; do
$CC -O1 -g $DEFS -fsanitize=$san -fno-sanitize-recover=undefined \
-Icore -Icore/libretro -Icore/deps -I$L/include \
-c "$s" -o "$WORK/o_$v/$(echo "$s" | tr / _).o"
done
$CXX -std=c++11 -fpermissive -w -O1 -g $DEFS -fsanitize=$san \
-fno-sanitize-recover=undefined \
-Icore -Icore/libretro -Icore/deps -I$L/include \
-o "$WORK/chd_streams_test_$v" tools/chd/chd_streams_test.cpp \
core/imgread/chd.cpp "$WORK/o_$v"/*.o \
-Wl,--wrap=chd_image_read_hunk -lm -lpthread
done
python3 tools/chd/make_fixture.py "$WORK/base"
python3 tools/chd/make_fixture.py "$WORK/child" --variant 1
python3 tools/chd/make_fixture.py "$WORK/gd/disc" --gdi
@@ -117,8 +155,17 @@ for c in cdzs cdlz; do
done
check "$WORK/chain/child.chd" "$WORK/child.bin"
export TSAN_OPTIONS=halt_on_error=1
for c in cdzs cdfl; do
for v in addressundefined thread; do
"$WORK/chd_streams_test_$v" "$WORK/$c.chd" "$WORK/base.bin" | tail -1 \
| grep -q PASS || { echo "FAIL (streams, $v): $c"; fail=1; }
done
done
if [ $fail = 0 ]; then
echo "chd_image_test: all images PASS (mapped and bounce-buffer builds)"
echo "chd_streams_test: CD audio and data read at once PASS"
else
exit 1
fi
+1 -1
View File
@@ -34,7 +34,7 @@ sort -u -o "$WORK/have" "$WORK/have"
# Only the symbols the core's own headers name: libc, GL and the C++
# runtime come from the toolchain.
nm -u "$A" 2>/dev/null | awk '{print $2}' | sort -u \
| grep -E '^(core_|archive_|chd_image_|rchd_|r7z_|rzip_|rinflate_|rzstd_|rflac_|filestream_|retro_vfs_|path_|fill_pathname|string_|strlcpy|strlcat|encoding_|cpu_features_|retro_opendir|retro_readdir|retro_closedir|retro_dirent|sthread_|slock_|scond_|retro_eventcount|retro_spsc|memalign_|rtime_|utf8|utf16)' \
| grep -E '^(core_|archive_|chd_image_|rchd_|r7z_|rzip_|rinflate_|rzstd_|rflac_|filestream_|retro_vfs_|path_|fill_pathname|string_|strlcpy|strlcat|encoding_|cpu_features_|retro_opendir|retro_readdir|retro_closedir|retro_dirent|sthread_|slock_|scond_|retro_eventcount|retro_spsc|vfs_prefetch_|memalign_|rtime_|utf8|utf16)' \
> "$WORK/need" || true
MISSING=$(comm -23 "$WORK/need" "$WORK/have")
if [ -n "$MISSING" ]; then
-71
View File
@@ -19,7 +19,6 @@
#include "lockfree.h"
#include "libretro/emu_baton.h"
#include "libretro/input_latch.h"
#include "imgread/hunk_prefetch.h"
static int failures;
@@ -308,75 +307,6 @@ static void test_mpsc_list(void)
CHECK(handled == (long)MPSC_PRODUCERS * MPSC_REQUESTS);
}
/* ---- HunkPrefetch ---------------------------------------------------- */
#define HUNK_BYTES 4096
#define HUNK_COUNT 512
#define HUNK_READS 60000
static bool hunk_is_bad(u32 hunk) { return hunk % 97 == 96; }
static bool hunk_decode(void *, u32 hunk, u8 *dst)
{
if (hunk_is_bad(hunk))
return false;
for (u32 i = 0; i < HUNK_BYTES; i++)
dst[i] = (u8)(hunk * 31 + i);
return true;
}
static bool hunk_matches(u32 hunk, const u8 *p)
{
for (u32 i = 0; i < HUNK_BYTES; i++)
if (p[i] != (u8)(hunk * 31 + i))
return false;
return true;
}
static void test_hunk_prefetch(void)
{
HunkPrefetch pf;
unsigned seed = 12345;
u32 hunk = 0;
long hits = 0;
CHECK(pf.Start(hunk_decode, NULL, HUNK_BYTES, HUNK_COUNT));
CHECK(!pf.Start(hunk_decode, NULL, HUNK_BYTES, HUNK_COUNT));
for (int i = 0; i < HUNK_READS; i++)
{
unsigned r = rng_next(&seed);
/* Mostly sequential, as a disc is read, with a seek now and then. */
if (r % 50 == 0)
hunk = rng_next(&seed) % HUNK_COUNT;
else
hunk = (hunk + 1) % HUNK_COUNT;
if (pf.Fetch(hunk))
{
/* The worker never hands over a hunk it could not decode. */
CHECK(!hunk_is_bad(hunk));
CHECK(hunk_matches(hunk, pf.Buffer()));
hits++;
}
else if (hunk_decode(NULL, hunk, pf.Buffer()))
CHECK(hunk_matches(hunk, pf.Buffer()));
/* Give the worker a moment now and then, as emulation between
* two disc reads does. */
if (i % 64 == 0)
retro_sleep(1);
}
pf.Stop();
CHECK(!pf.Running());
CHECK(hits > 0);
/* Starts again after a stop. */
CHECK(pf.Start(hunk_decode, NULL, HUNK_BYTES, HUNK_COUNT));
pf.Stop();
printf(" hunk prefetch: %ld of %d reads found decoded\n", hits, HUNK_READS);
}
/* ---- EmuBaton -------------------------------------------------------- */
#define BATON_FRAMES 30000
@@ -643,7 +573,6 @@ int main(void)
{ "cSlotCache", test_slot_cache },
{ "cTripleBuffer", test_triple_buffer },
{ "cMpscList", test_mpsc_list },
{ "HunkPrefetch", test_hunk_prefetch },
{ "EmuBaton", test_emu_baton },
{ "InputLatch", test_input_latch },
{ "cMarkSet", test_mark_set },