ringbuffer: add support for timestamps

Make it possible for subclasses to provide the timestamp (as an absolute time
against the pipeline clock) of the last read data.
Fix up alsa to provide the timestamp received from alsa. Because the alsa
timestamps are in monotonic time, we can only do this when the monotonic clock
has been selected as the pipeline clock.

Fixes https://bugzilla.gnome.org/show_bug.cgi?id=635256

ext/alsa/gstalsasrc.c

@@ -72,7 +72,8 @@ static void gst_alsasrc_set_property (GObject * object,
     guint prop_id, const GValue * value, GParamSpec * pspec);
 static void gst_alsasrc_get_property (GObject * object,
     guint prop_id, GValue * value, GParamSpec * pspec);
-
+static GstStateChangeReturn gst_alsasrc_change_state (GstElement * element,
+    GstStateChange transition);
 static GstCaps *gst_alsasrc_getcaps (GstBaseSrc * bsrc, GstCaps * filter);
 
 static gboolean gst_alsasrc_open (GstAudioSrc * asrc);
@@ -80,7 +81,8 @@ static gboolean gst_alsasrc_prepare (GstAudioSrc * asrc,
     GstAudioRingBufferSpec * spec);
 static gboolean gst_alsasrc_unprepare (GstAudioSrc * asrc);
 static gboolean gst_alsasrc_close (GstAudioSrc * asrc);
-static guint gst_alsasrc_read (GstAudioSrc * asrc, gpointer data, guint length);
+static guint gst_alsasrc_read
+    (GstAudioSrc * asrc, gpointer data, guint length, GstClockTime * timestamp);
 static guint gst_alsasrc_delay (GstAudioSrc * asrc);
 static void gst_alsasrc_reset (GstAudioSrc * asrc);
 
@@ -150,6 +152,7 @@ gst_alsasrc_class_init (GstAlsaSrcClass * klass)
   gstaudiosrc_class->read = GST_DEBUG_FUNCPTR (gst_alsasrc_read);
   gstaudiosrc_class->delay = GST_DEBUG_FUNCPTR (gst_alsasrc_delay);
   gstaudiosrc_class->reset = GST_DEBUG_FUNCPTR (gst_alsasrc_reset);
+  gstelement_class->change_state = GST_DEBUG_FUNCPTR (gst_alsasrc_change_state);
 
   g_object_class_install_property (gobject_class, PROP_DEVICE,
       g_param_spec_string ("device", "Device",
@@ -217,6 +220,41 @@ gst_alsasrc_get_property (GObject * object, guint prop_id,
   }
 }
 
+static GstStateChangeReturn
+gst_alsasrc_change_state (GstElement * element, GstStateChange transition)
+{
+  GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS;
+  GstAudioBaseSrc *src = GST_AUDIO_BASE_SRC (element);
+  GstAlsaSrc *alsa = GST_ALSA_SRC (element);
+  GstClock *clk;
+
+  switch (transition) {
+      /* show the compiler that we care */
+    case GST_STATE_CHANGE_NULL_TO_READY:
+    case GST_STATE_CHANGE_READY_TO_PAUSED:
+    case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
+    case GST_STATE_CHANGE_PAUSED_TO_READY:
+    case GST_STATE_CHANGE_READY_TO_NULL:
+      break;
+
+    case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
+      clk = src->clock;
+      alsa->driver_timestamps = FALSE;
+      if (GST_IS_SYSTEM_CLOCK (clk)) {
+        gint clocktype;
+        g_object_get (clk, "clock-type", &clocktype, NULL);
+        if (clocktype == GST_CLOCK_TYPE_MONOTONIC) {
+          GST_INFO ("Using driver timestamps !");
+          alsa->driver_timestamps = TRUE;
+        }
+      }
+      break;
+  }
+  ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
+
+  return ret;
+}
+
 static void
 gst_alsasrc_init (GstAlsaSrc * alsasrc)
 {
@@ -224,6 +262,7 @@ gst_alsasrc_init (GstAlsaSrc * alsasrc)
 
   alsasrc->device = g_strdup (DEFAULT_PROP_DEVICE);
   alsasrc->cached_caps = NULL;
+  alsasrc->driver_timestamps = FALSE;
 
   g_mutex_init (&alsasrc->alsa_lock);
 }
@@ -450,6 +489,9 @@ set_swparams (GstAlsaSrc * alsa)
   /* start the transfer on first read */
   CHECK (snd_pcm_sw_params_set_start_threshold (alsa->handle, params,
           0), start_threshold);
+  /* use monotonic timestamping */
+  CHECK (snd_pcm_sw_params_set_tstamp_mode (alsa->handle, params,
+          SND_PCM_TSTAMP_MMAP), tstamp_mode);
 
 #if GST_CHECK_ALSA_VERSION(1,0,16)
   /* snd_pcm_sw_params_set_xfer_align() is deprecated, alignment is always 1 */
@@ -488,6 +530,13 @@ set_avail:
     snd_pcm_sw_params_free (params);
     return err;
   }
