pinossrc: handle latency and liveness

Use NULL filter when the port format enumeration returns nothing.
Send clock update even when there is no clock.
Don't send data to a port when the node is not streaming
Add latency support to the clock update
Copy the unset_mask when copying the formats

pinos/client/stream.c

@@ -664,8 +664,9 @@ send_reuse_buffer (PinosStream *stream, uint32_t port_id, uint32_t buffer_id)
   SpaControlCmdNodeEvent cne;
   SpaNodeEvent ne;
   SpaNodeEventReuseBuffer rb;
+  guint8 buffer[128];
 
-  control_builder_init (stream, &builder);
+  spa_control_builder_init_into (&builder, buffer, sizeof (buffer), NULL, 0);
   cne.event = ≠
   ne.type = SPA_NODE_EVENT_TYPE_REUSE_BUFFER;
   ne.data = &rb;
@@ -791,7 +792,7 @@ handle_node_command (PinosStream    *stream,
       SpaControlBuilder builder;
       SpaControl control;
 
-      g_debug ("stream %p: stop", stream);
+      g_debug ("stream %p: pause", stream);
 
       control_builder_init (stream, &builder);
       add_state_change (stream, &builder, SPA_NODE_STATE_PAUSED);
@@ -834,6 +835,12 @@ handle_node_command (PinosStream    *stream,
     case SPA_NODE_COMMAND_CLOCK_UPDATE:
     {
       SpaNodeCommandClockUpdate *cu = command->data;
+      if (cu->flags & SPA_NODE_COMMAND_CLOCK_UPDATE_FLAG_LIVE) {
+        pinos_properties_set (priv->properties,
+                          "pinos.latency.is-live", "1");
+        pinos_properties_setf (priv->properties,
+                          "pinos.latency.min", "%"PRId64, cu->latency);
+      }
       priv->last_ticks = cu->ticks;
       priv->last_rate = cu->rate;
       priv->last_monotonic = cu->monotonic_time;

pinos/gst/gstpinossrc.c

@@ -270,7 +270,6 @@ gst_pinos_src_init (GstPinosSrc * src)
   gst_base_src_set_format (GST_BASE_SRC (src), GST_FORMAT_TIME);
 
   GST_OBJECT_FLAG_SET (src, GST_ELEMENT_FLAG_PROVIDE_CLOCK);
-  gst_base_src_set_live (GST_BASE_SRC (src), TRUE);
 
   g_queue_init (&src->queue);
 
@@ -347,16 +346,16 @@ static gboolean
 buffer_recycle (GstMiniObject *obj)
 {
   ProcessMemData *data;
-
+  GstPinosSrc *src;
-  GST_LOG_OBJECT (obj, "recycle buffer");
 
   gst_mini_object_ref (obj);
   data = gst_mini_object_get_qdata (obj,
                                     process_mem_data_quark);
   GST_BUFFER_FLAGS (obj) = data->flags;
+  src = data->src;
 
-  pinos_stream_recycle_buffer (data->src->stream, data->id);
   GST_LOG_OBJECT (obj, "recycle buffer");
+  pinos_stream_recycle_buffer (src->stream, data->id);
 
   return FALSE;
 }
@@ -524,10 +523,6 @@ parse_stream_properties (GstPinosSrc *pinossrc, PinosProperties *props)
 
   var = pinos_properties_get (props, "pinos.latency.max");
   pinossrc->max_latency = var ? (GstClockTime) atoi (var) : GST_CLOCK_TIME_NONE;
-
-  pinossrc->is_live = TRUE;
-  pinossrc->min_latency = 100000000;
-  pinossrc->max_latency = GST_CLOCK_TIME_NONE;
 }
 
 static gboolean

pinos/modules/spa/spa-audiotestsrc.c

@@ -86,6 +86,7 @@ make_node (SpaNode **node, const char *lib, const char *name)
 static void
 setup_node (PinosSpaAudioTestSrc *this)
 {
+#if 0
   PinosNode *node = PINOS_NODE (this);
   SpaResult res;
   SpaProps *props;
@@ -94,12 +95,9 @@ setup_node (PinosSpaAudioTestSrc *this)
   if ((res = spa_node_get_props (node->node, &props)) < 0)
     g_debug ("got get_props error %d", res);
 
-  value.value = "hw:1";
-  value.size = strlen (value.value)+1;
-  spa_props_set_prop (props, spa_props_index_for_name (props, "device"), &value);
-
   if ((res = spa_node_set_props (node->node, props)) < 0)
     g_debug ("got set_props error %d", res);
+#endif
 }
 
 static void

pinos/server/daemon.c

@@ -216,7 +216,7 @@ on_port_added (PinosNode *node, PinosDirection direction, guint port_id, PinosCl
 
     new_port = pinos_node_get_free_port (target, pinos_direction_reverse (direction));
     if (new_port == SPA_ID_INVALID) {
-      g_debug ("daemon %p: can't create free port", this);
+      g_warning ("daemon %p: can't get free port", this);
       return;
     }
     if (direction == PINOS_DIRECTION_OUTPUT)
@@ -224,6 +224,11 @@ on_port_added (PinosNode *node, PinosDirection direction, guint port_id, PinosCl
     else
       link = pinos_node_link (target, new_port, node, port_id, NULL, NULL);
 
+    if (link == NULL) {
+      g_warning ("daemon %p: can't link nodes", this);
+      return;
+    }
+
     pinos_client_add_object (client, G_OBJECT (link));
     g_object_unref (link);
   }

pinos/server/link.c

@@ -219,7 +219,7 @@ static SpaResult
 do_negotiate (PinosLink *this, SpaNodeState in_state, SpaNodeState out_state)
 {
   SpaResult res;
-  SpaFormat *filter, *format;
+  SpaFormat *filter = NULL, *format;
   void *istate = NULL, *ostate = NULL;
 
   /* both ports need a format */
@@ -231,8 +231,10 @@ again:
                                            &filter,
                                            NULL,
                                            &istate)) < 0) {
-      g_warning ("error input enum formats: %d", res);
+      if (res == SPA_RESULT_ENUM_END && istate != NULL) {
-      goto error;
+        g_warning ("error input enum formats: %d", res);
+        goto error;
+      }
     }
     spa_debug_format (filter);
 
@@ -336,6 +338,7 @@ do_allocation (PinosLink *this, SpaNodeState in_state, SpaNodeState out_state)
       }
       memcpy (priv->out_buffers, priv->in_buffers, priv->n_in_buffers * sizeof (SpaBuffer*));
       priv->n_out_buffers = priv->n_in_buffers;
