gstbasesink.c

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

  if (format == GST_FORMAT_TIME) {
    GST_DEBUG_OBJECT (basesink,
        "configured NEWSEGMENT update %d, rate %lf, applied rate %lf, "
        "format GST_FORMAT_TIME, "
        "%" GST_TIME_FORMAT " -- %" GST_TIME_FORMAT
        ", time %" GST_TIME_FORMAT ", accum %" GST_TIME_FORMAT,
        update, rate, arate, GST_TIME_ARGS (segment->start),
        GST_TIME_ARGS (segment->stop), GST_TIME_ARGS (segment->time),
        GST_TIME_ARGS (segment->accum));
  } else {
    GST_DEBUG_OBJECT (basesink,
        "configured NEWSEGMENT update %d, rate %lf, applied rate %lf, "
        "format %d, "
        "%" G_GINT64_FORMAT " -- %" G_GINT64_FORMAT ", time %"
        G_GINT64_FORMAT ", accum %" G_GINT64_FORMAT, update, rate, arate,
        segment->format, segment->start, segment->stop, segment->time,
        segment->accum);
  }
  GST_OBJECT_UNLOCK (basesink);
}

/* with PREROLL_LOCK, STREAM_LOCK */
static gboolean
gst_base_sink_commit_state (GstBaseSink * basesink)
{
  /* commit state and proceed to next pending state */
  GstState current, next, pending, post_pending;
  gboolean post_paused = FALSE;
  gboolean post_async_done = FALSE;
  gboolean post_playing = FALSE;
  gboolean sync;

  /* we are certainly not playing async anymore now */
  basesink->playing_async = FALSE;

  GST_OBJECT_LOCK (basesink);
  current = GST_STATE (basesink);
  next = GST_STATE_NEXT (basesink);
  pending = GST_STATE_PENDING (basesink);
  post_pending = pending;
  sync = basesink->sync;

  switch (pending) {
    case GST_STATE_PLAYING:
    {
      GstBaseSinkClass *bclass;
      GstStateChangeReturn ret;

      bclass = GST_BASE_SINK_GET_CLASS (basesink);

      GST_DEBUG_OBJECT (basesink, "commiting state to PLAYING");

      basesink->need_preroll = FALSE;
      post_async_done = TRUE;
      basesink->priv->commited = TRUE;
      post_playing = TRUE;
      /* post PAUSED too when we were READY */
      if (current == GST_STATE_READY) {
        post_paused = TRUE;
      }

      /* make sure we notify the subclass of async playing */
      if (bclass->async_play) {
        ret = bclass->async_play (basesink);
        if (ret == GST_STATE_CHANGE_FAILURE)
          goto async_failed;
      }
      break;
    }
    case GST_STATE_PAUSED:
      GST_DEBUG_OBJECT (basesink, "commiting state to PAUSED");
      post_paused = TRUE;
      post_async_done = TRUE;
      basesink->priv->commited = TRUE;
      post_pending = GST_STATE_VOID_PENDING;
      break;
    case GST_STATE_READY:
    case GST_STATE_NULL:
      goto stopping;
    case GST_STATE_VOID_PENDING:
      goto nothing_pending;
    default:
      break;
  }

  GST_STATE (basesink) = pending;
  GST_STATE_NEXT (basesink) = GST_STATE_VOID_PENDING;
  GST_STATE_PENDING (basesink) = GST_STATE_VOID_PENDING;
  GST_STATE_RETURN (basesink) = GST_STATE_CHANGE_SUCCESS;
  GST_OBJECT_UNLOCK (basesink);

  if (post_paused) {
    gst_element_post_message (GST_ELEMENT_CAST (basesink),
        gst_message_new_state_changed (GST_OBJECT_CAST (basesink),
            current, next, post_pending));
  }
  if (post_async_done) {
    gst_element_post_message (GST_ELEMENT_CAST (basesink),
        gst_message_new_async_done (GST_OBJECT_CAST (basesink)));
  }
  if (post_playing) {
    gst_element_post_message (GST_ELEMENT_CAST (basesink),
        gst_message_new_state_changed (GST_OBJECT_CAST (basesink),
            next, pending, GST_STATE_VOID_PENDING));
  }

  GST_STATE_BROADCAST (basesink);

  return TRUE;

nothing_pending:
  {
    /* Depending on the state, set our vars. We get in this situation when the
     * state change function got a change to update the state vars before the
     * streaming thread did. This is fine but we need to make sure that we
     * update the need_preroll var since it was TRUE when we got here and might
     * become FALSE if we got to PLAYING. */
    GST_DEBUG_OBJECT (basesink, "nothing to commit, now in %s",
        gst_element_state_get_name (current));
    switch (current) {
      case GST_STATE_PLAYING:
        basesink->need_preroll = FALSE;
        break;
      case GST_STATE_PAUSED:
        basesink->need_preroll = TRUE;
        break;
      default:
        basesink->need_preroll = FALSE;
        basesink->flushing = TRUE;
        break;
    }
    GST_OBJECT_UNLOCK (basesink);
    return TRUE;
  }
stopping:
  {
    /* app is going to READY */
    GST_DEBUG_OBJECT (basesink, "stopping");
    basesink->need_preroll = FALSE;
    basesink->flushing = TRUE;
    GST_OBJECT_UNLOCK (basesink);
    return FALSE;
  }
async_failed:
  {
    GST_DEBUG_OBJECT (basesink, "async commit failed");
    GST_STATE_RETURN (basesink) = GST_STATE_CHANGE_FAILURE;
    GST_OBJECT_UNLOCK (basesink);
    return FALSE;
  }
}


