Use total_frames to check for ringbuffer over-/underflow

Author: mgeier

examples/fetch_stats.py

@@ -14,8 +14,17 @@
 
 
 def print_stats(obj):
-    print('  blocks:', obj.stats.blocks)
-    print('  overflows:', obj.stats.input_overflows)
+    print('          blocks:', obj.stats.blocks)
+    print('       overflows:', obj.stats.input_overflows)
+
+
+def print_action(action):
+    print('            type:', next(
+        k for k, v in vars(rtmixer).items() if v == action.type))
+    print('  requested time:', action.requested_time)
+    print('     actual time:', action.actual_time)
+    print('    total frames:', action.total_frames)
+    print('     done frames:', action.done_frames)
 
 
 stream = rtmixer.Recorder(
@@ -31,23 +40,31 @@ def print_stats(obj):
 assert stream.stats.input_overflows == 0
 
 with stream:
-    print('waiting a few seconds')
+    print('waiting a few seconds ...')
     sd.sleep(3 * 1000)
     print('checking stats:')
     action = stream.fetch_and_reset_stats()
     stream.wait(action)
     print_stats(action)
-    print('starting recording to buffer')
+    print('starting recording to buffer ...')
     action = stream.record_buffer(buffer, channels=channels)
-    # TODO: check if recording started successfully
+    # TODO: check if recording started successfully?
     stream.wait(action)
+    print('result:')
+    print_action(action)
     print('stats from finished recording:')
     print_stats(action)
-    print('starting recording to ringbuffer')
+    print('starting recording to ringbuffer ...')
     action = stream.record_ringbuffer(ringbuffer, channels=channels)
-    # TODO: check if recording started successfully
-    # TODO: record ringbuffer, but don't read from it (detect overflow)
-    sd.sleep(3 * 1000)
+    # TODO: check if recording started successfully?
+    # NB: We are writing to the ringbuffer, but we are not reading from it,
+    # which will lead to an overflow
+    print('waiting for ring buffer to fill up ...')
+    stream.wait(action)
+    print('result:')
+    print_action(action)
+    print('recording was stopped because of ringbuffer overflow:',
+          action.done_frames != action.total_frames)
     print('checking stats:')
     action = stream.fetch_and_reset_stats()
     stream.wait(action)

examples/sampler.py

@@ -79,14 +79,15 @@ def on_key_release(self, event):
         # TODO: fade out (both recording and playback)?
         assert sample.action is not None
         if sample.action.type == rtmixer.RECORD_RINGBUFFER:
-            # TODO: check for errors/xruns? check for rinbuffer overflow?
             # Stop recording
+            sample.keep_alive = sample.action
             sample.action = self.stream.cancel(sample.action)
             # A CANCEL action is returned which is checked by poll_ringbuffer()
         elif sample.action.type == rtmixer.PLAY_BUFFER:
             # TODO: check for errors/xruns?
             # Stop playback (if still running)
             if sample.action in self.stream.actions:
+                sample.keep_alive = sample.action
                 sample.action = self.stream.cancel(sample.action)
                 # TODO: do something with sample.action?
         elif sample.action.type == rtmixer.CANCEL:
@@ -99,17 +100,22 @@ def poll_ringbuffer(self, sample):
         assert sample.action is not None
         assert sample.action.type in (rtmixer.RECORD_RINGBUFFER,
                                       rtmixer.CANCEL)
-        # TODO: check for errors? is everything still working OK?
-        # TODO: check for xruns?
-        chunk = sample.ringbuffer.read()
-        if chunk:
-            sample.buffer.extend(chunk)
-
         if sample.action not in self.stream.actions:
             # Recording is finished
-            # TODO: check for errors in CANCEL action?
             self.rec_counter -= 1
+            if sample.action.type == rtmixer.CANCEL:
+                # TODO: check for errors in CANCEL action?
+                action = sample.action.action
+            else:
+                action = sample.action
+            assert action.type == rtmixer.RECORD_RINGBUFFER
+            if action.done_frames != action.total_frames:
+                print('error while recording (ringbuffer too short?)')
+            # TODO: check for xruns?
+            # NB: We make sure that the ringbuffer is emptied after recording:
+            sample.buffer.extend(sample.ringbuffer.read())
         else:
+            sample.buffer.extend(sample.ringbuffer.read())
             # Set polling rate based on input latency (which may change!):
             self.after(int(self.stream.latency[0] * 1000),
                        self.poll_ringbuffer, sample)

src/rtmixer.c

@@ -28,7 +28,7 @@ void remove_action(struct action** addr, const struct state* state)
     , &action, 1);
   if (written != 1)
   {
-    // TODO: do something? Stop callback? Log error (in "state")?
+    // TODO: Stop callback! Unrecoverable error! Log error (in "state")?
     printf("result queue is full\n");
   }
 }
@@ -37,6 +37,8 @@ void get_stats(PaStreamCallbackFlags flags, struct stats* stats)
 {
   stats->blocks++;
 
+  // TODO: store min/max block size?
+
   if (flags & paInputUnderflow)  { stats->input_underflows++; }
   if (flags & paInputOverflow)   { stats->input_overflows++; }
   if (flags & paOutputUnderflow) { stats->output_underflows++; }
@@ -116,6 +118,9 @@ int callback(const void* input, void* output, frame_t frameCount
 
   get_new_actions(state);
 
+  // TODO: store min/max available space in result_q?
+  // TODO: use worst case from before/after the "while" loop?
+
   struct action** actionaddr = &(state->actions);
   while (*actionaddr)
   {
@@ -185,7 +190,8 @@ int callback(const void* input, void* output, frame_t frameCount
               {
                 // Removal is scheduled before playback/recording begins
 
-                // TODO: save some status information?
+                // TODO: save some more status information?
+                delinquent->total_frames = 0;
                 remove_action(i, state);
                 break;
               }
@@ -199,9 +205,9 @@ int callback(const void* input, void* output, frame_t frameCount
               }
             }
 
-            if (delinquent->total_frames + delinquent_offset > offset)
+            if (delinquent->total_frames == ULONG_MAX
+             || delinquent->total_frames + delinquent_offset > offset)
             {
-              CALLBACK_ASSERT(offset >= delinquent_offset);
               delinquent->total_frames = offset - delinquent_offset;
             }
             else
@@ -382,9 +388,6 @@ int callback(const void* input, void* output, frame_t frameCount
       if (totalsize < (ring_buffer_size_t)frames)
       {
         // Ring buffer is empty or full
-
-        // TODO: store some information in action?
-
         remove_action(actionaddr, state);
         continue;
       }

src/rtmixer.h

@@ -35,17 +35,16 @@ struct action
   bool allow_belated;
   const PaTime requested_time;
   PaTime actual_time;
-  struct action* next;
+  struct action* next;  // Used to create singly linked list of actions
   union {
     float* const buffer;
     struct PaUtilRingBuffer* const ringbuffer;
     struct action* const action;  // Used in CANCEL
   };
   frame_t total_frames;
   frame_t done_frames;
-  // TODO: something to store the result of the action?
   struct stats stats;
-  // TODO: queue usage: store smallest available write/read size?
+  // TODO: ringbuffer usage: store smallest available write/read size?
   const frame_t channels;  // Size of the following array
   const frame_t mapping[];  // "flexible array member"
 };
@@ -59,6 +58,7 @@ struct state
   struct PaUtilRingBuffer* const result_q;  // Q for results and cmd disposal
   struct action* actions;  // Singly linked list of actions
   struct stats stats;
+  // TODO: result_q usage?
 };
 
 int callback(const void* input, void* output, frame_t frameCount