From b6d27cf91271878cd19ef00c53630ed9590ea96f Mon Sep 17 00:00:00 2001 From: Sebastian Griego <177274079+sebastian-griego@users.noreply.github.com> Date: Sun, 11 Oct 2026 15:24:11 -0400 Subject: [PATCH 1/3] fix(audio): retain Linux partial writes and stop failed pumps --- bend2/effs/audio.c | 66 ++++++- guide/EFFECTS.md | 6 + tests/io/audio_linux_progress.bend | 95 ++++++++++ tests/io/audio_linux_progress.c | 268 +++++++++++++++++++++++++++++ tests/io/audio_linux_progress.js | 5 + 5 files changed, 431 insertions(+), 9 deletions(-) create mode 100644 tests/io/audio_linux_progress.bend create mode 100644 tests/io/audio_linux_progress.c create mode 100644 tests/io/audio_linux_progress.js diff --git a/bend2/effs/audio.c b/bend2/effs/audio.c index 7e9b9e217..c12f11f6a 100644 --- a/bend2/effs/audio.c +++ b/bend2/effs/audio.c @@ -22,10 +22,20 @@ typedef struct { #elif defined(__linux__) snd_pcm_t* unit; pthread_t pump; - _Atomic(u32) done; + _Atomic(u32) done, failed; #endif } IoRing; +#if defined(__linux__) && !defined(__OBJC__) +static void io_ring_error(u32 code) { + if (code != 0) { + char text[160]; + snprintf(text, sizeof text, "Audio: %s", snd_strerror(-(int)code)); + err_fail(text); + } +} +#endif + // n frames queued after the write; a write past the ring's room is // dropped and the queue answered as it is. static u64 io_ring_write(IoRing* p, const float* pcm, u32 n) { @@ -106,18 +116,47 @@ static void io_ring_free(IoRing* p) { #elif defined(__linux__) -// 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). +// Keep each chunk until ALSA accepts all frames. Nonblocking writes +// and bounded waits let close join a suspended pump. Stop after one +// second without progress; write or close reports the stored error. static void* io_ring_pump(void* ctx) { IoRing* p = ctx; float out[256 * 2]; + u32 at = 256; + u64 stalled = 0; 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; + stalled = 0; + continue; } + int code = n == 0 ? -EIO : (int)n; + if (code == -EAGAIN) { + code = snd_pcm_wait(p->unit, 20); + if (code >= 0) code = 0; + } + if (code == -ESTRPIPE) { + code = snd_pcm_resume(p->unit); + if (code == -EAGAIN) code = 0; + else if (code < 0) code = snd_pcm_prepare(p->unit); + } else if (code < 0) { + code = snd_pcm_recover(p->unit, code, 1); + } + if (atomic_load_explicit(&p->done, memory_order_relaxed) != 0) break; + u64 now = io_tick(); + if (stalled == 0) stalled = now; + if (code == 0 && now - stalled >= 1000000000ull) code = -ETIMEDOUT; + if (code < 0) { + atomic_store_explicit(&p->failed, (u32)-code, memory_order_release); + break; + } + struct timespec pause = { 0, 1000000 }; + nanosleep(&pause, NULL); } return NULL; } @@ -128,7 +167,8 @@ static void io_ring_hush(const char* file, int line, const char* fn, int err, static u32 io_ring_start(IoRing* p, u32 rate) { snd_lib_error_set_handler(io_ring_hush); - if (snd_pcm_open(&p->unit, "default", SND_PCM_STREAM_PLAYBACK, 0) < 0) { + if (snd_pcm_open(&p->unit, "default", SND_PCM_STREAM_PLAYBACK, + SND_PCM_NONBLOCK) < 0) { return ENODEV; } if (snd_pcm_set_params(p->unit, SND_PCM_FORMAT_FLOAT_LE, @@ -140,14 +180,17 @@ static u32 io_ring_start(IoRing* p, u32 rate) { } static void io_ring_free(IoRing* p) { + u32 code = 0; if (p->unit != NULL) { atomic_store_explicit(&p->done, 1, memory_order_relaxed); if (p->pump != 0) { pthread_join(p->pump, NULL); } + code = atomic_load_explicit(&p->failed, memory_order_acquire); snd_pcm_close(p->unit); } free(p); + io_ring_error(code); } #else @@ -198,6 +241,11 @@ Term audio_write_run(Env e, Term* f, IoWork* w) { n += 1; s = fb[1]; } +#if defined(__linux__) && !defined(__OBJC__) + if (atomic_load_explicit(&p->failed, memory_order_acquire) != 0) { + io_ring_free(p); + } +#endif u64 q = io_ring_write(p, pcm, n > IO_RING * 2 ? 0 : n / 2); return io_tup(e, f[0], q); } diff --git a/guide/EFFECTS.md b/guide/EFFECTS.md index 6d1c39ab6..cf98c1183 100644 --- a/guide/EFFECTS.md +++ b/guide/EFFECTS.md @@ -19,6 +19,12 @@ Each file registers the effect under its def with `io_eff(CID(Clock.now), serves `bend x.bend -o x.js` and `bend x.bend`. Every effect in Base is built this way: `bend2/effs/*.c` and `*.js` are the reference. +On Linux, the audio pump retains partial device writes and retries recoverable +errors. A zero write, failed recovery, or one second without progress stops +the pump. The next `Audio.write` or `Audio.close` prints the stored error and +exits with status 1. Both calls join the pump and release the device before they +report that error. A full ring still drops the new producer chunk. + ## The C side The compiler splices your `.c` into the program's C source after the diff --git a/tests/io/audio_linux_progress.bend b/tests/io/audio_linux_progress.bend new file mode 100644 index 000000000..b5584ae54 --- /dev/null +++ b/tests/io/audio_linux_progress.bend @@ -0,0 +1,95 @@ +# Test the Linux audio pump through the native Audio effects. +# The C test uses ALSA's null output and injects device return values. +import Base + +def AudioTest.mode() -> IO(U32): + import "./audio_linux_progress.c" + import "./audio_linux_progress.js" + +def AudioTest.prepare(mode: U32, width: U32) -> IO(Unit): + import "./audio_linux_progress.c" + import "./audio_linux_progress.js" + +def AudioTest.wait(mode: U32) -> 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 failed_call(close: Bool, a: Audio) -> IO(Unit): + match close: + case True{}: + Audio.close(a) + case False{}: + do IO: + pair : Audio & U32 <- Audio.write(a, []) + close_pair(pair) + +def outcome(good: Bool, mode: U32, a: Audio) -> IO(Unit): + match good: + case True{}: + do IO: + Unit <- Audio.close(a) + AudioTest.closed() + case False{}: + do IO: + Unit <- IO.print("before") + Unit <- failed_call(U32.is_eq(mode, 9), a) + IO.print("after") + +def finish(+mode: U32, pair: Audio & U32) -> IO(Unit): + (a, queued) = pair + do IO: + Unit <- AudioTest.wait(mode) + outcome(Bool.or(U32.is_lt(mode, 8), U32.is_gt(mode, 14)), mode, a) + +def consume(+mode: U32, width: U32) -> IO(Unit): + do IO: + Unit <- AudioTest.prepare(mode, width) + a : Audio <- IO.try(Audio, Audio.open(48000)) + pair : Audio & U32 <- Audio.write(a, samples(512n, 1)) + finish(mode, pair) + +def shorts(n: Nat) -> IO(Unit): + match n: + case 0n: + IO.pure(Unit, Unit{}) + case 1n+ +p: + do IO: + Unit <- consume(1, U32.from_nat(1n+p)) + shorts(p) + +def child(matrix: Bool, mode: U32) -> IO(Unit): + match matrix: + case True{}: + shorts(255n) + case False{}: + consume(mode, 256) + +def selected(parent: Bool, +mode: U32) -> IO(Unit): + match parent: + case True{}: + IO.print("audio") + case False{}: + child(U32.is_eq(mode, 1), mode) + +def main() -> IO(Unit): + do IO: + +mode : U32 <- AudioTest.mode() + selected(U32.is_eq(mode, 0), mode) + +#|audio diff --git a/tests/io/audio_linux_progress.c b/tests/io/audio_linux_progress.c new file mode 100644 index 000000000..ad7bc0709 --- /dev/null +++ b/tests/io/audio_linux_progress.c @@ -0,0 +1,268 @@ +#ifdef __linux__ +#include +#include +#include +#include + +// Only this program's null output is affected. Positive writes reach ALSA. +// A thread key reports pump exit without reading its private state. +static snd_pcm_t* audio_test_unit; +static void* audio_test_library; +static pthread_key_t audio_test_key; +static u32 audio_test_mode, audio_test_width; +static _Atomic(u32) audio_test_frames, audio_test_bad, audio_test_fault; +static _Atomic(u32) audio_test_writes, audio_test_recovers, audio_test_resumes; +static _Atomic(u32) audio_test_finished, audio_test_closes; + +static void audio_test_exit(void* p) { + atomic_store(&audio_test_finished, 1); +} + +static void* audio_test_symbol(const char* name) { + void* p = snd_dlsym(audio_test_library, 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 stream, int mode) { + typedef int (*Open)(snd_pcm_t**, const char*, snd_pcm_stream_t, int); + int code = ((Open)audio_test_symbol("snd_pcm_open"))(p, "null", stream, mode); + if (code == 0) audio_test_unit = *p; + return code; +} + +snd_pcm_sframes_t snd_pcm_writei(snd_pcm_t* p, const void* data, + snd_pcm_uframes_t frames) { + typedef snd_pcm_sframes_t (*Write)(snd_pcm_t*, const void*, snd_pcm_uframes_t); + const float* pcm = data; + pthread_setspecific(audio_test_key, (void*)1); + bool samples = false; + for (u64 i = 0; i < frames; ++i) samples |= pcm[i * 2] != 0; + if (p == audio_test_unit && (samples + || (audio_test_mode >= 7 && audio_test_mode <= 14 + && atomic_load(&audio_test_fault)))) { + u32 call = atomic_fetch_add(&audio_test_writes, 1); + if (call > 100000) err_fail("audio test: unbounded writes"); + bool first = atomic_exchange(&audio_test_fault, 1) == 0; + if (audio_test_mode == 1 && first) frames = audio_test_width; + if (audio_test_mode == 15 && frames > 17) frames = 17; + if ((audio_test_mode == 16 || audio_test_mode == 17) && call == 0) frames = 64; + if (audio_test_mode == 16 && call == 1) return -EINTR; + if (audio_test_mode == 17 && call == 1) return -EPIPE; + if ((audio_test_mode == 2 && first) || audio_test_mode == 12) return -EINTR; + if ((audio_test_mode == 3 && first) || audio_test_mode == 11) return -EPIPE; + if ((audio_test_mode == 4 && first) || audio_test_mode == 7 + || audio_test_mode == 14) return -ESTRPIPE; + if ((audio_test_mode == 5 && first) || audio_test_mode == 13) return -EAGAIN; + if (audio_test_mode == 8 || audio_test_mode == 9) return -ENODEV; + if (audio_test_mode == 10) return 0; + } + snd_pcm_sframes_t n = ((Write)audio_test_symbol("snd_pcm_writei"))(p, data, frames); + if (n > 0 && p == audio_test_unit) { + u32 count = atomic_load(&audio_test_frames); + for (snd_pcm_sframes_t i = 0; i < n; ++i) { + if (pcm[i * 2] == 0) { + if (pcm[i * 2 + 1] != 0) atomic_store(&audio_test_bad, 1); + } else { + ++count; + if (pcm[i * 2] != (float)count || pcm[i * 2 + 1] != -(float)count) + atomic_store(&audio_test_bad, 1); + } + } + atomic_store(&audio_test_frames, count); + } + if (!samples) { + struct timespec t = { 0, 1000000 }; + nanosleep(&t, NULL); + } + return n; +} + +int snd_pcm_recover(snd_pcm_t* p, int code, int silent) { + typedef int (*Recover)(snd_pcm_t*, int, int); + if (p == audio_test_unit) atomic_fetch_add(&audio_test_recovers, 1); + return ((Recover)audio_test_symbol("snd_pcm_recover"))(p, code, silent); +} + +int snd_pcm_resume(snd_pcm_t* p) { + if (p == audio_test_unit) { + u32 n = atomic_fetch_add(&audio_test_resumes, 1); + return audio_test_mode == 7 || n < 2 ? -EAGAIN : -ENOSYS; + } + typedef int (*Resume)(snd_pcm_t*); + return ((Resume)audio_test_symbol("snd_pcm_resume"))(p); +} + +int snd_pcm_prepare(snd_pcm_t* p) { + if (p == audio_test_unit && atomic_load(&audio_test_fault) + && (audio_test_mode == 11 || audio_test_mode == 14)) return -EIO; + typedef int (*Prepare)(snd_pcm_t*); + return ((Prepare)audio_test_symbol("snd_pcm_prepare"))(p); +} + +int snd_pcm_close(snd_pcm_t* p) { + typedef int (*Close)(snd_pcm_t*); + bool failed_close = p == audio_test_unit + && (audio_test_mode == 8 || audio_test_mode == 9); + if (p == audio_test_unit) { + if (audio_test_mode >= 8 && audio_test_mode <= 14 + && !atomic_load(&audio_test_finished)) err_fail("audio test: close before pump exit"); + atomic_fetch_add(&audio_test_closes, 1); + audio_test_unit = NULL; + } + int code = ((Close)audio_test_symbol("snd_pcm_close"))(p); + if (failed_close) { + if (code != 0 || !atomic_load(&audio_test_finished)) + err_fail("audio test: failed close did not release"); + puts("closed"); + } + return code; +} + +static void audio_test_read(int fd, char* out, size_t cap) { + size_t n = 0; + ssize_t got; + while (n + 1 < cap && (got = read(fd, out + n, cap - n - 1)) > 0) n += got; + out[n] = 0; + close(fd); +} + +static void audio_test_child(u32 mode) { + int out[2], err[2]; + if (pipe(out) || pipe(err)) err_fail("audio test: pipe"); + io_sync(); + extern char** environ; + size_t count = 0; + while (environ[count] != NULL) ++count; + char* vars[count + 2]; + size_t at = 0; + for (size_t i = 0; i < count; ++i) + if (strncmp(environ[i], "BEND_TEST_AUDIO_MODE=", + sizeof "BEND_TEST_AUDIO_MODE=" - 1)) vars[at++] = environ[i]; + char value[64]; + snprintf(value, sizeof value, "BEND_TEST_AUDIO_MODE=%u", mode); + vars[at++] = value; + vars[at] = NULL; + char* args[] = { "audio-test", "--threads", "1", "--gpu", "off", NULL }; + posix_spawn_file_actions_t actions; + posix_spawn_file_actions_init(&actions); + posix_spawn_file_actions_addclose(&actions, out[0]); + posix_spawn_file_actions_addclose(&actions, err[0]); + posix_spawn_file_actions_adddup2(&actions, out[1], 1); + posix_spawn_file_actions_adddup2(&actions, err[1], 2); + posix_spawn_file_actions_addclose(&actions, out[1]); + posix_spawn_file_actions_addclose(&actions, err[1]); + pid_t pid; + int spawn_code = posix_spawn(&pid, "/proc/self/exe", &actions, NULL, args, vars); + posix_spawn_file_actions_destroy(&actions); + if (spawn_code != 0) err_fail("audio test: spawn"); + close(out[1]); close(err[1]); + int status = 0; + u64 start = io_tick(); + while (waitpid(pid, &status, WNOHANG) == 0) { + if (io_tick() - start > 2500000000ull) { + kill(pid, SIGKILL); + waitpid(pid, &status, 0); + err_fail("audio test: child timeout"); + } + struct timespec t = { 0, 1000000 }; + nanosleep(&t, NULL); + } + char text[128], errors[512]; + audio_test_read(out[0], text, sizeof text); + audio_test_read(err[0], errors, sizeof errors); + bool failed = mode >= 8 && mode <= 14; + int code = mode == 8 || mode == 9 ? ENODEV + : mode == 12 || mode == 13 ? ETIMEDOUT : EIO; + char expected[160]; + snprintf(expected, sizeof expected, "bend: Audio: %s\n", snd_strerror(-code)); + if (!WIFEXITED(status) || WEXITSTATUS(status) != (failed ? 1 : 0) + || strcmp(text, mode == 8 || mode == 9 ? "before\nclosed\n" + : failed ? "before\n" : "") + || (failed ? strcmp(errors, expected) != 0 : errors[0] != 0)) { + fprintf(stderr, "audio test mode %u: status %d; stdout %s; stderr %s\n", + mode, status, text, errors); + err_fail("audio test: consumer failed"); + } +} +#endif + +Term audio_test_mode_run(Env e, Term* f, IoWork* w) { +#ifdef __linux__ + const char* mode = getenv("BEND_TEST_AUDIO_MODE"); + if (mode != NULL) return (u32)strtoul(mode, NULL, 10); + for (u32 i = 1; i <= 17; ++i) audio_test_child(i); +#endif + return 0; +} + +Term audio_test_prepare_run(Env e, Term* f, IoWork* w) { +#ifdef __linux__ + audio_test_mode = (u32)f[0]; + audio_test_width = (u32)f[1]; + const char* width = getenv("BEND_TEST_AUDIO_WIDTH"); + if (width != NULL && audio_test_mode == 1) + audio_test_width = (u32)strtoul(width, NULL, 10); + atomic_store(&audio_test_frames, 0); + atomic_store(&audio_test_bad, 0); + atomic_store(&audio_test_fault, 0); + atomic_store(&audio_test_writes, 0); + atomic_store(&audio_test_recovers, 0); + atomic_store(&audio_test_resumes, 0); + atomic_store(&audio_test_finished, 0); + atomic_store(&audio_test_closes, 0); +#endif + return term_pak(CID(Unit), 0); +} + +Term audio_test_wait_run(Env e, Term* f, IoWork* w) { +#ifdef __linux__ + u32 mode = (u32)f[0]; + u64 start = io_tick(); + while (mode == 7 ? atomic_load(&audio_test_resumes) < 1 + : mode >= 8 && mode <= 14 ? atomic_load(&audio_test_finished) == 0 + : atomic_load(&audio_test_frames) < 512) { + if (atomic_load(&audio_test_bad) + || io_tick() - start > (mode >= 8 && mode <= 14 ? 1500000000ull : 300000000ull)) { + fprintf(stderr, "audio test: frames=%u writes=%u recoveries=%u resumes=%u\n", + atomic_load(&audio_test_frames), atomic_load(&audio_test_writes), + atomic_load(&audio_test_recovers), atomic_load(&audio_test_resumes)); + err_fail("audio test: pump made no progress"); + } + struct timespec t = { 0, 100000 }; + nanosleep(&t, NULL); + } + if (atomic_load(&audio_test_bad)) err_fail("audio test: frame order or channels"); + if ((mode == 8 || mode == 9) && (atomic_load(&audio_test_writes) != 1 + || atomic_load(&audio_test_recovers) != 1)) err_fail("audio test: terminal retry"); + if ((mode == 12 || mode == 13) && atomic_load(&audio_test_writes) > 2000) + err_fail("audio test: retries have no wait"); + if ((mode == 16 || mode == 17) && atomic_load(&audio_test_recovers) != 1) + err_fail("audio test: partial write did not reach recovery"); +#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_finished) || atomic_load(&audio_test_closes) != 1) + err_fail("audio test: close did not join and release"); + if (audio_test_mode != 7 && (atomic_load(&audio_test_frames) != 512 + || atomic_load(&audio_test_bad))) 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_library = snd_dlopen(NULL, RTLD_LAZY, NULL, 0); + if (audio_test_library == NULL) err_fail("audio test: missing ALSA library"); + if (pthread_key_create(&audio_test_key, audio_test_exit)) + err_fail("audio test: thread key"); +#endif + io_eff(CID(AudioTest.mode), audio_test_mode_run); + io_eff(CID(AudioTest.prepare), audio_test_prepare_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..9ac3ec4c1 --- /dev/null +++ b/tests/io/audio_linux_progress.js @@ -0,0 +1,5 @@ +// The C test exercises the Linux native runtime. Other lanes skip it. +io_eff(CID(AudioTest.mode), () => 0); +io_eff(CID(AudioTest.prepare), () => ({ $: CID(Unit) })); +io_eff(CID(AudioTest.wait), () => ({ $: CID(Unit) })); +io_eff(CID(AudioTest.closed), () => ({ $: CID(Unit) })); From e7e1f8e319affa104afa792af29ebf040670c9cf Mon Sep 17 00:00:00 2001 From: Lorenzobattistela Date: Mon, 12 Oct 2026 11:38:17 -0300 Subject: [PATCH 2/3] fix(audio): keep a Linux chunk until ALSA takes it, stop on unrecoverable errors --- bend2/effs/audio.c | 58 +------ guide/EFFECTS.md | 6 - tests/io/audio_linux_progress.bend | 95 ---------- tests/io/audio_linux_progress.c | 268 ----------------------------- tests/io/audio_linux_progress.js | 5 - 5 files changed, 9 insertions(+), 423 deletions(-) delete mode 100644 tests/io/audio_linux_progress.bend delete mode 100644 tests/io/audio_linux_progress.c delete mode 100644 tests/io/audio_linux_progress.js diff --git a/bend2/effs/audio.c b/bend2/effs/audio.c index c12f11f6a..abba58116 100644 --- a/bend2/effs/audio.c +++ b/bend2/effs/audio.c @@ -22,20 +22,10 @@ typedef struct { #elif defined(__linux__) snd_pcm_t* unit; pthread_t pump; - _Atomic(u32) done, failed; + _Atomic(u32) done; #endif } IoRing; -#if defined(__linux__) && !defined(__OBJC__) -static void io_ring_error(u32 code) { - if (code != 0) { - char text[160]; - snprintf(text, sizeof text, "Audio: %s", snd_strerror(-(int)code)); - err_fail(text); - } -} -#endif - // n frames queued after the write; a write past the ring's room is // dropped and the queue answered as it is. static u64 io_ring_write(IoRing* p, const float* pcm, u32 n) { @@ -116,14 +106,15 @@ static void io_ring_free(IoRing* p) { #elif defined(__linux__) -// Keep each chunk until ALSA accepts all frames. Nonblocking writes -// and bounded waits let close join a suspended pump. Stop after one -// second without progress; write or close reports the stored error. +// 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). 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; - u64 stalled = 0; while (atomic_load_explicit(&p->done, memory_order_relaxed) == 0) { if (at == 256) { io_ring_pull(p, out, 256); @@ -132,31 +123,9 @@ static void* io_ring_pump(void* ctx) { snd_pcm_sframes_t n = snd_pcm_writei(p->unit, out + at * 2, 256 - at); if (n > 0) { at += (u32)n; - stalled = 0; - continue; - } - int code = n == 0 ? -EIO : (int)n; - if (code == -EAGAIN) { - code = snd_pcm_wait(p->unit, 20); - if (code >= 0) code = 0; - } - if (code == -ESTRPIPE) { - code = snd_pcm_resume(p->unit); - if (code == -EAGAIN) code = 0; - else if (code < 0) code = snd_pcm_prepare(p->unit); - } else if (code < 0) { - code = snd_pcm_recover(p->unit, code, 1); - } - if (atomic_load_explicit(&p->done, memory_order_relaxed) != 0) break; - u64 now = io_tick(); - if (stalled == 0) stalled = now; - if (code == 0 && now - stalled >= 1000000000ull) code = -ETIMEDOUT; - if (code < 0) { - atomic_store_explicit(&p->failed, (u32)-code, memory_order_release); + } else if (n == 0 || snd_pcm_recover(p->unit, (int)n, 1) < 0) { break; } - struct timespec pause = { 0, 1000000 }; - nanosleep(&pause, NULL); } return NULL; } @@ -167,8 +136,7 @@ static void io_ring_hush(const char* file, int line, const char* fn, int err, static u32 io_ring_start(IoRing* p, u32 rate) { snd_lib_error_set_handler(io_ring_hush); - if (snd_pcm_open(&p->unit, "default", SND_PCM_STREAM_PLAYBACK, - SND_PCM_NONBLOCK) < 0) { + if (snd_pcm_open(&p->unit, "default", SND_PCM_STREAM_PLAYBACK, 0) < 0) { return ENODEV; } if (snd_pcm_set_params(p->unit, SND_PCM_FORMAT_FLOAT_LE, @@ -180,17 +148,14 @@ static u32 io_ring_start(IoRing* p, u32 rate) { } static void io_ring_free(IoRing* p) { - u32 code = 0; if (p->unit != NULL) { atomic_store_explicit(&p->done, 1, memory_order_relaxed); if (p->pump != 0) { pthread_join(p->pump, NULL); } - code = atomic_load_explicit(&p->failed, memory_order_acquire); snd_pcm_close(p->unit); } free(p); - io_ring_error(code); } #else @@ -241,11 +206,6 @@ Term audio_write_run(Env e, Term* f, IoWork* w) { n += 1; s = fb[1]; } -#if defined(__linux__) && !defined(__OBJC__) - if (atomic_load_explicit(&p->failed, memory_order_acquire) != 0) { - io_ring_free(p); - } -#endif u64 q = io_ring_write(p, pcm, n > IO_RING * 2 ? 0 : n / 2); return io_tup(e, f[0], q); } diff --git a/guide/EFFECTS.md b/guide/EFFECTS.md index cf98c1183..6d1c39ab6 100644 --- a/guide/EFFECTS.md +++ b/guide/EFFECTS.md @@ -19,12 +19,6 @@ Each file registers the effect under its def with `io_eff(CID(Clock.now), serves `bend x.bend -o x.js` and `bend x.bend`. Every effect in Base is built this way: `bend2/effs/*.c` and `*.js` are the reference. -On Linux, the audio pump retains partial device writes and retries recoverable -errors. A zero write, failed recovery, or one second without progress stops -the pump. The next `Audio.write` or `Audio.close` prints the stored error and -exits with status 1. Both calls join the pump and release the device before they -report that error. A full ring still drops the new producer chunk. - ## The C side The compiler splices your `.c` into the program's C source after the diff --git a/tests/io/audio_linux_progress.bend b/tests/io/audio_linux_progress.bend deleted file mode 100644 index b5584ae54..000000000 --- a/tests/io/audio_linux_progress.bend +++ /dev/null @@ -1,95 +0,0 @@ -# Test the Linux audio pump through the native Audio effects. -# The C test uses ALSA's null output and injects device return values. -import Base - -def AudioTest.mode() -> IO(U32): - import "./audio_linux_progress.c" - import "./audio_linux_progress.js" - -def AudioTest.prepare(mode: U32, width: U32) -> IO(Unit): - import "./audio_linux_progress.c" - import "./audio_linux_progress.js" - -def AudioTest.wait(mode: U32) -> 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 failed_call(close: Bool, a: Audio) -> IO(Unit): - match close: - case True{}: - Audio.close(a) - case False{}: - do IO: - pair : Audio & U32 <- Audio.write(a, []) - close_pair(pair) - -def outcome(good: Bool, mode: U32, a: Audio) -> IO(Unit): - match good: - case True{}: - do IO: - Unit <- Audio.close(a) - AudioTest.closed() - case False{}: - do IO: - Unit <- IO.print("before") - Unit <- failed_call(U32.is_eq(mode, 9), a) - IO.print("after") - -def finish(+mode: U32, pair: Audio & U32) -> IO(Unit): - (a, queued) = pair - do IO: - Unit <- AudioTest.wait(mode) - outcome(Bool.or(U32.is_lt(mode, 8), U32.is_gt(mode, 14)), mode, a) - -def consume(+mode: U32, width: U32) -> IO(Unit): - do IO: - Unit <- AudioTest.prepare(mode, width) - a : Audio <- IO.try(Audio, Audio.open(48000)) - pair : Audio & U32 <- Audio.write(a, samples(512n, 1)) - finish(mode, pair) - -def shorts(n: Nat) -> IO(Unit): - match n: - case 0n: - IO.pure(Unit, Unit{}) - case 1n+ +p: - do IO: - Unit <- consume(1, U32.from_nat(1n+p)) - shorts(p) - -def child(matrix: Bool, mode: U32) -> IO(Unit): - match matrix: - case True{}: - shorts(255n) - case False{}: - consume(mode, 256) - -def selected(parent: Bool, +mode: U32) -> IO(Unit): - match parent: - case True{}: - IO.print("audio") - case False{}: - child(U32.is_eq(mode, 1), mode) - -def main() -> IO(Unit): - do IO: - +mode : U32 <- AudioTest.mode() - selected(U32.is_eq(mode, 0), mode) - -#|audio diff --git a/tests/io/audio_linux_progress.c b/tests/io/audio_linux_progress.c deleted file mode 100644 index ad7bc0709..000000000 --- a/tests/io/audio_linux_progress.c +++ /dev/null @@ -1,268 +0,0 @@ -#ifdef __linux__ -#include -#include -#include -#include - -// Only this program's null output is affected. Positive writes reach ALSA. -// A thread key reports pump exit without reading its private state. -static snd_pcm_t* audio_test_unit; -static void* audio_test_library; -static pthread_key_t audio_test_key; -static u32 audio_test_mode, audio_test_width; -static _Atomic(u32) audio_test_frames, audio_test_bad, audio_test_fault; -static _Atomic(u32) audio_test_writes, audio_test_recovers, audio_test_resumes; -static _Atomic(u32) audio_test_finished, audio_test_closes; - -static void audio_test_exit(void* p) { - atomic_store(&audio_test_finished, 1); -} - -static void* audio_test_symbol(const char* name) { - void* p = snd_dlsym(audio_test_library, 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 stream, int mode) { - typedef int (*Open)(snd_pcm_t**, const char*, snd_pcm_stream_t, int); - int code = ((Open)audio_test_symbol("snd_pcm_open"))(p, "null", stream, mode); - if (code == 0) audio_test_unit = *p; - return code; -} - -snd_pcm_sframes_t snd_pcm_writei(snd_pcm_t* p, const void* data, - snd_pcm_uframes_t frames) { - typedef snd_pcm_sframes_t (*Write)(snd_pcm_t*, const void*, snd_pcm_uframes_t); - const float* pcm = data; - pthread_setspecific(audio_test_key, (void*)1); - bool samples = false; - for (u64 i = 0; i < frames; ++i) samples |= pcm[i * 2] != 0; - if (p == audio_test_unit && (samples - || (audio_test_mode >= 7 && audio_test_mode <= 14 - && atomic_load(&audio_test_fault)))) { - u32 call = atomic_fetch_add(&audio_test_writes, 1); - if (call > 100000) err_fail("audio test: unbounded writes"); - bool first = atomic_exchange(&audio_test_fault, 1) == 0; - if (audio_test_mode == 1 && first) frames = audio_test_width; - if (audio_test_mode == 15 && frames > 17) frames = 17; - if ((audio_test_mode == 16 || audio_test_mode == 17) && call == 0) frames = 64; - if (audio_test_mode == 16 && call == 1) return -EINTR; - if (audio_test_mode == 17 && call == 1) return -EPIPE; - if ((audio_test_mode == 2 && first) || audio_test_mode == 12) return -EINTR; - if ((audio_test_mode == 3 && first) || audio_test_mode == 11) return -EPIPE; - if ((audio_test_mode == 4 && first) || audio_test_mode == 7 - || audio_test_mode == 14) return -ESTRPIPE; - if ((audio_test_mode == 5 && first) || audio_test_mode == 13) return -EAGAIN; - if (audio_test_mode == 8 || audio_test_mode == 9) return -ENODEV; - if (audio_test_mode == 10) return 0; - } - snd_pcm_sframes_t n = ((Write)audio_test_symbol("snd_pcm_writei"))(p, data, frames); - if (n > 0 && p == audio_test_unit) { - u32 count = atomic_load(&audio_test_frames); - for (snd_pcm_sframes_t i = 0; i < n; ++i) { - if (pcm[i * 2] == 0) { - if (pcm[i * 2 + 1] != 0) atomic_store(&audio_test_bad, 1); - } else { - ++count; - if (pcm[i * 2] != (float)count || pcm[i * 2 + 1] != -(float)count) - atomic_store(&audio_test_bad, 1); - } - } - atomic_store(&audio_test_frames, count); - } - if (!samples) { - struct timespec t = { 0, 1000000 }; - nanosleep(&t, NULL); - } - return n; -} - -int snd_pcm_recover(snd_pcm_t* p, int code, int silent) { - typedef int (*Recover)(snd_pcm_t*, int, int); - if (p == audio_test_unit) atomic_fetch_add(&audio_test_recovers, 1); - return ((Recover)audio_test_symbol("snd_pcm_recover"))(p, code, silent); -} - -int snd_pcm_resume(snd_pcm_t* p) { - if (p == audio_test_unit) { - u32 n = atomic_fetch_add(&audio_test_resumes, 1); - return audio_test_mode == 7 || n < 2 ? -EAGAIN : -ENOSYS; - } - typedef int (*Resume)(snd_pcm_t*); - return ((Resume)audio_test_symbol("snd_pcm_resume"))(p); -} - -int snd_pcm_prepare(snd_pcm_t* p) { - if (p == audio_test_unit && atomic_load(&audio_test_fault) - && (audio_test_mode == 11 || audio_test_mode == 14)) return -EIO; - typedef int (*Prepare)(snd_pcm_t*); - return ((Prepare)audio_test_symbol("snd_pcm_prepare"))(p); -} - -int snd_pcm_close(snd_pcm_t* p) { - typedef int (*Close)(snd_pcm_t*); - bool failed_close = p == audio_test_unit - && (audio_test_mode == 8 || audio_test_mode == 9); - if (p == audio_test_unit) { - if (audio_test_mode >= 8 && audio_test_mode <= 14 - && !atomic_load(&audio_test_finished)) err_fail("audio test: close before pump exit"); - atomic_fetch_add(&audio_test_closes, 1); - audio_test_unit = NULL; - } - int code = ((Close)audio_test_symbol("snd_pcm_close"))(p); - if (failed_close) { - if (code != 0 || !atomic_load(&audio_test_finished)) - err_fail("audio test: failed close did not release"); - puts("closed"); - } - return code; -} - -static void audio_test_read(int fd, char* out, size_t cap) { - size_t n = 0; - ssize_t got; - while (n + 1 < cap && (got = read(fd, out + n, cap - n - 1)) > 0) n += got; - out[n] = 0; - close(fd); -} - -static void audio_test_child(u32 mode) { - int out[2], err[2]; - if (pipe(out) || pipe(err)) err_fail("audio test: pipe"); - io_sync(); - extern char** environ; - size_t count = 0; - while (environ[count] != NULL) ++count; - char* vars[count + 2]; - size_t at = 0; - for (size_t i = 0; i < count; ++i) - if (strncmp(environ[i], "BEND_TEST_AUDIO_MODE=", - sizeof "BEND_TEST_AUDIO_MODE=" - 1)) vars[at++] = environ[i]; - char value[64]; - snprintf(value, sizeof value, "BEND_TEST_AUDIO_MODE=%u", mode); - vars[at++] = value; - vars[at] = NULL; - char* args[] = { "audio-test", "--threads", "1", "--gpu", "off", NULL }; - posix_spawn_file_actions_t actions; - posix_spawn_file_actions_init(&actions); - posix_spawn_file_actions_addclose(&actions, out[0]); - posix_spawn_file_actions_addclose(&actions, err[0]); - posix_spawn_file_actions_adddup2(&actions, out[1], 1); - posix_spawn_file_actions_adddup2(&actions, err[1], 2); - posix_spawn_file_actions_addclose(&actions, out[1]); - posix_spawn_file_actions_addclose(&actions, err[1]); - pid_t pid; - int spawn_code = posix_spawn(&pid, "/proc/self/exe", &actions, NULL, args, vars); - posix_spawn_file_actions_destroy(&actions); - if (spawn_code != 0) err_fail("audio test: spawn"); - close(out[1]); close(err[1]); - int status = 0; - u64 start = io_tick(); - while (waitpid(pid, &status, WNOHANG) == 0) { - if (io_tick() - start > 2500000000ull) { - kill(pid, SIGKILL); - waitpid(pid, &status, 0); - err_fail("audio test: child timeout"); - } - struct timespec t = { 0, 1000000 }; - nanosleep(&t, NULL); - } - char text[128], errors[512]; - audio_test_read(out[0], text, sizeof text); - audio_test_read(err[0], errors, sizeof errors); - bool failed = mode >= 8 && mode <= 14; - int code = mode == 8 || mode == 9 ? ENODEV - : mode == 12 || mode == 13 ? ETIMEDOUT : EIO; - char expected[160]; - snprintf(expected, sizeof expected, "bend: Audio: %s\n", snd_strerror(-code)); - if (!WIFEXITED(status) || WEXITSTATUS(status) != (failed ? 1 : 0) - || strcmp(text, mode == 8 || mode == 9 ? "before\nclosed\n" - : failed ? "before\n" : "") - || (failed ? strcmp(errors, expected) != 0 : errors[0] != 0)) { - fprintf(stderr, "audio test mode %u: status %d; stdout %s; stderr %s\n", - mode, status, text, errors); - err_fail("audio test: consumer failed"); - } -} -#endif - -Term audio_test_mode_run(Env e, Term* f, IoWork* w) { -#ifdef __linux__ - const char* mode = getenv("BEND_TEST_AUDIO_MODE"); - if (mode != NULL) return (u32)strtoul(mode, NULL, 10); - for (u32 i = 1; i <= 17; ++i) audio_test_child(i); -#endif - return 0; -} - -Term audio_test_prepare_run(Env e, Term* f, IoWork* w) { -#ifdef __linux__ - audio_test_mode = (u32)f[0]; - audio_test_width = (u32)f[1]; - const char* width = getenv("BEND_TEST_AUDIO_WIDTH"); - if (width != NULL && audio_test_mode == 1) - audio_test_width = (u32)strtoul(width, NULL, 10); - atomic_store(&audio_test_frames, 0); - atomic_store(&audio_test_bad, 0); - atomic_store(&audio_test_fault, 0); - atomic_store(&audio_test_writes, 0); - atomic_store(&audio_test_recovers, 0); - atomic_store(&audio_test_resumes, 0); - atomic_store(&audio_test_finished, 0); - atomic_store(&audio_test_closes, 0); -#endif - return term_pak(CID(Unit), 0); -} - -Term audio_test_wait_run(Env e, Term* f, IoWork* w) { -#ifdef __linux__ - u32 mode = (u32)f[0]; - u64 start = io_tick(); - while (mode == 7 ? atomic_load(&audio_test_resumes) < 1 - : mode >= 8 && mode <= 14 ? atomic_load(&audio_test_finished) == 0 - : atomic_load(&audio_test_frames) < 512) { - if (atomic_load(&audio_test_bad) - || io_tick() - start > (mode >= 8 && mode <= 14 ? 1500000000ull : 300000000ull)) { - fprintf(stderr, "audio test: frames=%u writes=%u recoveries=%u resumes=%u\n", - atomic_load(&audio_test_frames), atomic_load(&audio_test_writes), - atomic_load(&audio_test_recovers), atomic_load(&audio_test_resumes)); - err_fail("audio test: pump made no progress"); - } - struct timespec t = { 0, 100000 }; - nanosleep(&t, NULL); - } - if (atomic_load(&audio_test_bad)) err_fail("audio test: frame order or channels"); - if ((mode == 8 || mode == 9) && (atomic_load(&audio_test_writes) != 1 - || atomic_load(&audio_test_recovers) != 1)) err_fail("audio test: terminal retry"); - if ((mode == 12 || mode == 13) && atomic_load(&audio_test_writes) > 2000) - err_fail("audio test: retries have no wait"); - if ((mode == 16 || mode == 17) && atomic_load(&audio_test_recovers) != 1) - err_fail("audio test: partial write did not reach recovery"); -#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_finished) || atomic_load(&audio_test_closes) != 1) - err_fail("audio test: close did not join and release"); - if (audio_test_mode != 7 && (atomic_load(&audio_test_frames) != 512 - || atomic_load(&audio_test_bad))) 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_library = snd_dlopen(NULL, RTLD_LAZY, NULL, 0); - if (audio_test_library == NULL) err_fail("audio test: missing ALSA library"); - if (pthread_key_create(&audio_test_key, audio_test_exit)) - err_fail("audio test: thread key"); -#endif - io_eff(CID(AudioTest.mode), audio_test_mode_run); - io_eff(CID(AudioTest.prepare), audio_test_prepare_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 deleted file mode 100644 index 9ac3ec4c1..000000000 --- a/tests/io/audio_linux_progress.js +++ /dev/null @@ -1,5 +0,0 @@ -// The C test exercises the Linux native runtime. Other lanes skip it. -io_eff(CID(AudioTest.mode), () => 0); -io_eff(CID(AudioTest.prepare), () => ({ $: CID(Unit) })); -io_eff(CID(AudioTest.wait), () => ({ $: CID(Unit) })); -io_eff(CID(AudioTest.closed), () => ({ $: CID(Unit) })); From f682e965ef823c0d978b9089069d051a003e8bfe Mon Sep 17 00:00:00 2001 From: Lorenzobattistela Date: Mon, 12 Oct 2026 11:51:42 -0300 Subject: [PATCH 3/3] test(audio): the Linux pump keeps every frame through short and failed writes, and stops on a dead device One process runs 262 cases against ALSA's null output with injected write results: first writes of 1-255 frames, EINTR and EPIPE once, writes capped at 17, a partial write then EINTR or EPIPE (all must deliver 512 frames in order), and ENODEV or 0 forever (the pump must stop after one write, and close must join it). Other systems run no cases. Main fails it; so do the alt with the offset, the recover check or the zero check removed. --- tests/io/audio_linux_progress.bend | 52 +++++++++++ tests/io/audio_linux_progress.c | 137 +++++++++++++++++++++++++++++ tests/io/audio_linux_progress.js | 5 ++ 3 files changed, 194 insertions(+) create mode 100644 tests/io/audio_linux_progress.bend create mode 100644 tests/io/audio_linux_progress.c create mode 100644 tests/io/audio_linux_progress.js 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) }));