mirror/psxrecomp
1
0
Fork 0
mirror of https://github.com/RetroPortingToolKit/psxrecomp synced 2026-10-26 02:11:36 -04:00
Code Issues Packages Projects Releases Wiki Activity
Files
2a5e53ffbfcfc9ea0788e8175b8b911c5c323fdf
psxrecomp/runtime/src/render_thread.c
T

470 lines
15 KiB
C
Raw Blame History

/* render_thread.c — render thread core. See render_thread.h and
* docs/RENDER_THREAD.md.
*
* One producer (the emulation thread), one consumer (the render thread).
* Positions are monotonic 64-bit byte counters; the ring index is pos & mask.
* The consumer frees a record only after exec() returns, so a payload stays
* valid for the whole replay. Locks only guard sleeping: the sleeper raises
* its *_sleeping flag under the lock BEFORE re-checking its condition, and the
* waker publishes its position (seq_cst) BEFORE reading that flag, so a wake
* can never fall between a check and a wait. */
#include "render_thread.h"
#include <stdatomic.h>
#include <stdlib.h>
#include <string.h>
#ifdef _WIN32
# define WIN32_LEAN_AND_MEAN
# include <windows.h>
typedef SRWLOCK rt_mutex;
typedef CONDITION_VARIABLE rt_cond;
static void mtx_init(rt_mutex *m) { InitializeSRWLock(m); }
static void mtx_lock(rt_mutex *m) { AcquireSRWLockExclusive(m); }
static void mtx_unlock(rt_mutex *m) { ReleaseSRWLockExclusive(m); }
static void cnd_init(rt_cond *c) { InitializeConditionVariable(c); }
static void cnd_wait(rt_cond *c, rt_mutex *m) { SleepConditionVariableSRW(c, m, INFINITE, 0); }
static void cnd_bcast(rt_cond *c) { WakeAllConditionVariable(c); }
static uint64_t now_ns(void) {
LARGE_INTEGER f, t;
QueryPerformanceFrequency(&f);
QueryPerformanceCounter(&t);
return (uint64_t)((double)t.QuadPart * 1e9 / (double)f.QuadPart);
}
static void cpu_relax(void) { YieldProcessor(); }
#else
# include <pthread.h>
# include <sched.h>
# if defined(__APPLE__)
# include <pthread/qos.h>
# endif
# include <time.h>
typedef pthread_mutex_t rt_mutex;
typedef pthread_cond_t rt_cond;
static void mtx_init(rt_mutex *m) { pthread_mutex_init(m, NULL); }
static void mtx_lock(rt_mutex *m) { pthread_mutex_lock(m); }
static void mtx_unlock(rt_mutex *m) { pthread_mutex_unlock(m); }
static void cnd_init(rt_cond *c) { pthread_cond_init(c, NULL); }
static void cnd_wait(rt_cond *c, rt_mutex *m) { pthread_cond_wait(c, m); }
static void cnd_bcast(rt_cond *c) { pthread_cond_broadcast(c); }
static uint64_t now_ns(void) {
struct timespec ts;
clock_gettime(CLOCK_MONOTONIC, &ts);
return (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec;
}
static void cpu_relax(void) {
# if defined(__aarch64__)
__asm__ __volatile__("yield");
# elif defined(__x86_64__) || defined(__i386__)
__asm__ __volatile__("pause");
# endif
}
#endif
#define OP_PAD 0u /* filler to the ring end; consumer skips it */
#define F_FRAME 0x8000u /* op 0 with this flag: frame marker */
/* Producer-side wake batching: a sleeping render thread is woken once this
* much unread work is queued (frame ends, drains and acquires always wake). */
#define WAKE_BYTES (128u * 1024u)
#define WAKE_RECORDS 256u
#define SPIN_LOOPS 4096 /* consumer polls this long before sleeping */
#define ACQ_RING 64
static struct {
/* configuration */
uint8_t *ring;
uint64_t cap, mask;
int max_frames;
RtExecFn exec;
RtCtxFn ctx;
void *user;
/* shared positions / counters */
_Atomic uint64_t wpos, rpos;
_Atomic uint64_t frames_produced, frames_consumed;
_Atomic int render_sleeping, emu_sleeping;
_Atomic int running;
/* hand-off + lifecycle, guarded by mtx */
rt_mutex mtx;
rt_cond cv_work, cv_emu;
int release_req, stop;
_Atomic int released; /* read unlocked by emu_wait's first check */
/* producer-only */
uint64_t pend_start, pend_size, pend_payload_off;
uint16_t pend_op, pend_flags;
uint32_t pend_payload;
int pending;
uint64_t unwoken_bytes, unwoken_records;
uint32_t seq;
int held;
/* stats (producer unless noted) */
uint64_t records, bytes, acquires, releases, oversize;
uint64_t bp_waits, full_waits, bp_ns, full_ns, acq_ns, high_water;
_Atomic uint64_t busy_ns, idle_ns; /* consumer */
RtAcquireEvent acq_ring[ACQ_RING];
uint64_t acq_n;
#ifdef _WIN32
HANDLE thread;
#else
pthread_t thread;
#endif
} R;
#if defined(_MSC_VER)
static __declspec(thread) int t_render;
#else
static __thread int t_render;
#endif
int rt_running(void) { return atomic_load(&R.running); }
int rt_on_render_thread(void) { return t_render; }
int rt_held(void) { return R.held; }
uint64_t rt_render_idle_ns(void) { return atomic_load(&R.idle_ns); }
static inline uint64_t align_up(uint64_t n) {
return (n + (RT_ALIGN - 1)) & ~(uint64_t)(RT_ALIGN - 1);
}
/* ---- consumer ------------------------------------------------------------ */
static void wake_emu(void) {
if (atomic_load(&R.emu_sleeping)) {
mtx_lock(&R.mtx);
cnd_bcast(&R.cv_emu);
mtx_unlock(&R.mtx);
}
}
/* Called with mtx held when the ring is empty: hand the context over while
* the emulation thread asks for it. Returns with mtx held. */
static void serve_release_locked(void) {
if (!R.release_req) return;
mtx_unlock(&R.mtx);
R.ctx(R.user, 0);
mtx_lock(&R.mtx);
atomic_store(&R.released, 1);
cnd_bcast(&R.cv_emu);
while (R.release_req && !R.stop) cnd_wait(&R.cv_work, &R.mtx);
atomic_store(&R.released, 0);
if (R.stop) return; /* the stopper owns the context now */
mtx_unlock(&R.mtx);
R.ctx(R.user, 1);
mtx_lock(&R.mtx);
}
#ifdef _WIN32
static DWORD WINAPI render_main(LPVOID arg)
#else
static void *render_main(void *arg)
#endif
{
(void)arg;
t_render = 1;
R.ctx(R.user, 1);
uint64_t r = atomic_load(&R.rpos);
int have_ctx = 1;
for (;;) {
uint64_t w = atomic_load_explicit(&R.wpos, memory_order_acquire);
if (r == w) {
int spun = 0;
while (spun < SPIN_LOOPS &&
atomic_load_explicit(&R.wpos, memory_order_acquire) == r) {
cpu_relax();
spun++;
}
if (atomic_load_explicit(&R.wpos, memory_order_acquire) != r) continue;
uint64_t t0 = now_ns();
mtx_lock(&R.mtx);
atomic_store(&R.render_sleeping, 1);
while (atomic_load(&R.wpos) == r && !R.stop && !R.release_req)
cnd_wait(&R.cv_work, &R.mtx);
atomic_store(&R.render_sleeping, 0);
if (R.release_req && atomic_load(&R.wpos) == r) {
serve_release_locked();
if (R.stop) have_ctx = 0;
}
int done = R.stop && atomic_load(&R.wpos) == r;
mtx_unlock(&R.mtx);
atomic_fetch_add(&R.idle_ns, now_ns() - t0);
if (done) break;
continue;
}
uint64_t t0 = now_ns();
const RtCmd *c = (const RtCmd *)(R.ring + (r & R.mask));
if (c->op == OP_PAD) {
if (c->flags & F_FRAME)
atomic_fetch_add(&R.frames_consumed, 1);
} else {
R.exec(R.user, c, (const uint8_t *)c + sizeof(RtCmd));
}
r += c->size;
atomic_store(&R.rpos, r);
atomic_fetch_add(&R.busy_ns, now_ns() - t0);
wake_emu();
}
if (have_ctx) R.ctx(R.user, 0);
t_render = 0;
#ifdef _WIN32
return 0;
#else
return NULL;
#endif
}
/* ---- producer ------------------------------------------------------------ */
static void wake_render(int force) {
if (!force && R.unwoken_bytes < WAKE_BYTES && R.unwoken_records < WAKE_RECORDS)
return;
R.unwoken_bytes = R.unwoken_records = 0;
if (atomic_load(&R.render_sleeping)) {
mtx_lock(&R.mtx);
cnd_bcast(&R.cv_work);
mtx_unlock(&R.mtx);
}
}
/* Block until pred() holds; the render thread wakes us after every record. */
typedef int (*Pred)(uint64_t arg);
static void emu_wait(Pred pred, uint64_t arg) {
if (pred(arg)) return;
wake_render(1);
mtx_lock(&R.mtx);
atomic_store(&R.emu_sleeping, 1);
while (!pred(arg)) cnd_wait(&R.cv_emu, &R.mtx);
atomic_store(&R.emu_sleeping, 0);
mtx_unlock(&R.mtx);
}
static int pred_space(uint64_t need_end) { /* need_end: wpos after the write */
return need_end - atomic_load(&R.rpos) <= R.cap;
}
static int pred_drained(uint64_t w) { return atomic_load(&R.rpos) == w; }
static int pred_frames(uint64_t limit) {
return atomic_load(&R.frames_produced) - atomic_load(&R.frames_consumed) <= limit;
}
static int pred_released(uint64_t unused) { (void)unused; return atomic_load(&R.released); }
void *rt_cmd_begin(uint16_t op, uint16_t flags, uint32_t payload_bytes) {
if (!rt_running() || R.pending || op == OP_PAD) return NULL;
uint64_t size = align_up(sizeof(RtCmd) + (uint64_t)payload_bytes);
if (size > R.cap / 2) { R.oversize++; return NULL; }
uint64_t w = atomic_load_explicit(&R.wpos, memory_order_relaxed);
uint64_t off = w & R.mask;
uint64_t pad = (off + size > R.cap) ? R.cap - off : 0;
if (!pred_space(w + pad + size)) {
uint64_t t0 = now_ns();
R.full_waits++;
emu_wait(pred_space, w + pad + size);
R.full_ns += now_ns() - t0;
}
if (pad) {
RtCmd *p = (RtCmd *)(R.ring + off);
p->size = (uint32_t)pad; p->op = OP_PAD; p->flags = 0; p->payload = 0; p->seq = 0;
w += pad;
off = 0;
}
RtCmd *c = (RtCmd *)(R.ring + off);
c->size = (uint32_t)size;
c->op = op;
c->flags = flags;
c->payload = payload_bytes;
c->seq = R.seq++;
R.pend_start = w;
R.pend_size = size;
R.pending = 1;
return (uint8_t *)c + sizeof(RtCmd);
}
void rt_cmd_commit(void) {
if (!R.pending) return;
R.pending = 0;
uint64_t end = R.pend_start + R.pend_size;
atomic_store(&R.wpos, end);
R.records++;
R.bytes += R.pend_size;
uint64_t depth = end - atomic_load(&R.rpos);
if (depth > R.high_water) R.high_water = depth;
R.unwoken_bytes += R.pend_size;
R.unwoken_records++;
wake_render(0);
}
static void commit_marker(uint16_t flags) {
uint64_t w = atomic_load_explicit(&R.wpos, memory_order_relaxed);
uint64_t off = w & R.mask;
uint64_t size = sizeof(RtCmd);
if (!pred_space(w + size)) emu_wait(pred_space, w + size);
RtCmd *c = (RtCmd *)(R.ring + off); /* 16 bytes always fit before the end */
c->size = (uint32_t)size; c->op = OP_PAD; c->flags = flags; c->payload = 0; c->seq = R.seq++;
atomic_store(&R.wpos, w + size);
}
void rt_frame_end(void) {
if (!rt_running()) return;
if (R.held) {
/* Nothing is queued while the context is held; count the frame as
* produced and consumed so the in-flight bound stays meaningful. */
atomic_fetch_add(&R.frames_produced, 1);
atomic_fetch_add(&R.frames_consumed, 1);
return;
}
atomic_fetch_add(&R.frames_produced, 1);
commit_marker(F_FRAME);
wake_render(1);
if (!pred_frames((uint64_t)R.max_frames)) {
uint64_t t0 = now_ns();
R.bp_waits++;
emu_wait(pred_frames, (uint64_t)R.max_frames);
R.bp_ns += now_ns() - t0;
}
}
int rt_frames_ahead(void) {
int64_t d = (int64_t)(atomic_load(&R.frames_produced) -
atomic_load(&R.frames_consumed)) - 1;
return d < 0 ? 0 : (int)d;
}
void rt_drain(void) {
if (!rt_running() || R.held) return;
uint64_t w = atomic_load(&R.wpos);
emu_wait(pred_drained, w);
}
void rt_acquire(const char *reason) {
if (!rt_running() || R.held) return;
uint64_t t0 = now_ns();
rt_drain();
mtx_lock(&R.mtx);
R.release_req = 1;
cnd_bcast(&R.cv_work);
mtx_unlock(&R.mtx);
emu_wait(pred_released, 0);
R.ctx(R.user, 1);
R.held = 1;
uint64_t dt = now_ns() - t0;
R.acquires++;
R.acq_ns += dt;
RtAcquireEvent *e = &R.acq_ring[R.acq_n++ % ACQ_RING];
e->frame = atomic_load(&R.frames_produced);
e->wait_ns = dt;
e->reason = reason ? reason : "?";
}
void rt_release(void) {
if (!rt_running() || !R.held) return;
R.ctx(R.user, 0);
mtx_lock(&R.mtx);
R.release_req = 0;
cnd_bcast(&R.cv_work);
mtx_unlock(&R.mtx);
R.held = 0;
R.releases++;
}
/* ---- lifecycle ----------------------------------------------------------- */
int rt_start(const RtConfig *cfg) {
if (rt_running() || !cfg || !cfg->exec || !cfg->ctx) return 0;
size_t cap = cfg->ring_bytes;
if (cap < (1u << 20) || (cap & (cap - 1))) return 0;
uint8_t *ring = (uint8_t *)malloc(cap);
if (!ring) return 0;
memset(&R, 0, sizeof(R));
R.ring = ring;
R.cap = cap;
R.mask = cap - 1;
R.max_frames = cfg->max_frames < 1 ? 1 : cfg->max_frames;
R.exec = cfg->exec;
R.ctx = cfg->ctx;
R.user = cfg->user;
mtx_init(&R.mtx);
cnd_init(&R.cv_work);
cnd_init(&R.cv_emu);
/* The context leaves this thread before the render thread takes it. */
R.ctx(R.user, 0);
atomic_store(&R.running, 1);
#ifdef _WIN32
R.thread = CreateThread(NULL, 0, render_main, NULL, 0, NULL);
int ok = R.thread != NULL;
#else
pthread_attr_t attr;
pthread_attr_init(&attr);
# if defined(__APPLE__)
/* A plain pthread gets the default QoS and, on a busy host, efficiency
* cores. The render thread is on the frame's critical path, like the
* emulation (main) thread, so it asks for the same class. */
pthread_attr_set_qos_class_np(&attr, QOS_CLASS_USER_INTERACTIVE, 0);
# endif
int ok = pthread_create(&R.thread, &attr, render_main, NULL) == 0;
pthread_attr_destroy(&attr);
#endif
if (!ok) {
atomic_store(&R.running, 0);
R.ctx(R.user, 1);
free(R.ring);
R.ring = NULL;
return 0;
}
return 1;
}
void rt_stop(void) {
if (!rt_running()) return;
int was_held = R.held;
if (!was_held) rt_drain();
mtx_lock(&R.mtx);
R.stop = 1;
R.release_req = 0;
cnd_bcast(&R.cv_work);
mtx_unlock(&R.mtx);
#ifdef _WIN32
WaitForSingleObject(R.thread, INFINITE);
CloseHandle(R.thread);
#else
pthread_join(R.thread, NULL);
#endif
atomic_store(&R.running, 0);
R.held = 0;
if (!was_held) R.ctx(R.user, 1);
free(R.ring);
R.ring = NULL;
}
void rt_get_stats(RtStats *o) {
memset(o, 0, sizeof(*o));
o->records = R.records;
o->bytes = R.bytes;
o->frames_produced = atomic_load(&R.frames_produced);
o->frames_consumed = atomic_load(&R.frames_consumed);
o->acquires = R.acquires;
o->releases = R.releases;
o->oversize = R.oversize;
o->backpressure_waits = R.bp_waits;
o->ring_full_waits = R.full_waits;
o->backpressure_ns = R.bp_ns;
o->ring_full_ns = R.full_ns;
o->acquire_ns = R.acq_ns;
o->render_busy_ns = atomic_load(&R.busy_ns);
o->render_idle_ns = atomic_load(&R.idle_ns);
o->ring_high_water = R.high_water;
o->max_frames = R.max_frames;
o->running = rt_running();
o->held = R.held;
}
int rt_acquire_events(RtAcquireEvent *out, int cap) {
int n = 0;
for (uint64_t i = 0; i < R.acq_n && i < ACQ_RING && n < cap; i++)
out[n++] = R.acq_ring[(R.acq_n - 1 - i) % ACQ_RING];
return n;
}
Reference in New Issue View Git Blame Copy Permalink