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..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; } } @@ -128,7 +138,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)); @@ -420,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 0cdde95..b9e3d40 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: + self.abort() + 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): @@ -129,7 +139,8 @@ def _enqueue(self, action, keep_alive=None): 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 @@ -149,6 +160,15 @@ 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. 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'``). @@ -217,6 +237,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'``). @@ -282,6 +304,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'``).