+      g_debug ("allocated out_buffers %p, in_buffers %p", priv->out_buffers, priv->in_buffers);
     } else if ((out_flags & SPA_PORT_INFO_FLAG_CAN_ALLOC_BUFFERS) &&
         (in_flags & SPA_PORT_INFO_FLAG_CAN_ALLOC_BUFFERS)) {
       out_flags = SPA_PORT_INFO_FLAG_CAN_ALLOC_BUFFERS;
@@ -347,8 +350,10 @@ do_allocation (PinosLink *this, SpaNodeState in_state, SpaNodeState out_state)
     }
   } else if (in_state == SPA_NODE_STATE_READY && out_state > SPA_NODE_STATE_READY) {
     out_flags &= ~SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS;
+    in_flags &= ~SPA_PORT_INFO_FLAG_CAN_ALLOC_BUFFERS;
   } else if (out_state == SPA_NODE_STATE_READY && in_state > SPA_NODE_STATE_READY) {
     in_flags &= ~SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS;
+    out_flags &= ~SPA_PORT_INFO_FLAG_CAN_ALLOC_BUFFERS;
   } else
     return SPA_RESULT_OK;
 
@@ -363,6 +368,7 @@ do_allocation (PinosLink *this, SpaNodeState in_state, SpaNodeState out_state)
       g_warning ("error alloc buffers: %d", res);
       goto error;
     }
