diff --git a/bend2/effs/audio.c b/bend2/effs/audio.c index 7e9b9e217..abba58116 100644 --- a/bend2/effs/audio.c +++ b/bend2/effs/audio.c @@ -108,15 +108,23 @@ static void io_ring_free(IoRing* p) { // A thread feeds the default ALSA device 256 frames at a time (a // write blocks until the device has room, so the ring drains at the -// device's clock). +// device's clock). A chunk stays until the device has taken all of it; +// an error ALSA can't recover stops the thread, and the ring fills as +// with a stalled device. static void* io_ring_pump(void* ctx) { - IoRing* p = ctx; + IoRing* p = ctx; float out[256 * 2]; + u32 at = 256; while (atomic_load_explicit(&p->done, memory_order_relaxed) == 0) { - io_ring_pull(p, out, 256); - snd_pcm_sframes_t n = snd_pcm_writei(p->unit, out, 256); - if (n < 0) { - snd_pcm_recover(p->unit, (int)n, 1); + if (at == 256) { + io_ring_pull(p, out, 256); + at = 0; + } + snd_pcm_sframes_t n = snd_pcm_writei(p->unit, out + at * 2, 256 - at); + if (n > 0) { + at += (u32)n; + } else if (n == 0 || snd_pcm_recover(p->unit, (int)n, 1) < 0) { + break; } } return NULL; diff --git a/tests/io/audio_linux_progress.bend b/tests/io/audio_linux_progress.bend new file mode 100644 index 000000000..6663d557b --- /dev/null +++ b/tests/io/audio_linux_progress.bend @@ -0,0 +1,52 @@ +# The Linux audio pump against injected ALSA write results (ALSA's null +# output underneath, so no device). Other systems run no cases. +import Base + +def AudioTest.count() -> IO(U32): + import "./audio_linux_progress.c" + import "./audio_linux_progress.js" + +def AudioTest.start(i: U32) -> IO(Unit): + import "./audio_linux_progress.c" + import "./audio_linux_progress.js" + +def AudioTest.wait() -> IO(Unit): + import "./audio_linux_progress.c" + import "./audio_linux_progress.js" + +def AudioTest.closed() -> IO(Unit): + import "./audio_linux_progress.c" + import "./audio_linux_progress.js" + +def samples(n: Nat, +i: U32) -> List<&2, F32>: + match n: + case 0n: + [] + case 1n+p: + +x = U32.to_f32(i) + x <> ((0.0 - x : F32) <> samples(p, (i + 1 : U32))) + +def close_pair(pair: Audio & U32) -> IO(Unit): + (a, queued) = pair + Audio.close(a) + +def cases(n: Nat, +i: U32) -> IO(Unit): + match n: + case 0n: + IO.print("audio") + case 1n+p: + do IO: + Unit <- AudioTest.start(i) + a : Audio <- IO.try(Audio, Audio.open(48000)) + pair : Audio & U32 <- Audio.write(a, samples(512n, 1)) + Unit <- AudioTest.wait() + Unit <- close_pair(pair) + Unit <- AudioTest.closed() + cases(p, (i + 1 : U32)) + +def main() -> IO(Unit): + do IO: + n : U32 <- AudioTest.count() + cases(U32.to_nat(n), 0) + +#|audio diff --git a/tests/io/audio_linux_progress.c b/tests/io/audio_linux_progress.c new file mode 100644 index 000000000..fde8789f2 --- /dev/null +++ b/tests/io/audio_linux_progress.c @@ -0,0 +1,137 @@ +#ifdef __linux__ +#include +#include + +// Writes of real samples (frame k is k, -k) get the case's result; the +// rest reach ALSA's null output. Cases 0-254: the first write takes +// 1-255 frames. Then: EINTR once, EPIPE once, every write capped at 17, +// 64 frames then EINTR, 64 then EPIPE (all deliver 512 frames in +// order), ENODEV always and 0 always (the pump stops after one write). +enum { SHORT = 255, EINTR1, EPIPE1, CAP17, PART_EINTR, PART_EPIPE, DEAD, ZERO, CASES }; + +static snd_pcm_t* audio_test_unit; +static void* audio_test_lib; +static pthread_key_t audio_test_key; +static u32 audio_test_case; +static _Atomic(u32) audio_test_frames, audio_test_bad, audio_test_writes; +static _Atomic(u32) audio_test_recovers, audio_test_stopped; + +static void audio_test_exit(void* p) { + atomic_store(&audio_test_stopped, 1); +} + +static void* audio_test_sym(const char* name) { + void* p = snd_dlsym(audio_test_lib, name, NULL); + if (p == NULL) err_fail("audio test: missing ALSA symbol"); + return p; +} + +int snd_pcm_open(snd_pcm_t** p, const char* name, snd_pcm_stream_t s, int mode) { + int (*open)(snd_pcm_t**, const char*, snd_pcm_stream_t, int) = audio_test_sym("snd_pcm_open"); + int code = open(p, "null", s, mode); + if (code == 0) audio_test_unit = *p; + return code; +} + +typedef snd_pcm_sframes_t (*Write)(snd_pcm_t*, const void*, snd_pcm_uframes_t); + +snd_pcm_sframes_t snd_pcm_writei(snd_pcm_t* p, const void* data, snd_pcm_uframes_t frames) { + Write write = audio_test_sym("snd_pcm_writei"); + const float* pcm = data; + pthread_setspecific(audio_test_key, (void*)1); + bool real = false; + for (u64 i = 0; i < frames; ++i) real |= pcm[i * 2] != 0; + if (p == audio_test_unit && real) { + u32 c = audio_test_case; + u32 call = atomic_fetch_add(&audio_test_writes, 1); + if (call > 100000) err_fail("audio test: unbounded writes"); + if (c < SHORT && call == 0) frames = c + 1; + if (c == CAP17 && frames > 17) frames = 17; + if ((c == PART_EINTR || c == PART_EPIPE) && call == 0) frames = 64; + if ((c == EINTR1 && call == 0) || (c == PART_EINTR && call == 1)) return -EINTR; + if ((c == EPIPE1 && call == 0) || (c == PART_EPIPE && call == 1)) return -EPIPE; + if (c == DEAD) return -ENODEV; + if (c == ZERO) return 0; + } + snd_pcm_sframes_t n = write(p, data, frames); + if (n > 0 && p == audio_test_unit) { + u32 k = atomic_load(&audio_test_frames); + for (snd_pcm_sframes_t i = 0; i < n; ++i) { + if (pcm[i * 2] == 0) continue; + ++k; + if (pcm[i * 2] != (float)k || pcm[i * 2 + 1] != -(float)k) atomic_store(&audio_test_bad, 1); + } + atomic_store(&audio_test_frames, k); + } + return n; +} + +int snd_pcm_recover(snd_pcm_t* p, int code, int silent) { + int (*recover)(snd_pcm_t*, int, int) = audio_test_sym("snd_pcm_recover"); + if (p == audio_test_unit) atomic_fetch_add(&audio_test_recovers, 1); + return recover(p, code, silent); +} +#endif + +Term audio_test_count_run(Env e, Term* f, IoWork* w) { +#ifdef __linux__ + return CASES; +#endif + return 0; +} + +Term audio_test_start_run(Env e, Term* f, IoWork* w) { +#ifdef __linux__ + audio_test_case = (u32)f[0]; + atomic_store(&audio_test_frames, 0); + atomic_store(&audio_test_bad, 0); + atomic_store(&audio_test_writes, 0); + atomic_store(&audio_test_recovers, 0); + atomic_store(&audio_test_stopped, 0); +#endif + return term_pak(CID(Unit), 0); +} + +// Waits for all 512 frames, or for the pump to stop on a dead device. +Term audio_test_wait_run(Env e, Term* f, IoWork* w) { +#ifdef __linux__ + u32 c = audio_test_case; + bool dead = c == DEAD || c == ZERO; + u64 t0 = io_tick(); + while (dead ? !atomic_load(&audio_test_stopped) : atomic_load(&audio_test_frames) < 512) { + if (atomic_load(&audio_test_bad) || io_tick() - t0 > 1000000000ull) { + fprintf(stderr, "audio test case %u: frames=%u writes=%u recovers=%u\n", c, + atomic_load(&audio_test_frames), atomic_load(&audio_test_writes), + atomic_load(&audio_test_recovers)); + err_fail("audio test: pump lost or reordered frames"); + } + struct timespec t = { 0, 100000 }; + nanosleep(&t, NULL); + } + if (dead && (atomic_load(&audio_test_writes) != 1 + || atomic_load(&audio_test_recovers) != (c == DEAD))) + err_fail("audio test: dead device retried"); +#endif + return term_pak(CID(Unit), 0); +} + +Term audio_test_closed_run(Env e, Term* f, IoWork* w) { +#ifdef __linux__ + if (!atomic_load(&audio_test_stopped)) err_fail("audio test: close did not join the pump"); + if (audio_test_case != DEAD && audio_test_case != ZERO && atomic_load(&audio_test_frames) != 512) + err_fail("audio test: final frames"); +#endif + return term_pak(CID(Unit), 0); +} + +static void __attribute__((constructor)) audio_test_use(void) { +#ifdef __linux__ + audio_test_lib = snd_dlopen(NULL, RTLD_LAZY, NULL, 0); + if (audio_test_lib == NULL || pthread_key_create(&audio_test_key, audio_test_exit)) + err_fail("audio test: setup"); +#endif + io_eff(CID(AudioTest.count), audio_test_count_run); + io_eff(CID(AudioTest.start), audio_test_start_run); + io_eff(CID(AudioTest.wait), audio_test_wait_run); + io_eff(CID(AudioTest.closed), audio_test_closed_run); +} diff --git a/tests/io/audio_linux_progress.js b/tests/io/audio_linux_progress.js new file mode 100644 index 000000000..9ca3bd0f2 --- /dev/null +++ b/tests/io/audio_linux_progress.js @@ -0,0 +1,5 @@ +// The cases run only in the Linux C lane. +io_eff(CID(AudioTest.count), () => 0); +io_eff(CID(AudioTest.start), () => ({ $: CID(Unit) })); +io_eff(CID(AudioTest.wait), () => ({ $: CID(Unit) })); +io_eff(CID(AudioTest.closed), () => ({ $: CID(Unit) }));