/* with STREAM_LOCK, PREROLL_LOCK
 *
 * Returns TRUE if the object needs synchronisation and takes therefore
 * part in prerolling.
 *
 * rsstart/rsstop contain the start/stop in stream time.
 * rrstart/rrstop contain the start/stop in running time.
 */
static gboolean
gst_base_sink_get_sync_times (GstBaseSink * basesink, GstMiniObject * obj,
    GstClockTime * rsstart, GstClockTime * rsstop,
    GstClockTime * rrstart, GstClockTime * rrstop, gboolean * do_sync)
{
  GstBaseSinkClass *bclass;
  GstBuffer *buffer;
  GstClockTime start, stop;     /* raw start/stop timestamps */
  gint64 cstart, cstop;         /* clipped raw timestamps */
  gint64 rstart, rstop;         /* clipped timestamps converted to running time */
  GstClockTime sstart, sstop;   /* clipped timestamps converted to stream time */
  GstFormat format;
  GstSegment *segment;

  /* start with nothing */
  start = stop = sstart = sstop = rstart = rstop = -1;

  if (G_UNLIKELY (GST_IS_EVENT (obj))) {
    GstEvent *event = GST_EVENT_CAST (obj);

    switch (GST_EVENT_TYPE (event)) {
        /* EOS event needs syncing */
      case GST_EVENT_EOS:
        sstart = sstop = basesink->priv->current_sstop;
        rstart = rstop = basesink->priv->eos_rtime;
        *do_sync = rstart != -1;
        GST_DEBUG_OBJECT (basesink, "sync times for EOS %" GST_TIME_FORMAT,
            GST_TIME_ARGS (rstart));
        goto done;
        /* other events do not need syncing */
        /* FIXME, maybe NEWSEGMENT might need synchronisation
         * since the POSITION query depends on accumulated times and
         * we cannot accumulate the current segment before the previous
         * one completed.
         */
      default:
        return FALSE;
    }
  }

  /* else do buffer sync code */
  buffer = GST_BUFFER_CAST (obj);

  bclass = GST_BASE_SINK_GET_CLASS (basesink);

  /* just get the times to see if we need syncing */
  if (bclass->get_times)
    bclass->get_times (basesink, buffer, &start, &stop);

  if (start == -1) {
    gst_base_sink_get_times (basesink, buffer, &start, &stop);
    *do_sync = FALSE;
  } else {
    *do_sync = TRUE;
  }

  GST_DEBUG_OBJECT (basesink, "got times start: %" GST_TIME_FORMAT
      ", stop: %" GST_TIME_FORMAT ", do_sync %d", GST_TIME_ARGS (start),
      GST_TIME_ARGS (stop), *do_sync);

  /* collect segment and format for code clarity */
  segment = &basesink->segment;
  format = segment->format;

  /* no timestamp clipping if we did not * get a TIME segment format */
  if (G_UNLIKELY (format != GST_FORMAT_TIME)) {
    cstart = start;
    cstop = stop;
    goto do_times;
  }

  /* clip */
  if (G_UNLIKELY (!gst_segment_clip (segment, GST_FORMAT_TIME,
              (gint64) start, (gint64) stop, &cstart, &cstop)))
    goto out_of_segment;

  if (G_UNLIKELY (start != cstart || stop != cstop)) {
    GST_DEBUG_OBJECT (basesink, "clipped to: start %" GST_TIME_FORMAT
        ", stop: %" GST_TIME_FORMAT, GST_TIME_ARGS (cstart),
        GST_TIME_ARGS (cstop));
  }

  /* set last stop position */
  gst_segment_set_last_stop (segment, GST_FORMAT_TIME, cstop);

do_times:
  /* this can produce wrong values if we accumulated non-TIME segments. If this happens,
   * upstream is behaving very badly */
  sstart = gst_segment_to_stream_time (segment, format, cstart);
  sstop = gst_segment_to_stream_time (segment, format, cstop);
  rstart = gst_segment_to_running_time (segment, format, cstart);
  rstop = gst_segment_to_running_time (segment, format, cstop);

done:
  /* save times */
  *rsstart = sstart;
  *rsstop = sstop;
  *rrstart = rstart;
  *rrstop = rstop;

  /* buffers and EOS always need syncing and preroll */
  return TRUE;

  /* special cases */
out_of_segment:
  {
    /* should not happen since we clip them in the chain function already, 
     * we return FALSE so that we don't try to sync on it. */
    GST_ELEMENT_WARNING (basesink, STREAM, FAILED,
        (NULL), ("unexpected buffer out of segment found."));
    GST_LOG_OBJECT (basesink, "buffer skipped, not in segment");
    return FALSE;
  }
}