+    g_debug ("allocated in_buffers %p", priv->in_buffers);
   }
   else if (out_flags & SPA_PORT_INFO_FLAG_CAN_ALLOC_BUFFERS) {
     priv->n_out_buffers = 16;
@@ -372,8 +378,10 @@ do_allocation (PinosLink *this, SpaNodeState in_state, SpaNodeState out_state)
       g_warning ("error alloc buffers: %d", res);
       goto error;
     }
+    g_debug ("allocated out_buffers %p", priv->out_buffers);
   }
   if (in_flags & SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS) {
+    g_debug ("using out_buffers %p", priv->out_buffers);
     if ((res = spa_node_port_use_buffers (this->input_node->node, this->input_port,
                                           priv->out_buffers, priv->n_out_buffers)) < 0) {
       g_warning ("error use buffers: %d", res);
@@ -381,12 +389,14 @@ do_allocation (PinosLink *this, SpaNodeState in_state, SpaNodeState out_state)
     }
   }
   else if (out_flags & SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS) {
+    g_debug ("using in_buffers %p", priv->out_buffers);
     if ((res = spa_node_port_use_buffers (this->output_node->node, this->output_port,
                                           priv->in_buffers, priv->n_in_buffers)) < 0) {
       g_warning ("error use buffers: %d", res);
       goto error;
     }
-  }
+  } else
+    return SPA_RESULT_NO_BUFFERS;
 
   return SPA_RESULT_OK;
 

pinos/server/node.c

@@ -320,23 +320,32 @@ static void
 send_clock_update (PinosNode *this)
 {
   PinosNodePrivate *priv = this->priv;
+  SpaNodeCommand cmd;
+  SpaNodeCommandClockUpdate cu;
+  SpaResult res;
 
+  cmd.type = SPA_NODE_COMMAND_CLOCK_UPDATE;
+  cmd.data = &cu;
+  cmd.size = sizeof (cu);
+
+  cu.flags = 0;
+  cu.change_mask = SPA_NODE_COMMAND_CLOCK_UPDATE_TIME |
+                   SPA_NODE_COMMAND_CLOCK_UPDATE_SCALE |
+                   SPA_NODE_COMMAND_CLOCK_UPDATE_STATE |
+                   SPA_NODE_COMMAND_CLOCK_UPDATE_LATENCY;
   if (priv->clock) {
-    SpaNodeCommand cmd;
+    cu.flags = SPA_NODE_COMMAND_CLOCK_UPDATE_FLAG_LIVE;
-    SpaNodeCommandClockUpdate cu;
-    SpaResult res;
-
-    cmd.type = SPA_NODE_COMMAND_CLOCK_UPDATE;
-    cmd.data = &cu;
-    cmd.size = sizeof (cu);
-    cu.change_mask = SPA_NODE_COMMAND_CLOCK_UPDATE_TIME;
     res = spa_clock_get_time (priv->clock, &cu.rate, &cu.ticks, &cu.monotonic_time);
-    cu.scale = (1 << 16) | 1;
+  } else {
-    cu.state = SPA_CLOCK_STATE_RUNNING;
+    cu.rate = 1;
-
+    cu.ticks = 0;
-    if ((res = spa_node_send_command (this->node, &cmd)) < 0)
+    cu.monotonic_time = 0;
-      g_debug ("got error %d", res);
   }
+  cu.scale = (1 << 16) | 1;
+  cu.state = SPA_CLOCK_STATE_RUNNING;
+
+  if ((res = spa_node_send_command (this->node, &cmd)) < 0)
+    g_debug ("got error %d", res);
 }
 
 static gboolean
@@ -510,6 +519,9 @@ on_node_event (SpaNode *node, SpaNodeEvent *event, void *user_data)
         if (pl == NULL || pl->output_node->node != node || pl->output_port != oinfo[0].port_id)
           continue;
 
+        if (pl->input_node->node->state != SPA_NODE_STATE_STREAMING)
+          continue;
+
         iinfo[0].port_id = pl->input_port;
         iinfo[0].buffer_id = oinfo[0].buffer_id;
         iinfo[0].flags = SPA_PORT_INPUT_FLAG_NONE;
