gstbasesink.c

GStreamer是一个开源的多媒体框架库。利用它

  /* store timing info for current object */
  basesink->priv->current_rstart = rstart;
  basesink->priv->current_rstop = (rstop != -1 ? rstop : rstart);
  /* save sync time for eos when the previous object needed sync */
  basesink->priv->eos_rtime = (do_sync ? basesink->priv->current_rstop : -1);

  /* lock because we read this when answering the POSITION 
   * query. */
  GST_OBJECT_LOCK (basesink);
  basesink->priv->current_sstart = sstart;
  basesink->priv->current_sstop = (sstop != -1 ? sstop : sstart);
  GST_OBJECT_UNLOCK (basesink);

again:
  /* first do preroll, this makes sure we commit our state
   * to PAUSED and can continue to PLAYING. We cannot perform
   * any clock sync in PAUSED because there is no clock. 
   */
  while (G_UNLIKELY (basesink->need_preroll)) {
    GST_DEBUG_OBJECT (basesink, "prerolling object %p", obj);

    if (G_LIKELY (basesink->playing_async)) {
      /* commit state */
      if (G_UNLIKELY (!gst_base_sink_commit_state (basesink)))
        goto stopping;
    }

    /* need to recheck here because the commit state could have
     * made us not need the preroll anymore */
    if (G_LIKELY (basesink->need_preroll)) {
      /* block until the state changes, or we get a flush, or something */
      if (gst_base_sink_wait_preroll (basesink) != GST_FLOW_OK)
        goto flushing;
    }
  }

  if (!do_sync)
    goto done;

  /* preroll done, we can sync since we are in PLAYING now. */
  GST_DEBUG_OBJECT (basesink, "possibly waiting for clock to reach %"
      GST_TIME_FORMAT, GST_TIME_ARGS (rstart));

  /* this function will return immediatly if start == -1, no clock
   * or sync is disabled with GST_CLOCK_BADTIME. */
  status = gst_base_sink_wait_clock (basesink, rstart, &jitter);

  GST_DEBUG_OBJECT (basesink, "clock returned %d", status);

  /* invalid time, no clock or sync disabled, just render */
  if (status == GST_CLOCK_BADTIME)
    goto done;

  /* waiting could have been interrupted and we can be flushing now */
  if (G_UNLIKELY (basesink->flushing))
    goto flushing;

  /* check for unlocked by a state change, we are not flushing so
   * we can try to preroll on the current buffer. */
  if (G_UNLIKELY (status == GST_CLOCK_UNSCHEDULED)) {
    GST_DEBUG_OBJECT (basesink, "unscheduled, waiting some more");
    goto again;
  }

  /* successful syncing done, record observation */
  basesink->priv->current_jitter = jitter;

  /* check if the object should be dropped */
  *late = gst_base_sink_is_too_late (basesink, obj, rstart, rstop,
      status, jitter);

done:
  return GST_FLOW_OK;

  /* ERRORS */
not_syncable:
  {
    GST_DEBUG_OBJECT (basesink, "non syncable object %p", obj);
    return GST_FLOW_OK;
  }
flushing:
  {
    GST_DEBUG_OBJECT (basesink, "we are flushing");
    return GST_FLOW_WRONG_STATE;
  }
stopping:
  {
    GST_DEBUG_OBJECT (basesink, "stopping while commiting state");
    return GST_FLOW_WRONG_STATE;
  }
}

static gboolean
gst_base_sink_send_qos (GstBaseSink * basesink,
    gdouble proportion, GstClockTime time, GstClockTimeDiff diff)
{
  GstEvent *event;
  gboolean res;

  /* generate Quality-of-Service event */
  GST_CAT_DEBUG_OBJECT (GST_CAT_QOS, basesink,
      "qos: proportion: %lf, diff %" G_GINT64_FORMAT ", timestamp %"
      GST_TIME_FORMAT, proportion, diff, GST_TIME_ARGS (time));

  event = gst_event_new_qos (proportion, diff, time);

  /* send upstream */
  res = gst_pad_push_event (basesink->sinkpad, event);

  return res;
}

