audiomixer: small improvements

Improve the audiomixer state handling
Improve PortIO status handling.

pinos/server/node.c

@@ -357,7 +357,9 @@ on_node_event (SpaNode *node, SpaEvent *event, void *user_data)
 
         if ((res = spa_node_process_input (inport->node->node)) < 0)
           pinos_log_warn ("node %p: got process input %d", inport->node, res);
+
       }
+      po->status = SPA_RESULT_NEED_BUFFER;
     }
     res = spa_node_process_output (this->node);
     if (res < 0 && res != SPA_RESULT_HAVE_BUFFER)

spa/plugins/alsa/alsa-sink.c

@@ -550,32 +550,19 @@ spa_alsa_sink_node_process_input (SpaNode *node)
 {
   SpaALSASink *this;
   SpaPortIO *input;
-  SpaALSABuffer *b;
 
   spa_return_val_if_fail (node != NULL, SPA_RESULT_INVALID_ARGUMENTS);
 
   this = SPA_CONTAINER_OF (node, SpaALSASink, node);
+  input = this->io;
+  spa_return_val_if_fail (input != NULL, SPA_RESULT_WRONG_STATE);
 
-  if ((input = this->io) == NULL) {
+  if (input->status == SPA_RESULT_HAVE_BUFFER &&
-    return SPA_RESULT_OK;
+      input->buffer_id != SPA_ID_INVALID) {
-  }
+    SpaALSABuffer *b = &this->buffers[input->buffer_id];
 
-  if (!this->have_format) {
-    spa_log_error (this->log, "alsa-sink %p: no format", this);
-    input->status = SPA_RESULT_NO_FORMAT;
-    return SPA_RESULT_ERROR;
-  }
-
-  if (input->buffer_id >= this->n_buffers) {
-    spa_log_error (this->log, "alsa-sink %p: invalid buffer %u", this, input->buffer_id);
-    input->status = SPA_RESULT_INVALID_BUFFER_ID;
-    return SPA_RESULT_ERROR;
-  }
-
-  b = &this->buffers[input->buffer_id];
     if (!b->outstanding) {
-    spa_log_error (this->log, "alsa-sink %p: buffer %u in use", this, input->buffer_id);
+      spa_log_warn (this->log, "alsa-sink %p: buffer %u in use", this, input->buffer_id);
-    input->buffer_id = SPA_ID_INVALID;
       input->status = SPA_RESULT_INVALID_BUFFER_ID;
       return SPA_RESULT_ERROR;
     }
@@ -586,7 +573,7 @@ spa_alsa_sink_node_process_input (SpaNode *node)
     b->outstanding = false;
     input->buffer_id = SPA_ID_INVALID;
     input->status = SPA_RESULT_OK;
-
+  }
   return SPA_RESULT_OK;
 }
 

spa/plugins/alsa/alsa-source.c

@@ -612,8 +612,13 @@ spa_alsa_source_node_process_output (SpaNode *node)
   spa_return_val_if_fail (node != NULL, SPA_RESULT_INVALID_ARGUMENTS);
 
   this = SPA_CONTAINER_OF (node, SpaALSASource, node);
+  io = this->io;
+  spa_return_val_if_fail (io != NULL, SPA_RESULT_WRONG_STATE);
 
