Fix minor bugs with queue and teardown

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

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
 

src/rtmixer.c

@@ -20,17 +20,31 @@ static const struct stats EMPTY_STATS;
   }} while (false)
 #endif
 
-void remove_action(struct action** addr, const 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;
-  ring_buffer_size_t written = PaUtil_WriteRingBuffer(state->result_q
-    , &action, 1);
-  if (written != 1)
+  while (*done_addr)
   {
-    // TODO: Stop callback! Unrecoverable error! Log error (in "state")?
-    printf("result queue is full\n");
+    done_addr = &((*done_addr)->next);
+  }
+  *done_addr = action;
+  return &(action->next);
+}
+
+// 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)
+{
+  if (state->done_actions
+   && 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;
   }
 }
 
@@ -128,7 +142,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));
 
@@ -138,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;
@@ -174,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;
@@ -208,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;
               }
             }
@@ -253,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;
     }
 
@@ -263,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;
     }
 
@@ -406,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;
       }
     }
@@ -415,10 +435,11 @@ 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);
   }
+  flush_done_actions(state);
   return paContinue;
 }

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?
 };

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,17 +139,24 @@ 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
 
     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):
@@ -149,6 +166,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.  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.
+
     Uses default values from `sounddevice.default` (except *dtype*,
     which is always ``'float32'``).
 
@@ -217,6 +241,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 +308,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'``).