/* with STREAM_LOCK, PREROLL_LOCK
 *
 * Waits for the clock to reach @time. If @time is not valid, no
 * synchronisation is done and BADTIME is returned. 
 * If synchronisation is disabled in the element or there is no
 * clock, no synchronisation is done and BADTIME is returned.
 *
 * Else a blocking wait is performed on the clock. We save the ClockID
 * so we can unlock the entry at any time. While we are blocking, we 
 * release the PREROLL_LOCK so that other threads can interrupt the entry.
 *
 * @time is expressed in running time.
 */
static GstClockReturn
gst_base_sink_wait_clock (GstBaseSink * basesink, GstClockTime time,
    GstClockTimeDiff * jitter)
{
  GstClockID id;
  GstClockReturn ret;
  GstClock *clock;

  if (G_UNLIKELY (!GST_CLOCK_TIME_IS_VALID (time)))
    goto invalid_time;

  GST_OBJECT_LOCK (basesink);
  if (G_UNLIKELY (!basesink->sync))
    goto no_sync;

  if (G_UNLIKELY ((clock = GST_ELEMENT_CLOCK (basesink)) == NULL))
    goto no_clock;

  /* add base time and latency */
  time += GST_ELEMENT_CAST (basesink)->base_time;
  time += basesink->priv->latency;

  id = gst_clock_new_single_shot_id (clock, time);
  GST_OBJECT_UNLOCK (basesink);

  basesink->clock_id = id;
  /* release the preroll lock while waiting */
  GST_PAD_PREROLL_UNLOCK (basesink->sinkpad);

  ret = gst_clock_id_wait (id, jitter);

  GST_PAD_PREROLL_LOCK (basesink->sinkpad);
  gst_clock_id_unref (id);
  basesink->clock_id = NULL;

  return ret;

  /* no syncing needed */
invalid_time:
  {
    GST_DEBUG_OBJECT (basesink, "time not valid, no sync needed");
    return GST_CLOCK_BADTIME;
  }
no_sync:
  {
    GST_DEBUG_OBJECT (basesink, "sync disabled");
    GST_OBJECT_UNLOCK (basesink);
    return GST_CLOCK_BADTIME;
  }
no_clock:
  {
    GST_DEBUG_OBJECT (basesink, "no clock, can't sync");
    GST_OBJECT_UNLOCK (basesink);
    return GST_CLOCK_BADTIME;
  }
}