static void
gst_base_sink_perform_qos (GstBaseSink * sink, gboolean dropped)
{
  GstBaseSinkPrivate *priv;
  GstClockTime start, stop;
  GstClockTimeDiff jitter;
  GstClockTime pt, entered, left;
  GstClockTime duration;
  gdouble rate;

  priv = sink->priv;

  start = priv->current_rstart;

  /* if Quality-of-Service disabled, do nothing */
  if (!g_atomic_int_get (&priv->qos_enabled) || start == -1)
    return;

  stop = priv->current_rstop;
  jitter = priv->current_jitter;

  /* this is the time the buffer entered the sink */
  entered = start + jitter;
  /* this is the time the buffer left the sink */
  left = start + (jitter < 0 ? 0 : jitter);

  /* calculate duration of the buffer */
  if (stop != -1)
    duration = stop - start;
  else
    duration = -1;

  /* if we have the time when the last buffer left us, calculate
   * processing time */
  if (priv->last_left != -1) {
    if (entered > priv->last_left) {
      pt = entered - priv->last_left;
    } else {
      pt = 0;
    }
  } else {
    pt = priv->avg_pt;
  }

  GST_CAT_DEBUG_OBJECT (GST_CAT_QOS, sink, "start: %" GST_TIME_FORMAT
      ", entered %" GST_TIME_FORMAT ", left %" GST_TIME_FORMAT ", pt: %"
      GST_TIME_FORMAT ", duration %" GST_TIME_FORMAT ",jitter %"
      G_GINT64_FORMAT, GST_TIME_ARGS (start), GST_TIME_ARGS (entered),
      GST_TIME_ARGS (left), GST_TIME_ARGS (pt), GST_TIME_ARGS (duration),
      jitter);

  GST_CAT_DEBUG_OBJECT (GST_CAT_QOS, sink, "avg_duration: %" GST_TIME_FORMAT
      ", avg_pt: %" GST_TIME_FORMAT ", avg_rate: %g",
      GST_TIME_ARGS (priv->avg_duration), GST_TIME_ARGS (priv->avg_pt),
      priv->avg_rate);

  /* collect running averages. for first observations, we copy the
   * values */
  if (priv->avg_duration == -1)
    priv->avg_duration = duration;
  else
    priv->avg_duration = UPDATE_RUNNING_AVG (priv->avg_duration, duration);

  if (priv->avg_pt == -1)
    priv->avg_pt = pt;
  else
    priv->avg_pt = UPDATE_RUNNING_AVG (priv->avg_pt, pt);

  if (priv->avg_duration != 0)
    rate =
        gst_guint64_to_gdouble (priv->avg_pt) /
        gst_guint64_to_gdouble (priv->avg_duration);
  else
    rate = 0.0;

  if (priv->last_left != -1) {
    if (dropped || priv->avg_rate < 0.0) {
      priv->avg_rate = rate;
    } else {
      if (rate > 1.0)
        priv->avg_rate = UPDATE_RUNNING_AVG_N (priv->avg_rate, rate);
      else
        priv->avg_rate = UPDATE_RUNNING_AVG_P (priv->avg_rate, rate);
    }
  }

  GST_CAT_DEBUG_OBJECT (GST_CAT_QOS, sink,
      "updated: avg_duration: %" GST_TIME_FORMAT ", avg_pt: %" GST_TIME_FORMAT
      ", avg_rate: %g", GST_TIME_ARGS (priv->avg_duration),
      GST_TIME_ARGS (priv->avg_pt), priv->avg_rate);


  /* if we have a valid rate, start sending QoS messages */
  if (priv->avg_rate >= 0.0) {
    gst_base_sink_send_qos (sink, priv->avg_rate, priv->current_rstart,
        priv->current_jitter);
  }

  /* record when this buffer will leave us */
  priv->last_left = left;
}

/* reset all qos measuring */
static void
gst_base_sink_reset_qos (GstBaseSink * sink)
{
  GstBaseSinkPrivate *priv;

  priv = sink->priv;

  priv->last_in_time = -1;
  priv->last_left = -1;
  priv->avg_duration = -1;
  priv->avg_pt = -1;
  priv->avg_rate = -1.0;
  priv->avg_render = -1;
  priv->rendered = 0;
  priv->dropped = 0;
}

/* Checks if the object was scheduled too late.
 *
 * start/stop contain the raw timestamp start and stop values
 * of the object.
 *
 * status and jitter contain the return values from the clock wait.
 *
 * returns TRUE if the buffer was too late.
 */
static gboolean
gst_base_sink_is_too_late (GstBaseSink * basesink, GstMiniObject * obj,
    GstClockTime start, GstClockTime stop,
    GstClockReturn status, GstClockTimeDiff jitter)
{
  gboolean late;
  gint64 max_lateness;
  GstBaseSinkPrivate *priv;

  priv = basesink->priv;

  late = FALSE;

  /* only for objects that were too late */
  if (G_LIKELY (status != GST_CLOCK_EARLY))
    goto in_time;

  max_lateness = basesink->abidata.ABI.max_lateness;

  /* check if frame dropping is enabled */
  if (max_lateness == -1)
    goto no_drop;

  /* only check for buffers */
  if (G_UNLIKELY (!GST_IS_BUFFER (obj)))
    goto not_buffer;

  /* can't do check if we don't have a timestamp */
  if (G_UNLIKELY (start == -1))
    goto no_timestamp;

  /* we can add a valid stop time */
  if (stop != -1)
    max_lateness += stop;
  else
    max_lateness += start;

  /* if the jitter bigger than duration and lateness we are too late */
  if ((late = start + jitter > max_lateness)) {
    GST_DEBUG_OBJECT (basesink, "buffer is too late %" GST_TIME_FORMAT
        " > %" GST_TIME_FORMAT, GST_TIME_ARGS (start + jitter),
        GST_TIME_ARGS (max_lateness));
    /* !!emergency!!, if we did not receive anything valid for more than a 
     * second, render it anyway so the user sees something */
    if (priv->last_in_time && start - priv->last_in_time > GST_SECOND) {
      late = FALSE;
      GST_DEBUG_OBJECT (basesink,
          "**emergency** last buffer at %" GST_TIME_FORMAT " > GST_SECOND",
          GST_TIME_ARGS (priv->last_in_time));
    }
  }

done:
  if (!late) {
    priv->last_in_time = start;
  }
  return late;

  /* all is fine */
in_time:
  {
    GST_DEBUG_OBJECT (basesink, "object was scheduled in time");
    goto done;
  }
no_drop:
  {
    GST_DEBUG_OBJECT (basesink, "frame dropping disabled");
    goto done;
  }
not_buffer:
  {
    GST_DEBUG_OBJECT (basesink, "object is not a buffer");
    return FALSE;
  }
no_timestamp:
  {
    GST_DEBUG_OBJECT (basesink, "buffer has no timestamp");
    return FALSE;
  }
}