+tstamp_mode:
+  {
+    GST_ELEMENT_ERROR (alsa, RESOURCE, SETTINGS, (NULL),
+        ("Unable to set tstamp mode for playback: %s", snd_strerror (err)));
+    snd_pcm_sw_params_free (params);
+    return err;
+  }
 #if !GST_CHECK_ALSA_VERSION(1,0,16)
 set_align:
   {
@@ -644,7 +693,8 @@ gst_alsasrc_open (GstAudioSrc * asrc)
   alsa = GST_ALSA_SRC (asrc);
 
   CHECK (snd_pcm_open (&alsa->handle, alsa->device, SND_PCM_STREAM_CAPTURE,
-          SND_PCM_NONBLOCK), open_error);
+          (alsa->driver_timestamps == TRUE) ? 0 : SND_PCM_NONBLOCK),
+      open_error);
 
   return TRUE;
 
@@ -780,8 +830,66 @@ xrun_recovery (GstAlsaSrc * alsa, snd_pcm_t * handle, gint err)
   return err;
 }
 
+static GstClockTime
+gst_alsasrc_get_timestamp (GstAlsaSrc * asrc)
+{
+  snd_pcm_status_t *status;
+  snd_htimestamp_t tstamp;
+  GstClockTime timestamp;
+  snd_pcm_uframes_t avail;
+  gint err = -EPIPE;
+
+  if (G_UNLIKELY (!asrc)) {
+    GST_ERROR_OBJECT (asrc, "No alsa handle created yet !");
+    return GST_CLOCK_TIME_NONE;
+  }
+
+  if (G_UNLIKELY (snd_pcm_status_malloc (&status) != 0)) {
+    GST_ERROR_OBJECT (asrc, "snd_pcm_status_malloc failed");
+    return GST_CLOCK_TIME_NONE;
+  }
+
+  if (G_UNLIKELY (snd_pcm_status (asrc->handle, status) != 0)) {
+    GST_ERROR_OBJECT (asrc, "snd_pcm_status failed");
+    return GST_CLOCK_TIME_NONE;
+  }
+
+  /* in case an xrun condition has occured we need to handle this */
+  if (snd_pcm_status_get_state (status) != SND_PCM_STATE_RUNNING) {
+    if (xrun_recovery (asrc, asrc->handle, err) < 0) {
+      GST_WARNING_OBJECT (asrc, "Could not recover from xrun condition !");
+    }
+    /* reload the status alsa status object, since recovery made it invalid */
+    if (G_UNLIKELY (snd_pcm_status (asrc->handle, status) != 0)) {
+      GST_ERROR_OBJECT (asrc, "snd_pcm_status failed");
+    }
+  }
+
+  /* get high resolution time stamp from driver */
+  snd_pcm_status_get_htstamp (status, &tstamp);
+  timestamp = GST_TIMESPEC_TO_TIME (tstamp);
+
+  /* max available frames sets the depth of the buffer */
+  avail = snd_pcm_status_get_avail (status);
+
+  /* calculate the timestamp of the next sample to be read */
+  timestamp -= gst_util_uint64_scale_int (avail, GST_SECOND, asrc->rate);
+
+  /* compensate for the fact that we really need the timestamp of the
+   * previously read data segment */
+  timestamp -= asrc->period_time * 1000;
+
+  snd_pcm_status_free (status);
+
+  GST_LOG_OBJECT (asrc, "ALSA timestamp : %" GST_TIME_FORMAT
+      ", delay %lu", GST_TIME_ARGS (timestamp), avail);
+
+  return timestamp;
+}
+
 static guint