/**
 * gst_base_sink_wait_preroll:
 * @sink: the sink
 *
 * If the #GstBaseSinkClass::render method performs its own synchronisation against
 * the clock it must unblock when going from PLAYING to the PAUSED state and call
 * this method before continuing to render the remaining data.
 *
 * This function will block until a state change to PLAYING happens (in which
 * case this function returns #GST_FLOW_OK) or the processing must be stopped due
 * to a state change to READY or a FLUSH event (in which case this function
 * returns #GST_FLOW_WRONG_STATE).
 *
 * Since: 0.10.11
 *
 * Returns: #GST_FLOW_OK if the preroll completed and processing can
 * continue. Any other return value should be returned from the render vmethod.
 */
GstFlowReturn
gst_base_sink_wait_preroll (GstBaseSink * sink)
{
  /* block until the state changes, or we get a flush, or something */
  GST_DEBUG_OBJECT (sink, "wait for preroll...");
  sink->have_preroll = TRUE;
  GST_PAD_PREROLL_WAIT (sink->sinkpad);
  sink->have_preroll = FALSE;
  GST_DEBUG_OBJECT (sink, "preroll done");
  if (G_UNLIKELY (sink->flushing))
    goto stopping;
  GST_DEBUG_OBJECT (sink, "continue after preroll");

  return GST_FLOW_OK;

  /* ERRORS */
stopping:
  {
    GST_DEBUG_OBJECT (sink, "preroll interrupted");
    return GST_FLOW_WRONG_STATE;
  }
}

/* with STREAM_LOCK, PREROLL_LOCK
 *
 * Make sure we are in PLAYING and synchronize an object to the clock.
 *
 * If we need preroll, we are not in PLAYING. We try to commit the state
 * if needed and then block if we still are not PLAYING.
 *
 * We start waiting on the clock in PLAYING. If we got interrupted, we
 * immediatly try to re-preroll.
 *
 * Some objects do not need synchronisation (most events) and so this function
 * immediatly returns GST_FLOW_OK.
 *
 * for objects that arrive later than max-lateness to be synchronized to the 
 * clock have the @late boolean set to TRUE.
 *
 * This function keeps a running average of the jitter (the diff between the
 * clock time and the requested sync time). The jitter is negative for
 * objects that arrive in time and positive for late buffers.
 *
 * does not take ownership of obj.
 */
static GstFlowReturn
gst_base_sink_do_sync (GstBaseSink * basesink, GstPad * pad,
    GstMiniObject * obj, gboolean * late)
{
  GstClockTimeDiff jitter;
  gboolean syncable;
  GstClockReturn status = GST_CLOCK_OK;
  GstClockTime sstart, sstop, rstart, rstop;
  gboolean do_sync;

  sstart = sstop = rstart = rstop = -1;
  do_sync = TRUE;

  basesink->priv->current_rstart = -1;

  /* update timing information for this object */
  syncable = gst_base_sink_get_sync_times (basesink, obj,
      &sstart, &sstop, &rstart, &rstop, &do_sync);

  /* a syncable object needs to participate in preroll and
   * clocking. All buffers and EOS are syncable. */
  if (G_UNLIKELY (!syncable))
    goto not_syncable;