Improve process_output

process_output is now to finish processing of the output in async nodes,
which means we need to fill the io area before sending the HAVE_OUTPUT
event. This simplifies some things and improves performance because we
don't need to deal with queues and additional checks.

pinos/client/stream.c

@@ -748,9 +748,6 @@ stream_dispatch_func (void             *object,
               break;
           }
         }
-
-        spa_debug_buffer (b);
-
         pinos_signal_emit (&stream->add_buffer, stream, bid->id);
       }
 

pinos/gst/gstpinossrc.c

@@ -424,7 +424,7 @@ on_add_buffer (PinosListener *listener,
       case SPA_DATA_TYPE_DMABUF:
       {
         gmem = gst_fd_allocator_alloc (pinossrc->fd_allocator, dup (d->fd),
-                  d->maxsize, GST_FD_MEMORY_FLAG_NONE);
+                  d->mapoffset + d->maxsize, GST_FD_MEMORY_FLAG_NONE);
         gst_memory_resize (gmem, d->chunk->offset + d->mapoffset, d->chunk->size);
         data.offset = d->mapoffset;
         break;

spa/plugins/alsa/alsa-sink.c

@@ -698,7 +698,7 @@ spa_alsa_sink_node_process_input (SpaNode *node)
     this->ringbuffer->outstanding = true;
     this->ringbuffer = b;
   } else {
-    spa_list_insert (this->ready.prev, &b->list);
+    spa_list_insert (this->ready.prev, &b->link);
   }
   b->outstanding = false;
   input->status = SPA_RESULT_OK;

spa/plugins/alsa/alsa-source.c

@@ -408,7 +408,7 @@ recycle_buffer (SpaALSASource *this, uint32_t buffer_id)
     return;
 
   b->outstanding = false;
-  spa_list_insert (this->free.prev, &b->list);
+  spa_list_insert (this->free.prev, &b->link);
 }
 
 static SpaResult
@@ -590,7 +590,7 @@ spa_alsa_source_node_port_use_buffers (SpaNode         *node,
       default:
         break;
     }
-    spa_list_insert (this->free.prev, &b->list);
+    spa_list_insert (this->free.prev, &b->link);
   }
   this->n_buffers = n_buffers;
 
@@ -731,35 +731,6 @@ spa_alsa_source_node_process_input (SpaNode *node)
 static SpaResult
 spa_alsa_source_node_process_output (SpaNode *node)
 {
-  SpaALSASource *this;
-  SpaPortOutput *output;
-  SpaALSABuffer *b;
-
-  if (node == NULL)
-    return SPA_RESULT_INVALID_ARGUMENTS;
-
-  this = SPA_CONTAINER_OF (node, SpaALSASource, node);
-
-  output = this->io;
-
-  if (!this->have_format) {
-    output->status = SPA_RESULT_NO_FORMAT;
-    return SPA_RESULT_ERROR;
-  }
-
-  if (spa_list_is_empty (&this->ready)) {
-    output->status = SPA_RESULT_UNEXPECTED;
-    return SPA_RESULT_ERROR;
-  }
-  b = spa_list_first (&this->ready, SpaALSABuffer, list);
-  spa_list_remove (&b->list);
-  b->outstanding = true;
-
-  output->buffer_id = b->outbuf->id;
-  output->status = SPA_RESULT_OK;
-
-  spa_log_trace (this->log, "pull buffer %u", b->outbuf->id);
-
   return SPA_RESULT_OK;
 }
 

spa/plugins/alsa/alsa-utils.c