@@ -1229,19 +1241,22 @@ pinos_node_link (PinosNode       *output_node,
   } else {
     uint32_t input_port, output_port;
 
+    output_port = get_free_node_port (output_node, PINOS_DIRECTION_OUTPUT);
+    if (output_port == SPA_ID_INVALID && output_node->priv->n_output_ports > 0)
+      output_port = output_node->priv->output_port_ids[0];
+    else
+      return NULL;
+
+    input_port = get_free_node_port (input_node, PINOS_DIRECTION_INPUT);
+    if (input_port == SPA_ID_INVALID && input_node->priv->n_input_ports > 0)
+      input_port = input_node->priv->input_port_ids[0];
+    else
+      return NULL;
+
     if (output_node->priv->clock)
       input_node->priv->clock = output_node->priv->clock;
 
     g_debug ("node %p: clock %p", output_node, output_node->priv->clock);
-
-    output_port = get_free_node_port (output_node, PINOS_DIRECTION_OUTPUT);
-    if (output_port == SPA_ID_INVALID)
-      output_port = output_node->priv->output_port_ids[0];
-
-    input_port = get_free_node_port (input_node, PINOS_DIRECTION_INPUT);
-    if (input_port == SPA_ID_INVALID)
-      input_port = input_node->priv->input_port_ids[0];
-
     pl = g_object_new (PINOS_TYPE_LINK,
                        "daemon", priv->daemon,
                        "output-node", output_node,

spa/include/spa/node-command.h

@@ -61,6 +61,7 @@ typedef struct {
 #define SPA_NODE_COMMAND_CLOCK_UPDATE_TIME        (1 << 0)
 #define SPA_NODE_COMMAND_CLOCK_UPDATE_SCALE       (1 << 1)
 #define SPA_NODE_COMMAND_CLOCK_UPDATE_STATE       (1 << 2)
+#define SPA_NODE_COMMAND_CLOCK_UPDATE_LATENCY     (1 << 3)
   uint32_t      change_mask;
   int32_t       rate;
   int64_t       ticks;
@@ -68,6 +69,9 @@ typedef struct {
   int64_t       offset;
   int32_t       scale;
   SpaClockState state;
+#define SPA_NODE_COMMAND_CLOCK_UPDATE_FLAG_LIVE   (1 << 0)
+  uint32_t      flags;
+  int64_t       latency;
 } SpaNodeCommandClockUpdate;
 
 #ifdef __cplusplus

spa/lib/audio-raw.c

@@ -25,15 +25,6 @@
 #include <spa/audio/raw.h>
 #include <spa/audio/format.h>
 
-static const SpaAudioInfoRaw default_raw_info = {
-  SPA_AUDIO_FORMAT_S16,
-  SPA_AUDIO_FLAG_NONE,
-  SPA_AUDIO_LAYOUT_INTERLEAVED,
-  44100,
-  2,
-  0
-};
-
 static const uint32_t format_values[] = {
   SPA_AUDIO_FORMAT_S8,
   SPA_AUDIO_FORMAT_U8,
@@ -228,6 +219,14 @@ spa_format_audio_init (SpaMediaType     type,
         { SPA_PROP_ID_AUDIO_CHANNELS,     offsetof (SpaFormatAudio, info.raw.channels), },
         { SPA_PROP_ID_AUDIO_CHANNEL_MASK, offsetof (SpaFormatAudio, info.raw.channel_mask), },
       };
+      static const SpaAudioInfoRaw default_raw_info = {
+        SPA_AUDIO_FORMAT_S16,
+        SPA_AUDIO_FLAG_NONE,
+        SPA_AUDIO_LAYOUT_INTERLEAVED,
+        44100,
+        2,
+        0
+      };
       prop_info = raw_format_prop_info;
       n_prop_info = SPA_N_ELEMENTS (raw_format_prop_info);
       format->format.props.unset_mask = (1 << 1) | (1 << 3) | (1 << 4) | (1 << 5);
@@ -295,7 +294,8 @@ spa_format_audio_parse (const SpaFormat *format,
   if (props->prop_info[idx].type != SPA_PROP_TYPE_POINTER)
     goto fallback;
 
-  memcpy (&aformat->info, value.value, SPA_MIN (value.size, sizeof (SpaAudioInfoRaw)));
+  memcpy (&aformat->info, value.value, SPA_MIN (value.size, sizeof (aformat->info)));
+  aformat->format.props.unset_mask = props->unset_mask;
 
   return SPA_RESULT_OK;
 

spa/lib/control.c

@@ -1366,7 +1366,7 @@ spa_control_read (SpaControl   *control,
       break;
     }
     if (len != hdr->length)
-      return SPA_RESULT_ERROR;
+      goto wrong_length;
   }
 
   /* handle control messages */
@@ -1389,6 +1389,11 @@ recv_error:
     fprintf (stderr, "could not recvmsg: %s\n", strerror (errno));
     return SPA_RESULT_ERROR;
   }
+wrong_length:
+  {
+    fprintf (stderr, "wrong header length %zd != %u\n", len, hdr->length);
+    return SPA_RESULT_ERROR;
+  }
 }
 
 

spa/lib/memory.c

@@ -31,6 +31,14 @@
 #include "spa/memory.h"
 #include "memfd-wrappers.h"
 
+#undef USE_MEMFD
+
+#if 0
+#define SPA_DEBUG_MEMORY(format,args...) fprintf(stderr,format,##args)
+#else
+#define SPA_DEBUG_MEMORY(format,args...)
+#endif
+
 #define MAX_POOLS       16
 #define MAX_MEMORIES    1024
 
@@ -121,6 +129,8 @@ spa_memory_alloc (uint32_t pool_id)
   mem->mem.pool_id = pool_id;
   mem->mem.id = id;
 
+  SPA_DEBUG_MEMORY ("mem %p: alloc\n", mem);
+
   return mem;
 }
 
@@ -149,7 +159,9 @@ spa_memory_alloc_with_fd  (uint32_t pool_id, void *data, size_t size)
   if (!(mem = spa_memory_alloc (pool_id)))
     return NULL;
 
-#if 1
+#ifdef USE_MEMFD
+  mem->fd = memfd_create ("spa-memfd", MFD_CLOEXEC | MFD_ALLOW_SEALING);
+#else
   {
     char filename[] = "/dev/shm/spa-tmpfile.XXXXXX";
     mem->fd = mkostemp (filename, O_CLOEXEC);
@@ -159,8 +171,6 @@ spa_memory_alloc_with_fd  (uint32_t pool_id, void *data, size_t size)
     }
     unlink (filename);
   }