-  if ((io = this->io) && io->buffer_id != SPA_ID_INVALID) {
+  if (io->status == SPA_RESULT_HAVE_BUFFER)
+    return SPA_RESULT_HAVE_BUFFER;
+
+  if (io->buffer_id != SPA_ID_INVALID) {
     recycle_buffer (this, io->buffer_id);
     io->buffer_id = SPA_ID_INVALID;
   }

spa/plugins/alsa/alsa-utils.c

@@ -454,7 +454,7 @@ push_frames (SpaALSAState *state,
 
       b->outstanding = true;
       io->buffer_id = b->outbuf->id;
-      io->status = SPA_RESULT_OK;
+      io->status = SPA_RESULT_HAVE_BUFFER;
 
       state->event_cb (&state->node, &event, state->user_data);
     }
@@ -612,6 +612,9 @@ alsa_on_capture_timeout_event (SpaSource *source)
   state->last_ticks = state->sample_count + avail;
   state->last_monotonic = (int64_t)htstamp.tv_sec * SPA_NSEC_PER_SEC + (int64_t)htstamp.tv_nsec;
 
+  spa_log_trace (state->log, "timeout %ld %d %ld %ld %ld", avail, state->threshold,
+                      state->sample_count, htstamp.tv_sec, htstamp.tv_nsec);
+
   if (avail < state->threshold) {
     if (snd_pcm_state (hndl) == SND_PCM_STATE_SUSPENDED) {
       spa_log_error (state->log, "suspended: try resume");
@@ -645,8 +648,6 @@ alsa_on_capture_timeout_event (SpaSource *source)
   }
   calc_timeout (state->threshold, avail - total_read, state->rate, &htstamp, &ts.it_value);
 
-  spa_log_trace (state->log, "timeout %ld %ld %ld %ld", total_read, avail,
-                                ts.it_value.tv_sec, ts.it_value.tv_nsec);
   ts.it_interval.tv_sec = 0;
   ts.it_interval.tv_nsec = 0;
   timerfd_settime (state->timerfd, TFD_TIMER_ABSTIME, &ts, NULL);

spa/plugins/audiomixer/audiomixer.c

@@ -101,7 +101,6 @@ struct _SpaAudioMixer {
   uint8_t format_buffer[4096];
 
   bool started;
-  SpaResult state;
 };
 
 #define CHECK_PORT_NUM(this,d,p)     (((d) == SPA_DIRECTION_INPUT && (p) < MAX_PORTS) || \
@@ -573,51 +572,46 @@ spa_audiomixer_node_port_send_command (SpaNode        *node,
 }
 
 static inline void
-add_port_data (SpaAudioMixer *this, MixerBuffer *out, SpaAudioMixerPort *port, int layer)
+add_port_data (SpaAudioMixer *this, void *out, size_t outsize, SpaAudioMixerPort *port, int layer)
 {
-  int i;
+  void *in;
-  int16_t *op, *ip;
+  size_t insize;
-  size_t os, is, chunk;
   MixerBuffer *b;
-  SpaData *id, *od;
+  SpaData *id;
-
-  od = out->outbuf->datas;
-  op = SPA_MEMBER (od[0].data, od[0].chunk->offset, void);
-  os = od[0].chunk->size;
 
   b = spa_list_first (&port->queue, MixerBuffer, link);
 
   id = b->outbuf->datas;
-  ip = SPA_MEMBER (id[0].data, port->queued_offset + id[0].chunk->offset, void);
+  in = SPA_MEMBER (id[0].data, port->queued_offset + id[0].chunk->offset, void);
-  is = id[0].chunk->size - port->queued_offset;
+  insize = id[0].chunk->size - port->queued_offset;
+  outsize = SPA_MIN (outsize, insize);
 
-  chunk = SPA_MIN (os, is);
+  {
+    int i, size = outsize / 2;
+    int16_t *op = out, *ip = in;
 
     if (layer == 0) {
-    for (i = 0; i < chunk / 2; i++)
+      memcpy (op, ip, outsize);
-      op[i] = ip[i];
     }
     else {
-    for (i = 0; i < chunk / 2; i++)
+      for (i = 0; i < size; i++)
         op[i] = SPA_CLAMP (op[i] + ip[i], INT16_MIN, INT16_MAX);
     }
+  }
 
-  op += chunk / 2;
+  port->queued_offset += outsize;
-  os -= chunk;
+  port->queued_bytes -= outsize;
 
-  port->queued_offset += chunk;
+  if (outsize == insize) {
-  port->queued_bytes -= chunk;
-
-  if (chunk == is) {
     spa_log_trace (this->log, "audiomixer %p: return buffer %d on port %p %zd",
-              this, b->outbuf->id, port, chunk);
+              this, b->outbuf->id, port, outsize);
     port->io->buffer_id = b->outbuf->id;
     spa_list_remove (&b->link);
     b->outstanding = true;
     port->queued_offset = 0;
   } else {
     spa_log_trace (this->log, "audiomixer %p: keeping buffer %d on port %p %zd %zd",
-        this, b->outbuf->id, port, port->queued_bytes, chunk);
+        this, b->outbuf->id, port, port->queued_bytes, outsize);
   }
 }
 
@@ -646,8 +640,8 @@ mix_output (SpaAudioMixer *this, size_t n_bytes)
   od[0].chunk->size = n_bytes;
   od[0].chunk->stride = 0;
 
-  spa_log_trace (this->log, "audiomixer %p: dequeue output buffer %d %zd", this, outbuf->outbuf->id, n_bytes);
+  spa_log_trace (this->log, "audiomixer %p: dequeue output buffer %d %zd",
-
+                this, outbuf->outbuf->id, n_bytes);
   for (layer = 0, i = 0; i < MAX_PORTS; i++) {
     SpaAudioMixerPort *port = &this->in_ports[i];
 
@@ -660,7 +654,7 @@ mix_output (SpaAudioMixer *this, size_t n_bytes)
       port->queued_offset = 0;
       continue;
     }
-    add_port_data (this, outbuf, port, layer++);
+    add_port_data (this, od[0].data, n_bytes, port, layer++);
   }
   output->buffer_id = outbuf->outbuf->id;
   output->status = SPA_RESULT_HAVE_BUFFER;