@@ -282,7 +282,7 @@ pull_frames_queue (SpaALSAState *state,
     off_t offs;
     SpaALSABuffer *b;
 
-    b = spa_list_first (&state->ready, SpaALSABuffer, list);
+    b = spa_list_first (&state->ready, SpaALSABuffer, link);
 
     offs = SPA_MIN (b->outbuf->datas[0].chunk->offset, b->outbuf->datas[0].maxsize);
     src = SPA_MEMBER (b->outbuf->datas[0].data, offs, uint8_t);
@@ -299,7 +299,7 @@ pull_frames_queue (SpaALSAState *state,
     if (state->ready_offset >= size) {
       SpaNodeEventReuseBuffer rb;
 
-      spa_list_remove (&b->list);
+      spa_list_remove (&b->link);
       b->outstanding = true;
 
       rb.event.type = SPA_NODE_EVENT_TYPE_REUSE_BUFFER;
@@ -452,8 +452,8 @@ mmap_read (SpaALSAState *state)
     b = NULL;
     spa_log_warn (state->log, "no more buffers");
   } else {
-    b = spa_list_first (&state->free, SpaALSABuffer, list);
+    b = spa_list_first (&state->free, SpaALSABuffer, link);
-    spa_list_remove (&b->list);
+    spa_list_remove (&b->link);
 
     dest = b->outbuf->datas[0].data;
     destsize = b->outbuf->datas[0].maxsize;
@@ -495,14 +495,18 @@ mmap_read (SpaALSAState *state)
   if (b) {
     SpaNodeEventHaveOutput ho;
     SpaData *d;
+    SpaPortOutput *output;
 
     d = b->outbuf->datas;
     d[0].chunk->offset = 0;
     d[0].chunk->size = avail * state->frame_size;
     d[0].chunk->stride = 0;
 
-    spa_list_insert (state->ready.prev, &b->list);
+    if ((output = state->io)) {
-
+      b->outstanding = true;
+      output->buffer_id = b->outbuf->id;
+      output->status = SPA_RESULT_OK;
+    }
     ho.event.type = SPA_NODE_EVENT_TYPE_HAVE_OUTPUT;
     ho.event.size = sizeof (ho);
     ho.port_id = 0;

spa/plugins/alsa/alsa-utils.h

@@ -56,7 +56,7 @@ struct _SpaALSABuffer {
   SpaMetaHeader *h;
   SpaMetaRingbuffer *rb;
   bool outstanding;
-  SpaList list;
+  SpaList link;
 };
 
 typedef struct {

spa/plugins/audiotestsrc/audiotestsrc.c

@@ -55,7 +55,7 @@ struct _ATSBuffer {
   SpaMetaHeader *h;
   void *ptr;
   size_t size;
-  SpaList list;
+  SpaList link;
 };
 
 typedef struct {
@@ -103,7 +103,6 @@ struct _SpaAudioTestSrc {
 
   uint64_t sample_count;
   SpaList empty;
-  SpaList ready;
 };
 
 #define CHECK_PORT(this,d,p)  ((d) == SPA_DIRECTION_OUTPUT && (p) == 0)
@@ -264,14 +263,15 @@ audiotestsrc_on_output (SpaSource *source)
 {
   SpaAudioTestSrc *this = source->data;
   ATSBuffer *b;
+  SpaPortOutput *output;
 
   if (spa_list_is_empty (&this->empty)) {
     if (!this->props[1].live)
       update_loop_enabled (this, false);
     return;
   }
-  b = spa_list_first (&this->empty, ATSBuffer, list);
+  b = spa_list_first (&this->empty, ATSBuffer, link);
-  spa_list_remove (&b->list);
+  spa_list_remove (&b->link);
 
   fill_buffer (this, b);
 
@@ -296,9 +296,13 @@ audiotestsrc_on_output (SpaSource *source)
     timerfd_settime (this->timer_source.fd, TFD_TIMER_ABSTIME, &this->timerspec, NULL);
   }
 
-  spa_list_insert (this->ready.prev, &b->list);
+  if ((output = this->output)) {
+    b->outstanding = true;
+    output->buffer_id = b->outbuf->id;
+    output->status = SPA_RESULT_OK;
     send_have_output (this);
   }
+}
 
 static void
 update_state (SpaAudioTestSrc *this, SpaNodeState state)
@@ -526,7 +530,6 @@ clear_buffers (SpaAudioTestSrc *this)
     spa_log_info (this->log, "audiotestsrc %p: clear buffers", this);
     this->n_buffers = 0;
     spa_list_init (&this->empty);
-    spa_list_init (&this->ready);
   }
   return SPA_RESULT_OK;
 }
@@ -696,7 +699,7 @@ spa_audiotestsrc_node_port_use_buffers (SpaNode         *node,
       default:
         break;
     }
-    spa_list_insert (this->empty.prev, &b->list);
+    spa_list_insert (this->empty.prev, &b->link);
   }
   this->n_buffers = n_buffers;
 
@@ -793,7 +796,7 @@ spa_audiotestsrc_node_port_reuse_buffer (SpaNode         *node,
     return SPA_RESULT_OK;
 
   b->outstanding = false;
-  spa_list_insert (this->empty.prev, &b->list);
+  spa_list_insert (this->empty.prev, &b->link);
 
   if (!this->props[1].live)
     update_loop_enabled (this, true);
@@ -819,28 +822,6 @@ spa_audiotestsrc_node_process_input (SpaNode *node)
 static SpaResult
 spa_audiotestsrc_node_process_output (SpaNode *node)
 {
-  SpaAudioTestSrc *this;
-  SpaPortOutput *output;
-  ATSBuffer *b;
-
-  if (node == NULL)
-    return SPA_RESULT_INVALID_ARGUMENTS;
-
-  this = SPA_CONTAINER_OF (node, SpaAudioTestSrc, node);
-  if ((output = this->output) == NULL)
-    return SPA_RESULT_OK;
-
-  if (spa_list_is_empty (&this->ready)) {
-    output->status = SPA_RESULT_UNEXPECTED;
-    return SPA_RESULT_ERROR;
-  }
-  b = spa_list_first (&this->ready, ATSBuffer, list);
-  spa_list_remove (&b->list);
-  b->outstanding = true;
-
-  output->buffer_id = b->outbuf->id;
-  output->status = SPA_RESULT_OK;
-
   return SPA_RESULT_OK;
 }
 
@@ -1002,7 +983,6 @@ audiotestsrc_init (const SpaHandleFactory  *factory,
   reset_audiotestsrc_props (&this->props[1]);
 
   spa_list_init (&this->empty);
-  spa_list_init (&this->ready);
 
   this->timer_source.func = audiotestsrc_on_output;
   this->timer_source.data = this;

spa/plugins/v4l2/v4l2-source.c

@@ -61,8 +61,6 @@ struct _V4l2Buffer {
   bool outstanding;
   bool allocated;
   struct v4l2_buffer v4l2_buffer;
-//  V4l2Buffer *next;
-  SpaList list;
 };
 
 typedef struct _V4l2Format V4l2Format;
@@ -114,7 +112,6 @@ typedef struct {
 
   V4l2Buffer   buffers[MAX_BUFFERS];
   unsigned int n_buffers;
-  SpaList     ready;
 
   bool source_enabled;
   SpaSource source;
@@ -841,42 +838,6 @@ spa_v4l2_source_node_process_input (SpaNode *node)
 static SpaResult
 spa_v4l2_source_node_process_output (SpaNode *node)
 {
-  SpaV4l2Source *this;
-  SpaV4l2State *state;
-  SpaPortOutput *output;
-  V4l2Buffer *b;
-
-  if (node == NULL)
-    return SPA_RESULT_INVALID_ARGUMENTS;
-
-  this = SPA_CONTAINER_OF (node, SpaV4l2Source, node);
-
-  state = &this->state[0];
-
-  if ((output = state->io) == NULL)
-    return SPA_RESULT_OK;
-
-  if (state->current_format == NULL) {
-    output->status = SPA_RESULT_NO_FORMAT;
-    return SPA_RESULT_ERROR;
-  }
-
-  if (spa_list_is_empty (&state->ready)) {
-    output->status = SPA_RESULT_UNEXPECTED;
-    return SPA_RESULT_ERROR;
-  }
-
-  b = spa_list_first (&state->ready, V4l2Buffer, list);
-  if (b == NULL) {
-    output->status = SPA_RESULT_UNEXPECTED;
-    return SPA_RESULT_ERROR;
-  }
-  spa_list_remove (&b->list);
-  b->outstanding = true;
-
-  output->buffer_id = b->outbuf->id;
-  output->status = SPA_RESULT_OK;
-
   return SPA_RESULT_OK;
 }
 
@@ -1034,9 +995,6 @@ v4l2_source_init (const SpaHandleFactory  *factory,
   this->props[1].props.prop_info = prop_info;
   reset_v4l2_source_props (&this->props[1]);
 
-  spa_list_init (&this->state[0].ready);
-//  SPA_QUEUE_INIT (&this->state[0].ready);
-
   this->state[0].log = this->log;
   this->state[0].info.flags = SPA_PORT_INFO_FLAG_LIVE;
 

spa/plugins/v4l2/v4l2-utils.c

@@ -842,6 +842,7 @@ mmap_read (SpaV4l2Source *this)
   V4l2Buffer *b;
   SpaData *d;
   int64_t pts;
+  SpaPortOutput *output;
 
   CLEAR(buf);
   buf.type = V4L2_BUF_TYPE_VIDEO_CAPTURE;
@@ -880,8 +881,11 @@ mmap_read (SpaV4l2Source *this)
   d[0].chunk->size = buf.bytesused;
   d[0].chunk->stride = state->fmt.fmt.pix.bytesperline;
 
-  spa_list_insert (state->ready.prev, &b->list);
+  if ((output = state->io)) {
-
+    b->outstanding = true;
+    output->buffer_id = b->outbuf->id;
+    output->status = SPA_RESULT_OK;
+  }
   return SPA_RESULT_OK;
 }
 

spa/plugins/videotestsrc/videotestsrc.c

@@ -58,7 +58,7 @@ struct _VTSBuffer {
   SpaMetaHeader *h;
   void *ptr;
   size_t stride;
-  SpaList list;
+  SpaList link;
 };
 
 struct _SpaVideoTestSrc {
@@ -100,7 +100,6 @@ struct _SpaVideoTestSrc {
 
   uint64_t frame_count;
   SpaList empty;
-  SpaList ready;
 };
 
 #define CHECK_PORT(this,d,p)  ((d) == SPA_DIRECTION_OUTPUT && (p) == 0)
@@ -222,14 +221,15 @@ videotestsrc_on_output (SpaSource *source)
 {
   SpaVideoTestSrc *this = source->data;
   VTSBuffer *b;
+  SpaPortOutput *output;
 
   if (spa_list_is_empty (&this->empty)) {
     if (!this->props[1].live)
       update_loop_enabled (this, false);
     return;
   }
-  b = spa_list_first (&this->empty, VTSBuffer, list);
+  b = spa_list_first (&this->empty, VTSBuffer, link);
-  spa_list_remove (&b->list);
+  spa_list_remove (&b->link);
 
   fill_buffer (this, b);
 
@@ -254,9 +254,13 @@ videotestsrc_on_output (SpaSource *source)
     timerfd_settime (this->timer_source.fd, TFD_TIMER_ABSTIME, &this->timerspec, NULL);
   }
 
-  spa_list_insert (this->ready.prev, &b->list);
+  if ((output = this->output)) {
+    b->outstanding = true;
+    output->buffer_id = b->outbuf->id;
+    output->status = SPA_RESULT_OK;
     send_have_output (this);
   }
+}
 
 static void
 update_state (SpaVideoTestSrc *this, SpaNodeState state)
@@ -481,7 +485,6 @@ clear_buffers (SpaVideoTestSrc *this)
     spa_log_info (this->log, "videotestsrc %p: clear buffers", this);
     this->n_buffers = 0;
     spa_list_init (&this->empty);
-    spa_list_init (&this->ready);
   }
   return SPA_RESULT_OK;
 }
@@ -663,7 +666,7 @@ spa_videotestsrc_node_port_use_buffers (SpaNode         *node,
       default:
         break;
     }
-    spa_list_insert (this->empty.prev, &b->list);
+    spa_list_insert (this->empty.prev, &b->link);
   }
   this->n_buffers = n_buffers;
 
@@ -759,7 +762,7 @@ spa_videotestsrc_node_port_reuse_buffer (SpaNode         *node,
     return SPA_RESULT_OK;
 
   b->outstanding = false;
-  spa_list_insert (this->empty.prev, &b->list);
+  spa_list_insert (this->empty.prev, &b->link);
 
   if (!this->props[1].live)
     update_loop_enabled (this, true);
@@ -785,33 +788,6 @@ spa_videotestsrc_node_process_input (SpaNode *node)
 static SpaResult
 spa_videotestsrc_node_process_output (SpaNode *node)
 {
-  SpaVideoTestSrc *this;
-  VTSBuffer *b;
-  SpaPortOutput *output;
-
-  if (node == NULL)
-    return SPA_RESULT_INVALID_ARGUMENTS;
-
-  this = SPA_CONTAINER_OF (node, SpaVideoTestSrc, node);
-
-  if ((output = this->output) == NULL)
-    return SPA_RESULT_OK;
-
-  if (!this->have_format) {
-    output->status = SPA_RESULT_NO_FORMAT;
-    return SPA_RESULT_ERROR;
-  }
-
-  if (spa_list_is_empty (&this->ready)) {
-    output->status = SPA_RESULT_UNEXPECTED;
-    return SPA_RESULT_ERROR;
-  }
-  b = spa_list_first (&this->ready, VTSBuffer, list);
-  b->outstanding = true;
-
-  output->buffer_id = b->outbuf->id;
-  output->status = SPA_RESULT_OK;
-
   return SPA_RESULT_OK;
 }
 
@@ -969,7 +945,6 @@ videotestsrc_init (const SpaHandleFactory  *factory,
   reset_videotestsrc_props (&this->props[1]);
 
   spa_list_init (&this->empty);
-  spa_list_init (&this->ready);
 
   this->timer_source.func = videotestsrc_on_output;
   this->timer_source.data = this;