static void
gst_base_sink_do_render_stats (GstBaseSink * basesink, gboolean start)
{
  GstBaseSinkPrivate *priv;

  priv = basesink->priv;

  if (start) {
    g_get_current_time (&priv->start);
  } else {
    GstClockTime elapsed;

    g_get_current_time (&priv->stop);

    elapsed =
        GST_TIMEVAL_TO_TIME (priv->stop) - GST_TIMEVAL_TO_TIME (priv->start);

    if (priv->avg_render == -1)
      priv->avg_render = elapsed;
    else
      priv->avg_render = UPDATE_RUNNING_AVG (priv->avg_render, elapsed);

    GST_CAT_DEBUG_OBJECT (GST_CAT_QOS, basesink,
        "avg_render: %" GST_TIME_FORMAT, GST_TIME_ARGS (priv->avg_render));
  }
}

/* with STREAM_LOCK, PREROLL_LOCK,
 *
 * Synchronize the object on the clock and then render it.
 *
 * takes ownership of obj.
 */
static GstFlowReturn
gst_base_sink_render_object (GstBaseSink * basesink, GstPad * pad,
    GstMiniObject * obj)
{
  GstFlowReturn ret = GST_FLOW_OK;
  GstBaseSinkClass *bclass;
  gboolean late = FALSE;
  GstBaseSinkPrivate *priv;

  priv = basesink->priv;

  /* synchronize this object, non syncable objects return OK
   * immediatly. */
  ret = gst_base_sink_do_sync (basesink, pad, obj, &late);
  if (G_UNLIKELY (ret != GST_FLOW_OK))
    goto sync_failed;

  /* and now render, event or buffer. */
  if (G_LIKELY (GST_IS_BUFFER (obj))) {
    /* drop late buffers unconditionally, let's hope it's unlikely */
    if (G_UNLIKELY (late))
      goto dropped;

    bclass = GST_BASE_SINK_GET_CLASS (basesink);

    if (G_LIKELY (bclass->render)) {
      gint do_qos;

      /* read once, to get same value before and after */
      do_qos = g_atomic_int_get (&priv->qos_enabled);

      GST_DEBUG_OBJECT (basesink, "rendering buffer %p", obj);

      /* record rendering time for QoS and stats */
      if (do_qos)
        gst_base_sink_do_render_stats (basesink, TRUE);

      ret = bclass->render (basesink, GST_BUFFER_CAST (obj));

      priv->rendered++;

      if (do_qos)
        gst_base_sink_do_render_stats (basesink, FALSE);
    }
  } else {
    GstEvent *event = GST_EVENT_CAST (obj);
    gboolean event_res = TRUE;
    GstEventType type;

    bclass = GST_BASE_SINK_GET_CLASS (basesink);

    type = GST_EVENT_TYPE (event);

    GST_DEBUG_OBJECT (basesink, "rendering event %p, type %s", obj,
        gst_event_type_get_name (type));

    if (bclass->event)
      event_res = bclass->event (basesink, event);

    if (G_LIKELY (event_res)) {
      switch (type) {
        case GST_EVENT_EOS:
          /* the EOS event is completely handled so we mark
           * ourselves as being in the EOS state. eos is also 
           * protected by the object lock so we can read it when 
           * answering the POSITION query. */
          GST_OBJECT_LOCK (basesink);
          basesink->eos = TRUE;
          GST_OBJECT_UNLOCK (basesink);
          /* ok, now we can post the message */
          GST_DEBUG_OBJECT (basesink, "Now posting EOS");
          gst_element_post_message (GST_ELEMENT_CAST (basesink),
              gst_message_new_eos (GST_OBJECT_CAST (basesink)));
          break;
        case GST_EVENT_NEWSEGMENT:
          /* configure the segment */
          gst_base_sink_configure_segment (basesink, pad, event,
              &basesink->segment);
        default: