From 7125d164b87bc22973f3f578085d8933b709aec4 Mon Sep 17 00:00:00 2001 From: Eric Larson Date: Tue, 20 Oct 2026 10:57:08 -0400 Subject: [PATCH 1/6] Fix minor bugs with queue and teardown --- .gitignore | 1 + portaudio | 2 +- src/rtmixer.c | 5 ++++- src/rtmixer.py | 16 +++++++++++++++- 4 files changed, 21 insertions(+), 3 deletions(-) diff --git a/.gitignore b/.gitignore index 3c1a62c..679bf0f 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,4 @@ __pycache__ .eggs /dist/ /wheelhouse/ +.DS_Store \ No newline at end of file diff --git a/portaudio b/portaudio index 147dd72..7c45fed 160000 --- a/portaudio +++ b/portaudio @@ -1 +1 @@ -Subproject commit 147dd722548358763a8b649b3e4b41dfffbcfbb6 +Subproject commit 7c45fed9322bd86b5dd0da4547b15d9ac3b9b0ac diff --git a/src/rtmixer.c b/src/rtmixer.c index 9ac3133..6506220 100644 --- a/src/rtmixer.c +++ b/src/rtmixer.c @@ -128,7 +128,10 @@ int callback(const void* input, void* output, frame_t frameCount struct state* state = userData; CALLBACK_ASSERT(state); - memset(output, 0, sizeof(float) * state->output_channels * frameCount); + if (output) // NULL for input-only streams + { + memset(output, 0, sizeof(float) * state->output_channels * frameCount); + } get_stats(frameCount, statusFlags, &(state->stats)); diff --git a/src/rtmixer.py b/src/rtmixer.py index 0cdde95..16cb190 100644 --- a/src/rtmixer.py +++ b/src/rtmixer.py @@ -41,6 +41,15 @@ def __init__(self, kind, qsize=16, **kwargs): self._actions = {} self._temp_action_ptr = _ffi.new('struct action**') + def __del__(self): + # The C callback uses memory owned by this object, so the stream + # must be closed before that memory is freed. + try: + if not self.closed: + self.close() + except Exception: + pass # e.g. __init__() failed or interpreter shutdown + @property def actions(self): """The set of active "actions".""" @@ -76,7 +85,8 @@ def cancel(self, action, time=0, allow_belated=True): requested_time=time, action=action, )) - self._enqueue(cancel_action) + # The callback dereferences *action*, so it must be kept alive: + self._enqueue(cancel_action, keep_alive=action) return cancel_action def fetch_and_reset_stats(self, time=0, allow_belated=True): @@ -126,6 +136,10 @@ def _check_channels(self, channels, kind): def _enqueue(self, action, keep_alive=None): self._drain_result_q() + # Each action ends up in the result queue exactly once, so limiting + # the number of pending actions guarantees that it never overflows: + if len(self._actions) >= len(self._result_q): + raise RuntimeError('Too many pending actions') self._temp_action_ptr[0] = action ret = self._action_q.write(self._temp_action_ptr) if ret != 1: From 54c1d91a7db167228dbd95c2bc067e5a1f73f5b1 Mon Sep 17 00:00:00 2001 From: Eric Larson Date: Mon, 26 Oct 2026 15:50:21 -0400 Subject: [PATCH 2/6] Doc and tweak --- src/rtmixer.py | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/src/rtmixer.py b/src/rtmixer.py index 16cb190..705127b 100644 --- a/src/rtmixer.py +++ b/src/rtmixer.py @@ -45,8 +45,8 @@ def __del__(self): # The C callback uses memory owned by this object, so the stream # must be closed before that memory is freed. try: - if not self.closed: - self.close() + self.abort() + self.close() except Exception: pass # e.g. __init__() failed or interpreter shutdown @@ -139,7 +139,8 @@ def _enqueue(self, action, keep_alive=None): # Each action ends up in the result queue exactly once, so limiting # the number of pending actions guarantees that it never overflows: if len(self._actions) >= len(self._result_q): - raise RuntimeError('Too many pending actions') + raise RuntimeError( + 'Too many pending actions, consider increasing qsize') self._temp_action_ptr[0] = action ret = self._action_q.write(self._temp_action_ptr) if ret != 1: @@ -163,6 +164,13 @@ class Mixer(_Base): except *callback* (a callback function implemented in C is used internally) and *dtype* (which is always ``'float32'``). + The additional keyword argument *qsize* (default: 16) sets the size + of the queues used to communicate with the callback, which is also + the maximum number of pending actions, i.e. actions that have been + started (or scheduled) but haven't finished yet. If an action is + started while *qsize* actions are pending, a `RuntimeError` is + raised. It must be a power of 2. + Uses default values from `sounddevice.default` (except *dtype*, which is always ``'float32'``). @@ -231,6 +239,8 @@ class Recorder(_Base): except *callback* (a callback function implemented in C is used internally) and *dtype* (which is always ``'float32'``). + The additional keyword argument *qsize* is described in `Mixer`. + Uses default values from `sounddevice.default` (except *dtype*, which is always ``'float32'``). @@ -296,6 +306,8 @@ class MixerAndRecorder(Mixer, Recorder): *callback* (a callback function implemented in C is used internally) and *dtype* (which is always ``'float32'``). + The additional keyword argument *qsize* is described in `Mixer`. + Uses default values from `sounddevice.default` (except *dtype*, which is always ``'float32'``). From 23774176568c716a0238ed5ff3c7b443358c31f2 Mon Sep 17 00:00:00 2001 From: Eric Larson Date: Wed, 4 Nov 2026 14:44:05 -0500 Subject: [PATCH 3/6] Better queue --- src/rtmixer.c | 25 ++++++++++++++++++------- src/rtmixer.h | 1 + src/rtmixer.py | 20 +++++++++----------- 3 files changed, 28 insertions(+), 18 deletions(-) diff --git a/src/rtmixer.c b/src/rtmixer.c index 6506220..867d659 100644 --- a/src/rtmixer.c +++ b/src/rtmixer.c @@ -20,17 +20,27 @@ static const struct stats EMPTY_STATS; }} while (false) #endif -void remove_action(struct action** addr, const struct state* state) +// The action is moved to the list of done actions, see flush_done_actions() +void remove_action(struct action** addr, struct state* state) { struct action* action = *addr; *addr = action->next; // Current action is removed from list - action->next = NULL; - ring_buffer_size_t written = PaUtil_WriteRingBuffer(state->result_q - , &action, 1); - if (written != 1) + action->next = state->done_actions; + state->done_actions = action; +} + +// Done actions that don't fit into result_q are kept for the next block +void flush_done_actions(struct state* state) +{ + while (state->done_actions) { - // TODO: Stop callback! Unrecoverable error! Log error (in "state")? - printf("result queue is full\n"); + struct action* action = state->done_actions; + if (PaUtil_WriteRingBuffer(state->result_q, &action, 1) != 1) + { + break; // result_q is full, try again in the next block + } + state->done_actions = action->next; + action->next = NULL; } } @@ -423,5 +433,6 @@ int callback(const void* input, void* output, frame_t frameCount } actionaddr = &(action->next); } + flush_done_actions(state); return paContinue; } diff --git a/src/rtmixer.h b/src/rtmixer.h index 9df4a4d..df7677e 100644 --- a/src/rtmixer.h +++ b/src/rtmixer.h @@ -55,6 +55,7 @@ struct state struct PaUtilRingBuffer* const action_q; // Queue for incoming commands struct PaUtilRingBuffer* const result_q; // Q for results and cmd disposal struct action* actions; // Singly linked list of actions + struct action* done_actions; // Finished, but not yet in result_q struct stats stats; // TODO: result_q usage? }; diff --git a/src/rtmixer.py b/src/rtmixer.py index 705127b..b9e3d40 100644 --- a/src/rtmixer.py +++ b/src/rtmixer.py @@ -136,15 +136,11 @@ def _check_channels(self, channels, kind): def _enqueue(self, action, keep_alive=None): self._drain_result_q() - # Each action ends up in the result queue exactly once, so limiting - # the number of pending actions guarantees that it never overflows: - if len(self._actions) >= len(self._result_q): - raise RuntimeError( - 'Too many pending actions, consider increasing qsize') self._temp_action_ptr[0] = action ret = self._action_q.write(self._temp_action_ptr) if ret != 1: - raise RuntimeError('Action queue is full') + raise RuntimeError( + 'Action queue is full, consider increasing qsize') assert action not in self._actions self._actions[action] = keep_alive @@ -165,11 +161,13 @@ class Mixer(_Base): internally) and *dtype* (which is always ``'float32'``). The additional keyword argument *qsize* (default: 16) sets the size - of the queues used to communicate with the callback, which is also - the maximum number of pending actions, i.e. actions that have been - started (or scheduled) but haven't finished yet. If an action is - started while *qsize* actions are pending, a `RuntimeError` is - raised. It must be a power of 2. + of the queues used to communicate with the callback. It must be a + power of 2. The callback takes new actions from its queue once per + audio block, so *qsize* limits how many actions can be started + within one block (if exceeded, a `RuntimeError` is raised), but not + how many actions can be active at the same time. If many actions + finish at once, some of them may be removed from `actions` a few + blocks later. Uses default values from `sounddevice.default` (except *dtype*, which is always ``'float32'``). From 1486f9a9bc54309f0ec4e40d97d4be9498b9d03e Mon Sep 17 00:00:00 2001 From: Eric Larson Date: Mon, 9 Nov 2026 09:58:07 -0500 Subject: [PATCH 4/6] Better order --- src/rtmixer.c | 31 +++++++++++++++++++------------ src/rtmixer.h | 1 + src/rtmixer.py | 18 +++++++++++------- 3 files changed, 31 insertions(+), 19 deletions(-) diff --git a/src/rtmixer.c b/src/rtmixer.c index 867d659..7e910dc 100644 --- a/src/rtmixer.c +++ b/src/rtmixer.c @@ -20,27 +20,34 @@ static const struct stats EMPTY_STATS; }} while (false) #endif -// The action is moved to the list of done actions, see flush_done_actions() +// The action is appended to the list of done actions, +// see flush_done_actions() void remove_action(struct action** addr, struct state* state) { struct action* action = *addr; *addr = action->next; // Current action is removed from list - action->next = state->done_actions; - state->done_actions = action; + action->next = NULL; + if (state->last_done_action) + { + state->last_done_action->next = action; + } + else + { + state->done_actions = action; + } + state->last_done_action = action; } -// Done actions that don't fit into result_q are kept for the next block +// All done actions are sent as a single linked list (in the order they +// finished). If result_q is full, they are kept until the next block. void flush_done_actions(struct state* state) { - while (state->done_actions) + if (state->done_actions + && PaUtil_WriteRingBuffer(state->result_q, &(state->done_actions), 1)) { - struct action* action = state->done_actions; - if (PaUtil_WriteRingBuffer(state->result_q, &action, 1) != 1) - { - break; // result_q is full, try again in the next block - } - state->done_actions = action->next; - action->next = NULL; + // The actions now belong to the Python side, don't touch them anymore! + state->done_actions = NULL; + state->last_done_action = NULL; } } diff --git a/src/rtmixer.h b/src/rtmixer.h index df7677e..04e50f9 100644 --- a/src/rtmixer.h +++ b/src/rtmixer.h @@ -56,6 +56,7 @@ struct state struct PaUtilRingBuffer* const result_q; // Q for results and cmd disposal struct action* actions; // Singly linked list of actions struct action* done_actions; // Finished, but not yet in result_q + struct action* last_done_action; // To append to done_actions struct stats stats; // TODO: result_q usage? }; diff --git a/src/rtmixer.py b/src/rtmixer.py index b9e3d40..d481614 100644 --- a/src/rtmixer.py +++ b/src/rtmixer.py @@ -147,10 +147,16 @@ def _enqueue(self, action, keep_alive=None): def _drain_result_q(self): """Get actions from the result queue and discard them.""" while self._result_q.readinto(self._temp_action_ptr): - try: - del self._actions[self._temp_action_ptr[0]] - except KeyError: - assert False + # Each item is a linked list of actions + action = self._temp_action_ptr[0] + while action != _ffi.NULL: + # Get next pointer before the action is (potentially) freed: + next_action = action.next + try: + del self._actions[action] + except KeyError: + assert False + action = next_action class Mixer(_Base): @@ -165,9 +171,7 @@ class Mixer(_Base): power of 2. The callback takes new actions from its queue once per audio block, so *qsize* limits how many actions can be started within one block (if exceeded, a `RuntimeError` is raised), but not - how many actions can be active at the same time. If many actions - finish at once, some of them may be removed from `actions` a few - blocks later. + how many actions can be active at the same time. Uses default values from `sounddevice.default` (except *dtype*, which is always ``'float32'``). From 49ead25e90bf3cd4742699da855390ed028d450e Mon Sep 17 00:00:00 2001 From: Eric Larson Date: Mon, 9 Nov 2026 10:24:47 -0500 Subject: [PATCH 5/6] More --- .gitignore | 1 - AGENTS.md | 5 +++-- NEWS.rst | 7 +++++++ src/rtmixer.c | 2 +- 4 files changed, 11 insertions(+), 4 deletions(-) diff --git a/.gitignore b/.gitignore index 679bf0f..3c1a62c 100644 --- a/.gitignore +++ b/.gitignore @@ -6,4 +6,3 @@ __pycache__ .eggs /dist/ /wheelhouse/ -.DS_Store \ No newline at end of file diff --git a/AGENTS.md b/AGENTS.md index 5d733fd..fdb4a76 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -129,8 +129,9 @@ Each command (`play_buffer`, `cancel`, `fetch_and_reset_stats`, ...) allocates a PortAudio ring buffer. Once per audio block the callback reads new actions into a singly linked list (`CANCEL` actions go to the front, everything else to the back), processes each one against the block's timing info (`requested_time`, `allow_belated` encoded via `actual_time`), and moves -finished actions to `done_actions`, which are flushed into `result_q` as space allows. Python -drains `result_q` in `_drain_result_q()` (called from `_enqueue()` and the `actions` property) +finished actions to `done_actions`, which is sent to `result_q` as one linked list per block +(and kept for the next block if `result_q` is full). Python drains `result_q` in +`_drain_result_q()` (called from `_enqueue()` and the `actions` property), walks each list, and drops its references. ### Memory ownership diff --git a/NEWS.rst b/NEWS.rst index 1eee335..4f957ca 100644 --- a/NEWS.rst +++ b/NEWS.rst @@ -1,3 +1,10 @@ +Version 0.1.8 -- unreleased + * Fix lost results and a hanging ``wait()`` when many actions finish + in the same audio block + * Fix possible crashes when a stream is garbage-collected while it is + running, or when an action is garbage-collected after being cancelled + * Update PortAudio submodule to the current development version + Version 0.1.7 -- 2024-07-02 -- PyPI__ -- diff__ * Restore the ``x86_64`` Linux wheels that were missing from 0.1.6 diff --git a/src/rtmixer.c b/src/rtmixer.c index 7e910dc..7fcf0f9 100644 --- a/src/rtmixer.c +++ b/src/rtmixer.c @@ -43,7 +43,7 @@ void remove_action(struct action** addr, struct state* state) void flush_done_actions(struct state* state) { if (state->done_actions - && PaUtil_WriteRingBuffer(state->result_q, &(state->done_actions), 1)) + && PaUtil_WriteRingBuffer(state->result_q, &(state->done_actions), 1)) { // The actions now belong to the Python side, don't touch them anymore! state->done_actions = NULL; From 349ca064e76fe0733d21b4544480bcf48aa21aa3 Mon Sep 17 00:00:00 2001 From: Eric Larson Date: Tue, 10 Nov 2026 09:39:12 -0500 Subject: [PATCH 6/6] Better style --- src/rtmixer.c | 34 +++++++++++++++++----------------- src/rtmixer.h | 1 - 2 files changed, 17 insertions(+), 18 deletions(-) diff --git a/src/rtmixer.c b/src/rtmixer.c index 7fcf0f9..ec29ede 100644 --- a/src/rtmixer.c +++ b/src/rtmixer.c @@ -20,22 +20,20 @@ static const struct stats EMPTY_STATS; }} while (false) #endif -// The action is appended to the list of done actions, -// see flush_done_actions() -void remove_action(struct action** addr, struct state* state) +// Appends the action to the list of done actions (see flush_done_actions()) +// and returns the address where the next done action will be appended +struct action** remove_action(struct action** addr + , struct action** done_addr) { struct action* action = *addr; *addr = action->next; // Current action is removed from list action->next = NULL; - if (state->last_done_action) + while (*done_addr) { - state->last_done_action->next = action; + done_addr = &((*done_addr)->next); } - else - { - state->done_actions = action; - } - state->last_done_action = action; + *done_addr = action; + return &(action->next); } // All done actions are sent as a single linked list (in the order they @@ -47,7 +45,6 @@ void flush_done_actions(struct state* state) { // The actions now belong to the Python side, don't touch them anymore! state->done_actions = NULL; - state->last_done_action = NULL; } } @@ -158,6 +155,8 @@ int callback(const void* input, void* output, frame_t frameCount // TODO: use worst case from before/after the "while" loop? struct action** actionaddr = &(state->actions); + // Done actions from previous blocks are skipped in remove_action() + struct action** done_addr = &(state->done_actions); while (*actionaddr) { struct action* const action = *actionaddr; @@ -194,7 +193,7 @@ int callback(const void* input, void* output, frame_t frameCount if (action->actual_time == 0.0) { // allow_belated == False - remove_action(actionaddr, state); + done_addr = remove_action(actionaddr, done_addr); continue; } action->actual_time = time; @@ -228,7 +227,7 @@ int callback(const void* input, void* output, frame_t frameCount // TODO: save some more status information? delinquent->total_frames = 0; - remove_action(i, state); + done_addr = remove_action(i, done_addr); break; } } @@ -273,7 +272,8 @@ int callback(const void* input, void* output, frame_t frameCount } // TODO: what if the action to cancel wasn't found? - remove_action(actionaddr, state); // Remove the CANCEL action itself + // Remove the CANCEL action itself + done_addr = remove_action(actionaddr, done_addr); continue; } @@ -283,7 +283,7 @@ int callback(const void* input, void* output, frame_t frameCount { action->stats = state->stats; state->stats = EMPTY_STATS; - remove_action(actionaddr, state); + done_addr = remove_action(actionaddr, done_addr); continue; } @@ -426,7 +426,7 @@ int callback(const void* input, void* output, frame_t frameCount if (totalsize < (ring_buffer_size_t)frames) { // Ring buffer is empty or full - remove_action(actionaddr, state); + done_addr = remove_action(actionaddr, done_addr); continue; } } @@ -435,7 +435,7 @@ int callback(const void* input, void* output, frame_t frameCount if (action->done_frames == action->total_frames) { - remove_action(actionaddr, state); + done_addr = remove_action(actionaddr, done_addr); continue; } actionaddr = &(action->next); diff --git a/src/rtmixer.h b/src/rtmixer.h index 04e50f9..df7677e 100644 --- a/src/rtmixer.h +++ b/src/rtmixer.h @@ -56,7 +56,6 @@ struct state struct PaUtilRingBuffer* const result_q; // Q for results and cmd disposal struct action* actions; // Singly linked list of actions struct action* done_actions; // Finished, but not yet in result_q - struct action* last_done_action; // To append to done_actions struct stats stats; // TODO: result_q usage? };