@@ -685,7 +679,7 @@ spa_audiomixer_node_process_input (SpaNode *node)
   output = outport->io;
   spa_return_val_if_fail (output != NULL, SPA_RESULT_ERROR);
 
-  if (this->state == SPA_RESULT_HAVE_BUFFER)
+  if (output->status == SPA_RESULT_HAVE_BUFFER)
     return SPA_RESULT_HAVE_BUFFER;
 
   for (i = 0; i < MAX_PORTS; i++) {
@@ -701,12 +695,14 @@ spa_audiomixer_node_process_input (SpaNode *node)
       MixerBuffer *b = &port->buffers[input->buffer_id];
 
       if (!b->outstanding) {
-        spa_log_warn (this->log, "audiomixer %p: buffer was not outstanding %d", this, input->buffer_id);
+        spa_log_warn (this->log, "audiomixer %p: buffer %u in use", this, input->buffer_id);
+        input->status = SPA_RESULT_INVALID_BUFFER_ID;
         continue;
       }
 
       b->outstanding = false;
       input->buffer_id = SPA_ID_INVALID;
+      input->status = SPA_RESULT_OK;
 
       spa_list_insert (port->queue.prev, &b->link);
       port->queued_bytes += b->outbuf->datas[0].chunk->size;
@@ -719,11 +715,11 @@ spa_audiomixer_node_process_input (SpaNode *node)
   }
 
   if (min_queued != SIZE_MAX && min_queued > 0) {
-    this->state = mix_output (this, min_queued);
+    output->status = mix_output (this, min_queued);
   } else {
-    this->state = SPA_RESULT_NEED_BUFFER;
+    output->status = SPA_RESULT_NEED_BUFFER;
   }
-  return this->state;
+  return output->status;
 }
 
 static SpaResult
@@ -733,6 +729,7 @@ spa_audiomixer_node_process_output (SpaNode *node)
   SpaAudioMixerPort *port;
   SpaPortIO *output;
   int i;
+  size_t min_queued = SIZE_MAX;
 
   spa_return_val_if_fail (node != NULL, SPA_RESULT_INVALID_ARGUMENTS);
 
@@ -742,15 +739,15 @@ spa_audiomixer_node_process_output (SpaNode *node)
   output = port->io;
   spa_return_val_if_fail (output != NULL, SPA_RESULT_ERROR);
 
+  if (output->status == SPA_RESULT_HAVE_BUFFER)
+    return SPA_RESULT_HAVE_BUFFER;
+
   /* recycle */
   if (output->buffer_id != SPA_ID_INVALID) {
     recycle_buffer (this, output->buffer_id);
     output->buffer_id = SPA_ID_INVALID;
   }
   /* produce more output if possible */
-  if (this->state == SPA_RESULT_HAVE_BUFFER) {
-    size_t min_queued = SIZE_MAX;
-
   for (i = 0; i < MAX_PORTS; i++) {
     SpaAudioMixerPort *port = &this->in_ports[i];
 
@@ -761,13 +758,10 @@ spa_audiomixer_node_process_output (SpaNode *node)
       min_queued = port->queued_bytes;
   }
   if (min_queued != SIZE_MAX && min_queued > 0) {
-      this->state = mix_output (this, min_queued);
+    output->status = mix_output (this, min_queued);
-    } else {
-      this->state = SPA_RESULT_NEED_BUFFER;
-    }
   }
+  else {
     /* take requested output range and apply to input */
-  if (this->state == SPA_RESULT_NEED_BUFFER) {
     for (i = 0; i < MAX_PORTS; i++) {
       SpaAudioMixerPort *port = &this->in_ports[i];
       SpaPortIO *input;
@@ -786,7 +780,7 @@ spa_audiomixer_node_process_output (SpaNode *node)
           i, output->range.min_size, port->queued_bytes, input->status);
     }
   }