-#else
-  mem->fd = memfd_create ("spa-memfd", MFD_CLOEXEC | MFD_ALLOW_SEALING);
 #endif
 
   if (data) {
@@ -176,7 +186,7 @@ spa_memory_alloc_with_fd  (uint32_t pool_id, void *data, size_t size)
       return NULL;
     }
   }
-#if 0
+#ifdef USE_MEMFD
   {
     unsigned int seals;
 
@@ -231,6 +241,7 @@ spa_memory_import (SpaMemoryRef *ref)
   } else {
     mem->refcount++;
   }
+  SPA_DEBUG_MEMORY ("mem %p: import %u:%u\n", mem, pool_id, id);
 
   return mem;
 }
@@ -253,6 +264,8 @@ spa_memory_free (SpaMemory *mem)
 {
   SpaMemoryPool *pool;
 
+  SPA_DEBUG_MEMORY ("mem %p: free\n", mem);
+
   if (mem->fd != -1) {
     if (mem->ptr)
       munmap (mem->ptr, mem->size);

spa/lib/video-raw.c

@@ -625,7 +625,8 @@ spa_format_video_parse (const SpaFormat *format,
   if (props->prop_info[idx].type != SPA_PROP_TYPE_POINTER)
     goto fallback;
 
-  memcpy (&vformat->info, value.value, SPA_MIN (value.size, sizeof (SpaVideoInfoRaw)));
+  memcpy (&vformat->info, value.value, SPA_MIN (value.size, sizeof (vformat->info)));
+  vformat->format.props.unset_mask = props->unset_mask;
 
   return SPA_RESULT_OK;
 

spa/plugins/audiotestsrc/audiotestsrc.c

@@ -67,11 +67,9 @@ struct _SpaAudioTestSrc {
   SpaFormatAudio query_format;
   SpaFormatAudio current_format;
 
+  bool have_buffers;
   SpaMemory *alloc_mem;
   ATSBuffer *alloc_buffers;
-
-  bool have_buffers;
-  SpaBuffer **buffers;
   unsigned int n_buffers;
 
   ATSBuffer *empty;
@@ -406,6 +404,20 @@ spa_audiotestsrc_node_port_enum_formats (SpaNode          *node,
   return SPA_RESULT_OK;
 }
 
+static SpaResult
+clear_buffers (SpaAudioTestSrc *this)
+{
+  if (this->have_buffers) {
+    fprintf (stderr, "clear buffers");
+    if (this->alloc_mem)
+      spa_memory_unref (&this->alloc_mem->mem);
+    this->alloc_mem = NULL;
+    this->n_buffers = 0;
+    this->have_buffers = false;
+  }
+  return SPA_RESULT_OK;
+}
+
 static SpaResult
 spa_audiotestsrc_node_port_set_format (SpaNode            *node,
                                        uint32_t            port_id,
@@ -425,7 +437,7 @@ spa_audiotestsrc_node_port_set_format (SpaNode            *node,
 
   if (format == NULL) {
     this->have_format = false;
-    this->have_buffers = false;
+    clear_buffers (this);
   } else {
     if ((res = spa_format_audio_parse (format, &this->current_format)) < 0)
       return res;
@@ -535,14 +547,7 @@ spa_audiotestsrc_node_port_use_buffers (SpaNode         *node,
   if (!this->have_format)
     return SPA_RESULT_NO_FORMAT;
 
-  if (this->have_buffers) {
+  clear_buffers (this);
-    if (this->alloc_mem)
-      spa_memory_unref (&this->alloc_mem->mem);
-    this->alloc_mem = NULL;
-    this->buffers = NULL;
-    this->n_buffers = 0;
-    this->have_buffers = false;
-  }
   if (buffers != NULL && n_buffers != 0) {
     unsigned int i;
 
@@ -578,16 +583,16 @@ spa_audiotestsrc_node_port_use_buffers (SpaNode         *node,
       b->ptr = SPA_MEMBER (spa_memory_ensure_ptr (mem), d[0].mem.offset, void);
       b->size = d[0].mem.size;
     }
-    this->buffers = buffers;
     this->n_buffers = n_buffers;
     this->have_buffers = true;
     this->empty = this->alloc_buffers;
   }
 
-  if (this->have_buffers)
+  if (this->have_buffers) {
     update_state (this, SPA_NODE_STATE_PAUSED);
-  else
+  } else {
     update_state (this, SPA_NODE_STATE_READY);
+  }
 
   return SPA_RESULT_OK;
 }
@@ -600,6 +605,22 @@ spa_audiotestsrc_node_port_alloc_buffers (SpaNode         *node,
                                           SpaBuffer      **buffers,
                                           uint32_t        *n_buffers)
 {
+  SpaAudioTestSrc *this;
+
+  if (node == NULL || node->handle == NULL)
+    return SPA_RESULT_INVALID_ARGUMENTS;
+
+  this = (SpaAudioTestSrc *) node->handle;
+
+  if (port_id != 0)
+    return SPA_RESULT_INVALID_PORT;
+
+  if (!this->have_format)
+    return SPA_RESULT_NO_FORMAT;
+
+  if (!this->have_buffers)
+    return SPA_RESULT_NO_BUFFERS;
+
   return SPA_RESULT_NOT_IMPLEMENTED;
 }
 

spa/plugins/v4l2/v4l2-utils.c

@@ -862,12 +862,8 @@ v4l2_on_fd_events (SpaPollNotifyData *data)
   SpaNodeEvent event;
   SpaNodeEventHaveOutput ho;
 
-  if (data->fds[0].revents & POLLERR) {
+  if (data->fds[0].revents & POLLERR)
-
     return -1;
-  }
-
-
 
   if (mmap_read (this) < 0)
     return 0;