-gst_alsasrc_read (GstAudioSrc * asrc, gpointer data, guint length)
+gst_alsasrc_read (GstAudioSrc * asrc, gpointer data, guint length,
+    GstClockTime * timestamp)
 {
   GstAlsaSrc *alsa;
   gint err;
@@ -810,6 +918,10 @@ gst_alsasrc_read (GstAudioSrc * asrc, gpointer data, guint length)
   }
   GST_ALSA_SRC_UNLOCK (asrc);
 
+  /* if driver timestamps are enabled we need to return this here */
+  if (alsa->driver_timestamps && timestamp)
+    *timestamp = gst_alsasrc_get_timestamp (alsa);
+
   return length - (cptr * alsa->bpf);
 
 read_error:

ext/alsa/gstalsasrc.h

@@ -64,7 +64,6 @@ struct _GstAlsaSrc {
   guint                 channels;
   gint                  bpf;
   gboolean              driver_timestamps;
-  GstClockTime          first_alsa_ts;
 
   guint                 buffer_time;
   guint                 period_time;

gst-libs/gst/audio/gstaudiobasesrc.c

@@ -759,7 +759,9 @@ gst_audio_base_src_create (GstBaseSrc * bsrc, guint64 offset, guint length,
   GstAudioRingBufferSpec *spec;
   guint read;
   GstClockTime timestamp, duration;
+  GstClockTime rb_timestamp = GST_CLOCK_TIME_NONE;
   GstClock *clock;
+  gboolean first;
 
   ringbuffer = src->ringbuffer;
   spec = &ringbuffer->spec;
@@ -803,8 +805,16 @@ gst_audio_base_src_create (GstBaseSrc * bsrc, guint64 offset, guint length,
 
   gst_buffer_map (buf, &info, GST_MAP_WRITE);
   ptr = info.data;
+  first = TRUE;
   do {
-    read = gst_audio_ring_buffer_read (ringbuffer, sample, ptr, samples);
+    GstClockTime tmp_ts;
+
+    read =
+        gst_audio_ring_buffer_read (ringbuffer, sample, ptr, samples, &tmp_ts);
+    if (first && GST_CLOCK_TIME_IS_VALID (tmp_ts)) {
+      first = FALSE;
+      rb_timestamp = tmp_ts;
+    }
     GST_DEBUG_OBJECT (src, "read %u of %u", read, samples);
     /* if we read all, we're done */
     if (read == samples)
@@ -984,8 +994,13 @@ gst_audio_base_src_create (GstBaseSrc * bsrc, guint64 offset, guint length,
   } else {
     GstClockTime base_time;
 
-    /* to get the timestamp against the clock we also need to add our offset */
-    timestamp = gst_audio_clock_adjust (clock, timestamp);
+    if (GST_CLOCK_TIME_IS_VALID (rb_timestamp)) {
+      /* the read method returned a timestamp so we use this instead */
+      timestamp = rb_timestamp;
+    } else {
+      /* to get the timestamp against the clock we also need to add our offset */
+      timestamp = gst_audio_clock_adjust (clock, timestamp);
+    }
 
     /* we are not slaved, subtract base_time */
     base_time = GST_ELEMENT_CAST (src)->base_time;
@@ -1014,6 +1029,9 @@ no_sync:
 
   *outbuf = buf;
 
+  GST_LOG_OBJECT (src, "Pushed buffer timestamp %" GST_TIME_FORMAT,
+      GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buf)));
+
   return GST_FLOW_OK;
 
   /* ERRORS */

gst-libs/gst/audio/gstaudioringbuffer.c

@@ -521,7 +521,7 @@ gst_audio_ring_buffer_acquire (GstAudioRingBuffer * buf,
 {
   gboolean res = FALSE;
   GstAudioRingBufferClass *rclass;
-  gint segsize, bpf;
+  gint segsize, bpf, i;
 
   g_return_val_if_fail (GST_IS_AUDIO_RING_BUFFER (buf), FALSE);
 
@@ -548,6 +548,14 @@ gst_audio_ring_buffer_acquire (GstAudioRingBuffer * buf,
   if (G_UNLIKELY (!res))
     goto acquire_failed;
 
+  GST_INFO_OBJECT (buf, "Allocating an array for %d timestamps",
+      spec->segtotal);
+  buf->timestamps = g_slice_alloc0 (sizeof (GstClockTime) * spec->segtotal);
+  /* initialize array with invalid timestamps */
+  for (i = 0; i < spec->segtotal; i++) {
+    buf->timestamps[i] = GST_CLOCK_TIME_NONE;
+  }
+
   if (G_UNLIKELY ((bpf = buf->spec.info.bpf) == 0))
     goto invalid_bpf;
 
@@ -632,6 +640,14 @@ gst_audio_ring_buffer_release (GstAudioRingBuffer * buf)
   gst_audio_ring_buffer_stop (buf);
 
   GST_OBJECT_LOCK (buf);
+
+  if (G_LIKELY (buf->timestamps)) {
+    GST_INFO_OBJECT (buf, "Freeing timestamp buffer, %d entries",
+        buf->spec.segtotal);
+    g_slice_free1 (sizeof (GstClockTime) * buf->spec.segtotal, buf->timestamps);
+    buf->timestamps = NULL;
+  }
+
   if (G_UNLIKELY (!buf->acquired))
     goto was_released;
 
@@ -1597,6 +1613,7 @@ gst_audio_ring_buffer_commit (GstAudioRingBuffer * buf, guint64 * sample,
  * @sample: the sample position of the data
  * @data: where the data should be read
  * @len: the number of samples in data to read
+ * @timestamp: where the timestamp is returned
  *
  * Read @len samples from the ringbuffer into the memory pointed 
  * to by @data.
@@ -1606,6 +1623,8 @@ gst_audio_ring_buffer_commit (GstAudioRingBuffer * buf, guint64 * sample,
  * @len should not be a multiple of the segment size of the ringbuffer
  * although it is recommended.
  *
+ * @timestamp will return the timestamp associated with the data returned.
+ *
  * Returns: The number of samples read from the ringbuffer or -1 on
  * error.
  *
@@ -1613,10 +1632,10 @@ gst_audio_ring_buffer_commit (GstAudioRingBuffer * buf, guint64 * sample,
  */
 guint
 gst_audio_ring_buffer_read (GstAudioRingBuffer * buf, guint64 sample,
-    guint8 * data, guint len)
+    guint8 * data, guint len, GstClockTime * timestamp)
 {
   gint segdone;
-  gint segsize, segtotal, channels, bps, bpf, sps;
+  gint segsize, segtotal, channels, bps, bpf, sps, readseg = 0;
   guint8 *dest;
   guint to_read;
   gboolean need_reorder;
@@ -1638,7 +1657,7 @@ gst_audio_ring_buffer_read (GstAudioRingBuffer * buf, guint64 sample,
   /* read enough samples */
   while (to_read > 0) {
     gint sampleslen;
-    gint readseg, sampleoff;
+    gint sampleoff;
 
     /* figure out the segment and the offset inside the segment where
      * the sample should be read from. */
@@ -1710,6 +1729,12 @@ gst_audio_ring_buffer_read (GstAudioRingBuffer * buf, guint64 sample,
     data += sampleslen * bpf;
   }
 
+  if (buf->timestamps && timestamp) {
+    *timestamp = buf->timestamps[readseg % segtotal];
+    GST_INFO_OBJECT (buf, "Retrieved timestamp %" GST_TIME_FORMAT
+        " @ %d", GST_TIME_ARGS (*timestamp), readseg % segtotal);
+  }
+
   return len - to_read;
 
   /* ERRORS */
@@ -1894,3 +1919,30 @@ gst_audio_ring_buffer_set_channel_positions (GstAudioRingBuffer * buf,
     }
   }
 }
+
+/**
+ * gst_ring_buffer_set_timestamp:
+ * @buf: the #GstRingBuffer
+ * @readseg: the current data segment
+ * @timestamp: The new timestamp of the buffer.
+ *
+ * Set a new timestamp on the buffer.
+ *
+ * MT safe.
+ *
+ * Since:
+ */
+void
+gst_audio_ring_buffer_set_timestamp (GstAudioRingBuffer * buf, gint readseg,
+    GstClockTime timestamp)
+{
+  g_return_if_fail (GST_IS_AUDIO_RING_BUFFER (buf));
+
+  GST_INFO_OBJECT (buf, "Storing timestamp %" GST_TIME_FORMAT
+      " @ %d", GST_TIME_ARGS (timestamp), readseg);
+  if (buf->timestamps) {
+    buf->timestamps[readseg] = timestamp;
+  } else {
+    GST_ERROR_OBJECT (buf, "Could not store timestamp, no timestamps buffer");
+  }
+}

gst-libs/gst/audio/gstaudioringbuffer.h

@@ -176,6 +176,7 @@ struct _GstAudioRingBuffer {
   gboolean                    acquired;
   guint8                     *memory;
   gsize                       size;
+  GstClockTime               *timestamps;
   GstAudioRingBufferSpec      spec;
   gint                        samples_per_seg;
   guint8                     *empty_seg;
@@ -309,7 +310,11 @@ guint           gst_audio_ring_buffer_commit          (GstAudioRingBuffer * buf,
 
 /* read samples */
 guint           gst_audio_ring_buffer_read            (GstAudioRingBuffer *buf, guint64 sample,
-                                                       guint8 *data, guint len);
+                                                       guint8 *data, guint len, GstClockTime *timestamp);
+
+/* Set timestamp on buffer */
+void            gst_audio_ring_buffer_set_timestamp   (GstAudioRingBuffer * buf, gint readseg, GstClockTime 
+                                                       timestamp);
 
 /* mostly protected */
 /* not yet implemented

gst-libs/gst/audio/gstaudiosrc.c

@@ -190,7 +190,8 @@ gst_audio_src_ring_buffer_class_init (GstAudioSrcRingBufferClass * klass)
       GST_DEBUG_FUNCPTR (gst_audio_src_ring_buffer_delay);
 }
 
-typedef guint (*ReadFunc) (GstAudioSrc * src, gpointer data, guint length);
+typedef guint (*ReadFunc)
+  (GstAudioSrc * src, gpointer data, guint length, GstClockTime * timestamp);
 
 /* this internal thread does nothing else but read samples from the audio device.
  * It will read each segment in the ringbuffer and will update the play
@@ -212,8 +213,7 @@ audioringbuffer_thread_func (GstAudioRingBuffer * buf)
 
   GST_DEBUG_OBJECT (src, "enter thread");
 
-  readfunc = csrc->read;
-  if (readfunc == NULL)
+  if ((readfunc = csrc->read) == NULL)
     goto no_function;
 
   /* FIXME: maybe we should at least use a custom pointer type here? */
@@ -229,13 +229,14 @@ audioringbuffer_thread_func (GstAudioRingBuffer * buf)
     gint left, len;
     guint8 *readptr;
     gint readseg;
+    GstClockTime timestamp = GST_CLOCK_TIME_NONE;
 
     if (gst_audio_ring_buffer_prepare_read (buf, &readseg, &readptr, &len)) {
       gint read;
 
       left = len;
       do {
-        read = readfunc (src, readptr, left);
+        read = readfunc (src, readptr, left, &timestamp);
         GST_LOG_OBJECT (src, "transfered %d bytes of %d to segment %d", read,
             left, readseg);
         if (read < 0 || read > left) {
@@ -248,6 +249,9 @@ audioringbuffer_thread_func (GstAudioRingBuffer * buf)
         readptr += read;
       } while (left > 0);
 
+      /* Update timestamp on buffer if required */
+      gst_audio_ring_buffer_set_timestamp (buf, readseg, timestamp);
+
       /* we read one segment */
       gst_audio_ring_buffer_advance (buf, 1);
     } else {

gst-libs/gst/audio/gstaudiosrc.h

@@ -81,7 +81,8 @@ struct _GstAudioSrcClass {
   /* close the device */
   gboolean (*close)     (GstAudioSrc *src);
   /* read samples from the device */
-  guint    (*read)      (GstAudioSrc *src, gpointer data, guint length);
+  guint    (*read)      (GstAudioSrc *src, gpointer data, guint length,
+      GstClockTime *timestamp);
   /* get number of samples queued in the device */
   guint    (*delay)     (GstAudioSrc *src);
   /* reset the audio device, unblock from a write */