-  return this->state;
+  return output->status;
 }
 
 static const SpaNode audiomixer_node = {
@@ -872,7 +866,6 @@ spa_audiomixer_init (const SpaHandleFactory *factory,
   init_type (&this->type, this->map);
 
   this->node = audiomixer_node;
-  this->state = SPA_RESULT_NEED_BUFFER;
 
   this->out_ports[0].io = NULL;
   this->out_ports[0].info.flags = SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS |

spa/plugins/v4l2/v4l2-source.c

@@ -804,13 +804,16 @@ spa_v4l2_source_node_process_output (SpaNode *node)
   spa_return_val_if_fail (node != NULL, SPA_RESULT_INVALID_ARGUMENTS);
 
   this = SPA_CONTAINER_OF (node, SpaV4l2Source, node);
+  io = this->state[0].io;
+  spa_return_val_if_fail (io != NULL, SPA_RESULT_WRONG_STATE);
+
+  if (io->status == SPA_RESULT_HAVE_BUFFER)
+    return SPA_RESULT_HAVE_BUFFER;
 
-  if ((io = this->state[0].io)) {
   if (io->buffer_id != SPA_ID_INVALID) {
     res = spa_v4l2_buffer_recycle (this, io->buffer_id);
     io->buffer_id = SPA_ID_INVALID;
   }
-  }
   return res;
 }
 

spa/plugins/v4l2/v4l2-utils.c

@@ -933,7 +933,7 @@ mmap_read (SpaV4l2Source *this)
   V4l2Buffer *b;
   SpaData *d;
   int64_t pts;
-  SpaPortIO *io;
+  SpaPortIO *io = state->io;
 
   CLEAR(buf);
   buf.type = V4L2_BUF_TYPE_VIDEO_CAPTURE;
@@ -972,12 +972,12 @@ mmap_read (SpaV4l2Source *this)
   d[0].chunk->size = buf.bytesused;
   d[0].chunk->stride = state->fmt.fmt.pix.bytesperline;
 
-  if ((io = state->io)) {
+  {
     SpaEvent event = SPA_EVENT_INIT (this->type.event_node.HaveOutput);
 
     b->outstanding = true;
     io->buffer_id = b->outbuf->id;
-    io->status = SPA_RESULT_OK;
+    io->status = SPA_RESULT_HAVE_BUFFER;
 
     this->event_cb (&this->node, &event, this->user_data);
   }

spa/plugins/videotestsrc/videotestsrc.c

@@ -803,17 +803,23 @@ static SpaResult
 spa_videotestsrc_node_process_output (SpaNode *node)
 {
   SpaVideoTestSrc *this;
+  SpaPortIO *io;
 
   spa_return_val_if_fail (node != NULL, SPA_RESULT_INVALID_ARGUMENTS);
 
   this = SPA_CONTAINER_OF (node, SpaVideoTestSrc, node);
+  io = this->io;
+  spa_return_val_if_fail (io != NULL, SPA_RESULT_WRONG_STATE);
 
-  if (this->io && this->io->buffer_id != SPA_ID_INVALID) {
+  if (io->status == SPA_RESULT_HAVE_BUFFER)
+    return SPA_RESULT_HAVE_BUFFER;
+
+  if (io->buffer_id != SPA_ID_INVALID) {
     reuse_buffer (this, this->io->buffer_id);
     this->io->buffer_id = SPA_ID_INVALID;
   }
 
-  if (!this->async)
+  if (!this->async && (io->status == SPA_RESULT_NEED_BUFFER))
     return videotestsrc_make_buffer (this);
   else
     return SPA_RESULT_OK;

spa/tests/test-v4l2.c

@@ -163,11 +163,26 @@ make_node (AppData *data, SpaNode **node, const char *lib, const char *name)
   return SPA_RESULT_ERROR;
 }
 
+static void
+handle_events (AppData *data)
+{
+  SDL_Event event;
+  while (SDL_PollEvent (&event)) {
+    switch (event.type) {
+      case SDL_QUIT:
+        exit (0);
+        break;
+    }
+  }
+}
+
 static void
 on_source_event (SpaNode *node, SpaEvent *event, void *user_data)
 {
   AppData *data = user_data;
 
+  handle_events (data);
+
   if (SPA_EVENT_TYPE (event) == data->type.event_node.HaveOutput) {
     SpaResult res;
     SpaBuffer *b;
@@ -496,7 +511,7 @@ run_async_source (AppData *data)
     data->running = false;
   }
 
-  sleep (10);
+  sleep (10000);
 
   if (data->running) {
     data->running = false;