Gstreamer Dash直播数据下载分析

Gstreamer Dash直播数据下载分析始于列表下载，止于container数据送到文件demux，比如送到qtdemux，主要是数据下载，尔后的流程不在本文讨论范围。主要包括gst_adaptive_demux_stream_download_loop任务，gst_adaptive_demux_updates_loop任务和gst_system_clock_async_thread，_src_chain这几方面的任务，Dash直播的时候，音视频可能会分开，因此，数据下载可能会有多个任务，也就是gst_adaptive_demux_stream_download_loop可能会有多个，但分析一个不影响。

gst_adaptive_demux_updates_loop用于播放列表的更新，gst_adaptive_demux_stream_download_loop设置source，管理流下载，source用_src_chain函数向gst_adaptive_demux推流，_src_chain函数将流推给后端的demux。

播放过程中，同一个Adaptation Set中的Representation可以随意切换，并且复用同一条流，不用expose pad。如果dash或者hls要求切换pad，那么，当前流下载完成或者被取消以后，切换。因此需要特别注意当前流的下载状态。

详细分析如下：

gst_adaptive_demux_stream_download_loop任务

主要包含以下几个方面功能：

1. 流下载结束判断，结束任务。

2. PAD未Link时，待Link上了，重新下载数据。

3. 更新下一个下载片断的头部地址，数据地址，范围等信息。

4. 等待直播片断准备就绪

5. 数据下载

6. 直播数据更新fragment时遇到EOS出错,更新主列表，正常时更新当前流播放列表。这儿有一个问题，更新主列表时，exposed的旧流需要断开。

7. 非直播更新fragment时出错时，第一次出错时立即更新主列表。否则等待fragment时长一半，再更新主列表。超过一定次数后不会再重新下载。

8. 下一个片断更新和主播放列表的更新实作在hls/dash demux中进行。

/* this function will take the manifest_lock and will keep it until the end.* It will release it temporarily only when going to sleep.* Every time it takes the manifest_lock, it will check for cancelled condition*/
static void
gst_adaptive_demux_stream_download_loop (GstAdaptiveDemuxStream * stream)
{GstAdaptiveDemux *demux = stream->demux;/* 下一次下载的时间，单位ns */GstClockTime next_download = gst_adaptive_demux_get_monotonic_time (demux);
……/* 流被取消，如下条件成立时此标记为TRUE.* 1：当前流是旧流，新流建立expose_streams时，remove掉当前流，释放demux->streams中的流。因为Gtask，这些流并没有完全被清理干净，接着放到demux->priv->old_streams中，在如下2的线程中继续清除。* 2：正常下载时，demux->priv->old_streams存在，下载线程还存在时。* 3：SEEK,FLUSH时，调用gst_adaptive_demux_stop_tasks时。*/if (G_UNLIKELY (stream->cancelled)) {stream->last_ret = GST_FLOW_FLUSHING;g_mutex_unlock (&stream->fragment_download_lock);/* 直接返回了*/goto cancelled;}g_mutex_unlock (&stream->fragment_download_lock);/* Check if we're done with our segment，分段结束 */GST_ADAPTIVE_DEMUX_SEGMENT_LOCK (demux);if (demux->segment.rate > 0) {if (GST_CLOCK_TIME_IS_VALID (demux->segment.stop)&& stream->segment.position >= stream->segment.stop) {GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux);ret = GST_FLOW_EOS;gst_task_stop (stream->download_task);goto end_of_manifest;}} else {/* 后退播放时时，下载位置比start小，结束数据下载 */if (GST_CLOCK_TIME_IS_VALID (demux->segment.start)&& stream->segment.position <= stream->segment.start) {GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux);ret = GST_FLOW_EOS;gst_task_stop (stream->download_task);goto end_of_manifest;}}GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux);/* Cleanup old streams if any,exposed pad时保存的旧流环境，它们的任务已经停止，做资源清理 */if (G_UNLIKELY (demux->priv->old_streams != NULL)) {GList *old_streams = demux->priv->old_streams;demux->priv->old_streams = NULL;GST_DEBUG_OBJECT (stream->pad, "Cleaning up old streams");g_list_free_full (old_streams,(GDestroyNotify) gst_adaptive_demux_stream_free);GST_DEBUG_OBJECT (stream->pad, "Cleaning up old streams (done)");/* gst_adaptive_demux_stream_free had temporarily released the manifest_lock.* Recheck the cancelled flag.*/g_mutex_lock (&stream->fragment_download_lock);if (G_UNLIKELY (stream->cancelled)) {stream->last_ret = GST_FLOW_FLUSHING;g_mutex_unlock (&stream->fragment_download_lock);goto cancelled;}g_mutex_unlock (&stream->fragment_download_lock);}/* Restarting download, figure out new position* stream的pad没有Link上时为真，重新计算需要下载的位置并下载*/if (G_UNLIKELY (stream->restart_download)) {GstEvent *seg_event;GstClockTime cur, ts = 0;gint64 pos;if (gst_pad_peer_query_position (stream->pad, GST_FORMAT_TIME, &pos)) {ts = (GstClockTime) pos;} else {/* query other pads as some faulty element in the pad's branch might* reject position queries. This should be better than using the* demux segment position that can be much ahead */GList *iter;for (iter = demux->streams; iter != NULL; iter = g_list_next (iter)) {GstAdaptiveDemuxStream *cur_stream =(GstAdaptiveDemuxStream *) iter->data;if (gst_pad_peer_query_position (cur_stream->pad, GST_FORMAT_TIME,&pos)) {ts = (GstClockTime) pos;break;}}}cur =gst_segment_to_stream_time (&stream->segment, GST_FORMAT_TIME,stream->segment.position);/* we might have already pushed this data */ts = MAX (ts, cur);if (GST_CLOCK_TIME_IS_VALID (ts)) {GstClockTime offset, period_start;offset =gst_adaptive_demux_stream_get_presentation_offset (demux, stream);period_start = gst_adaptive_demux_get_period_start_time (demux);/* TODO check return */gst_adaptive_demux_stream_seek (demux, stream, demux->segment.rate >= 0,0, ts, &ts);stream->segment.position = ts - period_start + offset;}/* The stream's segment is still correct except for* the position, so let's send a new one with the* updated position */seg_event = gst_event_new_segment (&stream->segment);gst_event_set_seqnum (seg_event, demux->priv->segment_seqnum);gst_pad_push_event (stream->pad, seg_event);stream->discont = TRUE;stream->restart_download = FALSE;}live = gst_adaptive_demux_is_live (demux);/* Get information about the fragment to download，让hls或者dash这类demux来更新，更新url，range 等参数，后面介绍这个函数 */ret = gst_adaptive_demux_stream_update_fragment_info (demux, stream);if (ret == GST_FLOW_OK) {/* wait for live fragments to be available，等待直播的fragment可用 */if (live) {gint64 wait_time =gst_adaptive_demux_stream_get_fragment_waiting_time (demux, stream);if (wait_time > 0) {
……}}stream->last_ret = GST_FLOW_OK;next_download = gst_adaptive_demux_get_monotonic_time (demux);ret = gst_adaptive_demux_stream_download_fragment (stream);if (ret == GST_FLOW_FLUSHING) {g_mutex_lock (&stream->fragment_download_lock);if (G_UNLIKELY (stream->cancelled)) {stream->last_ret = GST_FLOW_FLUSHING;g_mutex_unlock (&stream->fragment_download_lock);goto cancelled;}g_mutex_unlock (&stream->fragment_download_lock);}} else {stream->last_ret = ret;}switch (ret) {case GST_FLOW_OK:break;                    /* all is good, let's go */case GST_FLOW_EOS:/* we push the EOS after releasing the object lock ，直播*/if (gst_adaptive_demux_is_live (demux)&& (demux->segment.rate == 1.0|| gst_adaptive_demux_stream_in_live_seek_range (demux,stream))) {GstAdaptiveDemuxClass *demux_class =GST_ADAPTIVE_DEMUX_GET_CLASS (demux);/* this might be a fragment download error, refresh the manifest, just in case，更新主列表，dash就是更新MPD文件，HLS会更新主M3U8文件 */if (!demux_class->requires_periodical_playlist_update (demux)) {ret = gst_adaptive_demux_update_manifest (demux);break;/* Wait only if we can ensure current manifest has been expired.* The meaning "we have next period" *WITH* EOS is that, current* period has been ended but we can continue to the next period，更新流的播放列表 */} else if (!gst_adaptive_demux_has_next_period (demux) &&gst_adaptive_demux_stream_wait_manifest_update (demux, stream)) {goto end;}/* 停止下载任务 */gst_task_stop (stream->download_task);if (stream->replaced) {goto end;}} else {gst_task_stop (stream->download_task);}/* demux中的所有流都结束EOS了 */if (gst_adaptive_demux_combine_flows (demux) == GST_FLOW_EOS) {if (gst_adaptive_demux_has_next_period (demux)) {/* 播放下一个period */gst_adaptive_demux_advance_period (demux);ret = GST_FLOW_OK;}}break;case GST_FLOW_NOT_LINKED: {GstFlowReturn ret;gst_task_stop (stream->download_task);/* demux的所有流都没有link */ret = gst_adaptive_demux_combine_flows (demux);if (ret == GST_FLOW_NOT_LINKED) {GST_ELEMENT_FLOW_ERROR (demux, ret);}}break;case GST_FLOW_FLUSHING:{GList *iter;for (iter = demux->streams; iter; iter = g_list_next (iter)) {GstAdaptiveDemuxStream *other;other = iter->data;/* 只是停止自身任务，并没有拉其他标志 */gst_task_stop (other->download_task);}}break;default:if (ret <= GST_FLOW_ERROR) {gboolean is_live = gst_adaptive_demux_is_live (demux);GST_WARNING_OBJECT (demux, "Error while downloading fragment");if (++stream->download_error_count > MAX_DOWNLOAD_ERROR_COUNT) {goto download_error;}g_clear_error (&stream->last_error);/* First try to update the playlist for non-live playlists* in case the URIs have changed in the meantime. But only* try it the first time, after that we're going to wait a* a bit to not flood the server */if (stream->download_error_count == 1 && !is_live) {/* TODO hlsdemux had more options to this function (boolean and err) */if (gst_adaptive_demux_update_manifest (demux) == GST_FLOW_OK) {/* Retry immediately, the playlist actually has changed */GST_DEBUG_OBJECT (demux, "Updated the playlist");goto end;}}/* Wait half the fragment duration before retrying */next_download += stream->fragment.duration / 2;
……/* Refetch the playlist now after we waited，点播时 */if (!is_live&& gst_adaptive_demux_update_manifest (demux) == GST_FLOW_OK) {GST_DEBUG_OBJECT (demux, "Updated the playlist");}goto end;}break;}
end_of_manifest:if (G_UNLIKELY (ret == GST_FLOW_EOS)) {if (GST_OBJECT_PARENT (stream->pad) != NULL) {if (demux->next_streams == NULL && demux->prepared_streams == NULL) {GST_DEBUG_OBJECT (stream->src, "Pushing EOS on pad");gst_adaptive_demux_stream_push_event (stream, gst_event_new_eos ());} else {GST_DEBUG_OBJECT (stream->src,"Stream is EOS, but we're switching fragments. Not sending.");}} else {GST_ERROR_OBJECT (demux, "Can't push EOS on non-exposed pad");goto download_error;}}
end:GST_MANIFEST_UNLOCK (demux);GST_LOG_OBJECT (stream->pad, "download loop end");return;
cancelled:{GST_DEBUG_OBJECT (stream->pad, "Stream has been cancelled");goto end;}
download_error:{GstMessage *msg;if (stream->last_error) {gchar *debug = g_strdup_printf ("Error on stream %s:%s",GST_DEBUG_PAD_NAME (stream->pad));msg =gst_message_new_error (GST_OBJECT_CAST (demux), stream->last_error,debug);GST_ERROR_OBJECT (stream->pad, "Download error: %s",stream->last_error->message);g_free (debug);} else {GError *err =g_error_new (GST_RESOURCE_ERROR, GST_RESOURCE_ERROR_NOT_FOUND,_("Couldn't download fragments"));msg =gst_message_new_error (GST_OBJECT_CAST (demux), err,"Fragment downloading has failed consecutive times");g_error_free (err);GST_ERROR_OBJECT (stream->pad,"Download error: Couldn't download fragments, too many failures");}gst_task_stop (stream->download_task);if (stream->src) {GstElement *src = stream->src;stream->src = NULL;GST_MANIFEST_UNLOCK (demux);gst_element_set_locked_state (src, TRUE);gst_element_set_state (src, GST_STATE_NULL);gst_bin_remove (GST_BIN_CAST (demux), src);GST_MANIFEST_LOCK (demux);}gst_element_post_message (GST_ELEMENT_CAST (demux), msg);goto end;}
}
/* 描述一下dash语法层级关系：* MPD->periods->Representations->Segments->fragment* 所以此处更新的是最小级单位。* 函数为获取到下一个要下载fragment的url,range等信息。*/
static GstFlowReturn
gst_dash_demux_stream_update_fragment_info (GstAdaptiveDemuxStream * stream)
{GstDashDemuxStream *dashstream = (GstDashDemuxStream *) stream;GstDashDemux *dashdemux = GST_DASH_DEMUX_CAST (stream->demux);GstClockTime ts;GstMediaFragmentInfo fragment;gboolean isombff;/* 清除fragment的全部信息 */gst_adaptive_demux_stream_fragment_clear (&stream->fragment);/* mp4的点播 */isombff = gst_mpd_client_has_isoff_ondemand_profile (dashdemux->client);/* Reset chunk size if any */stream->fragment.chunk_size = 0;dashstream->current_fragment_keyframe_distance = GST_CLOCK_TIME_NONE;/* 当前点播fragment处理header和index的信息，获取header和index的url */if (GST_ADAPTIVE_DEMUX_STREAM_NEED_HEADER (stream) && isombff) {gst_dash_demux_stream_update_headers_info (stream);/* sidx entries may not be available in here */if (stream->fragment.index_uri&& dashstream->sidx_position != GST_CLOCK_TIME_NONE) {/* request only the index to be downloaded as we need to reposition the* stream to a subsegment */return GST_FLOW_OK;}}/* moof_sync_samples用来保存当前fragment的关键帧信息，播放方式为TRICK 关键帧*/if (dashstream->moof_sync_samples&& GST_ADAPTIVE_DEMUX_IN_TRICKMODE_KEY_UNITS (dashdemux)) {GstDashStreamSyncSample *sync_sample =&g_array_index (dashstream->moof_sync_samples, GstDashStreamSyncSample,dashstream->current_sync_sample);gst_mpd_client_get_next_fragment (dashdemux->client, dashstream->index,&fragment);if (isombff && dashstream->sidx_position != GST_CLOCK_TIME_NONE&& SIDX (dashstream)->entries) {GstSidxBoxEntry *entry = SIDX_CURRENT_ENTRY (dashstream);dashstream->current_fragment_timestamp = fragment.timestamp = entry->pts;dashstream->current_fragment_duration = fragment.duration =entry->duration;} else {dashstream->current_fragment_timestamp = fragment.timestamp;dashstream->current_fragment_duration = fragment.duration;}dashstream->current_fragment_keyframe_distance =fragment.duration / dashstream->moof_sync_samples->len;dashstream->actual_position =fragment.timestamp +dashstream->current_sync_sample *dashstream->current_fragment_keyframe_distance;if (stream->segment.rate < 0.0)dashstream->actual_position +=dashstream->current_fragment_keyframe_distance;dashstream->actual_position =MIN (dashstream->actual_position,fragment.timestamp + fragment.duration);stream->fragment.uri = fragment.uri;stream->fragment.timestamp = GST_CLOCK_TIME_NONE;stream->fragment.duration = GST_CLOCK_TIME_NONE;stream->fragment.range_start = sync_sample->start_offset;stream->fragment.range_end = sync_sample->end_offset;GST_DEBUG_OBJECT (stream->pad, "Actual position %" GST_TIME_FORMAT,GST_TIME_ARGS (dashstream->actual_position));return GST_FLOW_OK;}/* 获取下一次播放的时间戳，包含SegmentList的MPD文件，从stream->segments就可以得到，否则如SegmentTemplate用index和duration计算出来 */if (gst_mpd_client_get_next_fragment_timestamp (dashdemux->client,dashstream->index, &ts)) {if (GST_ADAPTIVE_DEMUX_STREAM_NEED_HEADER (stream)) {gst_adaptive_demux_stream_fragment_clear (&stream->fragment);/* 获取header的uri信息，包括两个部分：* 一部分是初始化数据uri，如mp4的init segment在文件中的位置。用Initialization描述;* 另外一部分是SegmentBase描述的index的range,如mp4的sidx信息在URI所代表文件中的位置，如：* <Representation bandwidth="320000" id="2">*  <BaseURL>ED-CM-5.1-DVD_length_fixed-653s-6-heaac-320000bps_seg.mp4</BaseURL>*  <SegmentBase indexRange="607-2210"> // 指出段的索引信息，即sidx在上述URL代表的文件中的字节范围 *  <Initialization range="0-606"/> // 指出初始化信息在上述URL代表的文件中的字节范围*  </SegmentBase>* </Representation>* 文件ED-CM-5.1-DVD_length_fixed-653s-6-heaac-320000bps_seg.mp4中信息如下：607字节前面就是ftyp moov这些元素。* Segment Index Box * Start offset 607 (0X0000025F) * Box size 1604 (0X00000644) * Box type sidx (0X73696478) * Version 0 (0X00000000) * Flags 0 (0X00000000) */gst_dash_demux_stream_update_headers_info (stream);}/* 获取数据信息 */gst_mpd_client_get_next_fragment (dashdemux->client, dashstream->index,&fragment);/* fragment中包括音视频文件URI,初始化数据URI, 索引表URI三种类型的URI。 */stream->fragment.uri = fragment.uri;/* If mpd does not specify indexRange (i.e., null index_uri),* sidx entries may not be available until download it sidx_position代表上一个个索引项的PTS */if (isombff && dashstream->sidx_position != GST_CLOCK_TIME_NONE&& SIDX (dashstream)->entries) {GstSidxBoxEntry *entry = SIDX_CURRENT_ENTRY (dashstream);/* entry里面的size代表了ES数据的长度，* entry->offset是第一笔ES数据开始，到前一个entry描述的数据长度总和，即cumulative_entry_size，初值为0*/stream->fragment.range_start =dashstream->sidx_base_offset + entry->offset;dashstream->actual_position = stream->fragment.timestamp = entry->pts;dashstream->current_fragment_timestamp = stream->fragment.timestamp =entry->pts;dashstream->current_fragment_duration = stream->fragment.duration =entry->duration;if (stream->demux->segment.rate < 0.0) {stream->fragment.range_end =stream->fragment.range_start + entry->size - 1;/* 后退时的处理 */dashstream->actual_position += entry->duration;} else {stream->fragment.range_end = fragment.range_end;}} else {dashstream->actual_position = stream->fragment.timestamp =fragment.timestamp;dashstream->current_fragment_timestamp = fragment.timestamp;dashstream->current_fragment_duration = stream->fragment.duration =fragment.duration;if (stream->demux->segment.rate < 0.0)dashstream->actual_position += fragment.duration;stream->fragment.range_start =MAX (fragment.range_start, dashstream->sidx_base_offset);stream->fragment.range_end = fragment.range_end;}GST_DEBUG_OBJECT (stream->pad, "Actual position %" GST_TIME_FORMAT,GST_TIME_ARGS (dashstream->actual_position));return GST_FLOW_OK;}return GST_FLOW_EOS;
}
/** 更新Dash播放列表,调用如下函数前调用gst_adaptive_demux_update_manifest_default下载列表数据。* dash每次都要生成新的new_client，所以每次更新都要重新产生流。
*/
static GstFlowReturn
gst_dash_demux_update_manifest_data (GstAdaptiveDemux * demux,GstBuffer * buffer)
{GstDashDemux *dashdemux = GST_DASH_DEMUX_CAST (demux);GstMPDClient *new_client = NULL;GstMapInfo mapinfo;GST_DEBUG_OBJECT (demux, "Updating manifest file from URL");/* parse the manifest file */new_client = gst_mpd_client_new ();gst_mpd_client_set_uri_downloader (new_client, demux->downloader);new_client->mpd_uri = g_strdup (demux->manifest_uri);new_client->mpd_base_uri = g_strdup (demux->manifest_base_uri);gst_buffer_map (buffer, &mapinfo, GST_MAP_READ);if (gst_mpd_client_parse (new_client, (gchar *) mapinfo.data, mapinfo.size)) {const gchar *period_id;guint period_idx;GList *iter;GList *streams_iter;GList *streams;/* prepare the new manifest and try to transfer the stream position* status from the old manifest client  */GST_DEBUG_OBJECT (demux, "Updating manifest");/* 通过period_idx获取id,这个id有可能不存在，存在的话，在Presentation中必唯一,一个带ID的例子如下：* <Period id="P1" duration="PT30.000S">*   <AdaptationSet segmentAlignment="true" lang="und" subsegmentAlignment="true" subsegmentStartsWithSAP="1">…… </AdaptationSet>*   <AdaptationSet segmentAlignment="true" maxWidth="1920" maxHeight="1080" maxFrameRate="30" par="16:9" lang="und" subsegmentAlignment="true" subsegmentStartsWithSAP="1">……</AdaptationSet>* </Period>* <Period id="P2" duration="PT300.000S">*   <AdaptationSet segmentAlignment="true" lang="und" subsegmentAlignment="true" subsegmentStartsWithSAP="1">……</AdaptationSet>*   <AdaptationSet segmentAlignment="true" maxWidth="1920" maxHeight="1080" maxFrameRate="30" par="16:9" lang="und" subsegmentAlignment="true" subsegmentStartsWithSAP="1">……</AdaptationSet>* </Period>*/period_id = gst_mpd_client_get_period_id (dashdemux->client);period_idx = gst_mpd_client_get_period_index (dashdemux->client);/* setup video, audio and subtitle streams, starting from current Period，* 实际上是重新创建了periods结构，包括period的编号，开始时间，时长等信息*/if (!gst_mpd_client_setup_media_presentation (new_client, -1,(period_id ? -1 : period_idx), period_id)) {/* TODO */}/* 选择period */if (period_id) {/* 使用和旧的client中相同的ID，如果新的client中无此ID，则播放出播放。 */if (!gst_mpd_client_set_period_id (new_client, period_id)) {GST_DEBUG_OBJECT (demux, "Error setting up the updated manifest file");gst_mpd_client_free (new_client);gst_buffer_unmap (buffer, &mapinfo);return GST_FLOW_EOS;}} else {if (!gst_mpd_client_set_period_index (new_client, period_idx)) {GST_DEBUG_OBJECT (demux, "Error setting up the updated manifest file");gst_mpd_client_free (new_client);gst_buffer_unmap (buffer, &mapinfo);return GST_FLOW_EOS;}}/* 根据period获取Adaptation_set，* 为Adaptation_set中的每一个Representation集合中的最小带宽的Representation创建GstActiveStream * 放到new_client->active_streams链表中。*/if (!gst_dash_demux_setup_mpdparser_streams (dashdemux, new_client)) {GST_ERROR_OBJECT (demux, "Failed to setup streams on manifest " "update");gst_mpd_client_free (new_client);gst_buffer_unmap (buffer, &mapinfo);return GST_FLOW_ERROR;}/* If no pads have been exposed yet, need to use those */streams = NULL;if (demux->streams == NULL) {if (demux->prepared_streams) {streams = demux->prepared_streams;}} else {streams = demux->streams;}/* update the streams to play from the next segment，查找到对应的时间点，然后将当前的stream置成active stream */for (iter = streams, streams_iter = new_client->active_streams;iter && streams_iter;iter = g_list_next (iter), streams_iter = g_list_next (streams_iter)) {GstDashDemuxStream *demux_stream = iter->data;GstActiveStream *new_stream = streams_iter->data;GstClockTime ts;if (!new_stream) {GST_DEBUG_OBJECT (demux,"Stream of index %d is missing from manifest update",demux_stream->index);gst_mpd_client_free (new_client);gst_buffer_unmap (buffer, &mapinfo);return GST_FLOW_EOS;}if (gst_mpd_client_get_next_fragment_timestamp (dashdemux->client,demux_stream->index, &ts)|| gst_mpd_client_get_last_fragment_timestamp_end (dashdemux->client,demux_stream->index, &ts)) {/* Due to rounding when doing the timescale conversions it might happen* that the ts falls back to a previous segment, leading the same data* to be downloaded twice. We try to work around this by always adding* 10 microseconds to get back to the correct segment. The errors are* usually on the order of nanoseconds so it should be enough.*//* _get_next_fragment_timestamp() returned relative timestamp to* corresponding period start, but _client_stream_seek expects absolute* MPD time. */ts += gst_mpd_client_get_period_start_time (dashdemux->client);GST_DEBUG_OBJECT (GST_ADAPTIVE_DEMUX_STREAM_PAD (demux_stream),"Current position: %" GST_TIME_FORMAT ", updating to %"GST_TIME_FORMAT, GST_TIME_ARGS (ts),GST_TIME_ARGS (ts + (10 * GST_USECOND)));ts += 10 * GST_USECOND;gst_mpd_client_stream_seek (new_client, new_stream,demux->segment.rate >= 0, 0, ts, NULL);}demux_stream->active_stream = new_stream;}gst_mpd_client_free (dashdemux->client);/* 使用新的client,但此处并没有移除旧的流和exposed新的流 */dashdemux->client = new_client;GST_DEBUG_OBJECT (demux, "Manifest file successfully updated");if (dashdemux->clock_drift) {gst_dash_demux_poll_clock_drift (dashdemux);}} else {/* In most cases, this will happen if we set a wrong url in the* source element and we have received the 404 HTML response instead of* the manifest */GST_WARNING_OBJECT (demux, "Error parsing the manifest.");gst_mpd_client_free (new_client);gst_buffer_unmap (buffer, &mapinfo);return GST_FLOW_ERROR;}gst_buffer_unmap (buffer, &mapinfo);return GST_FLOW_OK;
}

gst_adaptive_demux_updates_loop任务

static void
gst_adaptive_demux_updates_loop (GstAdaptiveDemux * demux)
{GstClockTime next_update;GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);/* Loop for updating of the playlist. This periodically checks if* the playlist is updated and does so, then signals the streaming* thread in case it can continue downloading now. *//* block until the next scheduled update or the signal to quit this thread */GST_DEBUG_OBJECT (demux, "Started updates task");GST_MANIFEST_LOCK (demux);/* 列表的更新时间间隔,位于MPD标签中：* <MPD xmlns="urn:mpeg:dash:schema:mpd:2011" minBufferTime="PT10.000S" type="dynamic" publishTime="2023-03-15T09:09:54Z" availabilityStartTime="2023-03-15T09:09:12.410Z" timeShiftBufferDepth="PT0H0M0.000S" minimumUpdatePeriod="PT0H0M10.000S" maxSegmentDuration="PT0H0M14.306S" profiles="urn:mpeg:dash:profile:isoff-live:2011">……</MPD> * 就是minimumUpdatePeriod,无此选项默认30分钟更新一次，注意一下这个默认参数是否过大。*/next_update =gst_adaptive_demux_get_monotonic_time (demux) +klass->get_manifest_update_interval (demux) * GST_USECOND;/* Updating playlist only needed for live playlists */while (gst_adaptive_demux_is_live (demux)) {GstFlowReturn ret = GST_FLOW_OK;/* Wait here until we should do the next update or we're cancelled */GST_DEBUG_OBJECT (demux, "Wait for next playlist update");GST_MANIFEST_UNLOCK (demux);g_mutex_lock (&demux->priv->updates_timed_lock);if (demux->priv->stop_updates_task) {g_mutex_unlock (&demux->priv->updates_timed_lock);goto quit;}gst_adaptive_demux_wait_until (demux->realtime_clock,&demux->priv->updates_timed_cond,&demux->priv->updates_timed_lock, next_update);g_mutex_unlock (&demux->priv->updates_timed_lock);g_mutex_lock (&demux->priv->updates_timed_lock);if (demux->priv->stop_updates_task) {g_mutex_unlock (&demux->priv->updates_timed_lock);goto quit;}g_mutex_unlock (&demux->priv->updates_timed_lock);GST_MANIFEST_LOCK (demux);GST_DEBUG_OBJECT (demux, "Updating playlist");/* 更新播放列表 */ret = gst_adaptive_demux_update_manifest (demux);if (ret == GST_FLOW_EOS) {} else if (ret != GST_FLOW_OK) {/* update_failed_count is used only here, no need to protect it * 更新失败小于既定次数，等待一段时间后更新。*/demux->priv->update_failed_count++;if (demux->priv->update_failed_count <= DEFAULT_FAILED_COUNT) {GST_WARNING_OBJECT (demux, "Could not update the playlist, flow: %s",gst_flow_get_name (ret));next_update = gst_adaptive_demux_get_monotonic_time (demux)+ klass->get_manifest_update_interval (demux) * GST_USECOND;} else {GST_ELEMENT_ERROR (demux, STREAM, FAILED,(_("Internal data stream error.")), ("Could not update playlist"));GST_DEBUG_OBJECT (demux, "Stopped updates task because of error");gst_task_stop (demux->priv->updates_task);GST_MANIFEST_UNLOCK (demux);goto end;}} else {/* 更新成功后，唤醒下载任务 */GST_DEBUG_OBJECT (demux, "Updated playlist successfully");demux->priv->update_failed_count = 0;next_update =gst_adaptive_demux_get_monotonic_time (demux) +klass->get_manifest_update_interval (demux) * GST_USECOND;/* Wake up download tasks */g_mutex_lock (&demux->priv->manifest_update_lock);g_cond_broadcast (&demux->priv->manifest_cond);g_mutex_unlock (&demux->priv->manifest_update_lock);}}GST_MANIFEST_UNLOCK (demux);
quit:{GST_DEBUG_OBJECT (demux, "Stop updates task request detected.");}
end:{return;}
}

_src_chain函数

chain的数据来自于source下载的推流，流程如下：

_src_chain 行 2707 gst-plugins-bad\gst-libs\gst\adaptivedemux\gstadaptivedemux.c(2707)

gst_pad_chain_data_unchecked 行 4449 gstreamer\gst\gstpad.c(4449)

gst_pad_push_data 行 4714 gstreamer\gst\gstpad.c(4714)

……

gst_queue_loop 行 1542 gstreamer\plugins\elements\gstqueue.c(1542)

static GstFlowReturn
_src_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
{GstAdaptiveDemuxStream *stream;GstAdaptiveDemux *demux;GstAdaptiveDemuxClass *klass;GstFlowReturn ret = GST_FLOW_OK;
……/* do not make any changes if the stream is cancelled */g_mutex_lock (&stream->fragment_download_lock);if (G_UNLIKELY (stream->cancelled)) {g_mutex_unlock (&stream->fragment_download_lock);gst_buffer_unref (buffer);ret = stream->last_ret = GST_FLOW_FLUSHING;GST_MANIFEST_UNLOCK (demux);return ret;}g_mutex_unlock (&stream->fragment_download_lock);/* 开始_stream_download_fragment()的时候设置为TRUE* 数据可能是正在下载的初始化信息或者索引信息。*/if (stream->starting_fragment) {GstClockTime offset =gst_adaptive_demux_stream_get_presentation_offset (demux, stream);GstClockTime period_start =gst_adaptive_demux_get_period_start_time (demux);stream->starting_fragment = FALSE;if (klass->start_fragment) {/* dash是gst_dash_demux_stream_fragment_start这个函数 */if (!klass->start_fragment (demux, stream)) {ret = GST_FLOW_ERROR;goto error;}}GST_BUFFER_PTS (buffer) = stream->fragment.timestamp;if (GST_BUFFER_PTS_IS_VALID (buffer))GST_BUFFER_PTS (buffer) += offset;
……if (GST_BUFFER_PTS_IS_VALID (buffer)) {GST_ADAPTIVE_DEMUX_SEGMENT_LOCK (demux);/* position是时间戳，不是字节地址 */stream->segment.position = GST_BUFFER_PTS (buffer);/* Convert from position inside the stream's segment to the demuxer's* segment, they are not necessarily the same */if (stream->segment.position - offset + period_start >demux->segment.position)demux->segment.position =stream->segment.position - offset + period_start;GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux);}} else {/* 只有第一个buffer的pts才配置成段的起始PTS。 */GST_BUFFER_PTS (buffer) = GST_CLOCK_TIME_NONE;}/* 激活source前在download_uri()中设置为TRUE。 */if (stream->downloading_first_buffer) {gint64 chunk_size = 0;stream->downloading_first_buffer = FALSE;/* 第一个buffer不是下载索引和初始化信息，代表下载的是媒体数据，计算bitrate */if (!stream->downloading_header && !stream->downloading_index) {/* If this is the first buffer of a fragment (not the headers or index)* and we don't have a birate from the sub-class, then see if we* can work it out from the fragment size and duration */if (stream->fragment.bitrate == 0 &&stream->fragment.duration != 0 &&gst_element_query_duration (stream->uri_handler, GST_FORMAT_BYTES,&chunk_size) && chunk_size != -1) {guint bitrate = MIN (G_MAXUINT, gst_util_uint64_scale (chunk_size,8 * GST_SECOND, stream->fragment.duration));
……stream->fragment.bitrate = bitrate;}if (stream->fragment.bitrate) {stream->bitrate_changed = TRUE;} else {GST_WARNING_OBJECT (demux, "Bitrate for fragment not available");}}}stream->download_total_bytes += gst_buffer_get_size (buffer);/* gst_dash_demux_data_received，主要包括这几方面的工作：* 1. 将buffer推到adapter中。* 2. 从adapter中取buffer解释初始化头部信息和索引表信息。* 3. 将流推给下游的demux。*/ret = klass->data_received (demux, stream, buffer);if (ret == GST_FLOW_FLUSHING) {/* do not make any changes if the stream is cancelled */g_mutex_lock (&stream->fragment_download_lock);if (G_UNLIKELY (stream->cancelled)) {g_mutex_unlock (&stream->fragment_download_lock);GST_MANIFEST_UNLOCK (demux);return ret;}g_mutex_unlock (&stream->fragment_download_lock);}if (ret != GST_FLOW_OK) {gboolean finished = FALSE;if (ret < GST_FLOW_EOS) {GST_ELEMENT_FLOW_ERROR (demux, ret);/* TODO push this on all pads */gst_pad_push_event (stream->pad, gst_event_new_eos ());} else {GST_DEBUG_OBJECT (stream->pad, "stream stopped, reason %s",gst_flow_get_name (ret));}if (ret == (GstFlowReturn) GST_ADAPTIVE_DEMUX_FLOW_SWITCH) {ret = GST_FLOW_EOS;       /* return EOS to make the source stop */} else if (ret == GST_ADAPTIVE_DEMUX_FLOW_END_OF_FRAGMENT) {/* Behaves like an EOS event from upstream */stream->fragment.finished = TRUE;/* 调用的是gst_dash_demux_stream_fragment_finished */ret = klass->finish_fragment (demux, stream);if (ret == (GstFlowReturn) GST_ADAPTIVE_DEMUX_FLOW_SWITCH) {ret = GST_FLOW_EOS;     /* return EOS to make the source stop */} else if (ret != GST_FLOW_OK) {goto error;}finished = TRUE;}gst_adaptive_demux_stream_fragment_download_finish (stream, ret, NULL);if (finished)ret = GST_FLOW_EOS;}
error:GST_MANIFEST_UNLOCK (demux);return ret;
}
static GstFlowReturn
gst_dash_demux_stream_fragment_finished (GstAdaptiveDemux * demux,GstAdaptiveDemuxStream * stream)
{GstDashDemux *dashdemux = GST_DASH_DEMUX_CAST (demux);GstDashDemuxStream *dashstream = (GstDashDemuxStream *) stream;/* We need to mark every first buffer of a key unit as discont,* and also every first buffer of a moov and moof. This ensures* that qtdemux takes note of our buffer offsets for each of those* buffers instead of keeping track of them itself from the first* buffer. We need offsets to be consistent between moof and mdat*/if (dashstream->is_isobmff && dashdemux->allow_trickmode_key_units&& GST_ADAPTIVE_DEMUX_IN_TRICKMODE_KEY_UNITS (demux)&& dashstream->active_stream->mimeType == GST_STREAM_VIDEO)stream->discont = TRUE;/* 有索引表并且完整，TRICK，点播，直接选片断 */if (!(dashstream->moof_sync_samples&& GST_ADAPTIVE_DEMUX_IN_TRICKMODE_KEY_UNITS (dashdemux))&& gst_mpd_client_has_isoff_ondemand_profile (dashdemux->client)&& dashstream->sidx_parser.status == GST_ISOFF_SIDX_PARSER_FINISHED) {/* fragment is advanced on data_received when byte limits are reached */if (dashstream->pending_seek_ts != GST_CLOCK_TIME_NONE) {if (SIDX (dashstream)->entry_index < SIDX (dashstream)->entries_count)return GST_FLOW_OK;} else if (gst_dash_demux_stream_has_next_subfragment (stream)) {return GST_FLOW_OK;}}/* 下载的非码流信息直接返回 */if (G_UNLIKELY (stream->downloading_header || stream->downloading_index))return GST_FLOW_OK;/* 调用到 gst_adaptive_demux_stream_advance_fragment_unlocked */return gst_adaptive_demux_stream_advance_fragment (demux, stream,stream->fragment.duration);
}
/* must be called with manifest_lock taken */
GstFlowReturn
gst_adaptive_demux_stream_advance_fragment_unlocked (GstAdaptiveDemux * demux,GstAdaptiveDemuxStream * stream, GstClockTime duration)
{GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);GstFlowReturn ret;
……stream->download_error_count = 0;g_clear_error (&stream->last_error);
……/* Don't update to the end of the segment if in reverse playback */GST_ADAPTIVE_DEMUX_SEGMENT_LOCK (demux);if (GST_CLOCK_TIME_IS_VALID (duration) && demux->segment.rate > 0) {/* presentation相对于period start的时间偏移 */GstClockTime offset =gst_adaptive_demux_stream_get_presentation_offset (demux, stream);GstClockTime period_start =gst_adaptive_demux_get_period_start_time (demux);/* 更新segment时间，下一次起播的开始时间 */stream->segment.position += duration;/* Convert from position inside the stream's segment to the demuxer's* segment, they are not necessarily the same */if (stream->segment.position - offset + period_start >demux->segment.position)demux->segment.position =stream->segment.position - offset + period_start;}GST_ADAPTIVE_DEMUX_SEGMENT_UNLOCK (demux);/* When advancing with a non 1.0 rate on live streams, we need to check* the live seeking range again to make sure we can still advance to* that position */if (demux->segment.rate != 1.0 && gst_adaptive_demux_is_live (demux)) {if (!gst_adaptive_demux_stream_in_live_seek_range (demux, stream))ret = GST_FLOW_EOS;elseret = klass->stream_advance_fragment (stream);} else if (gst_adaptive_demux_is_live (demux)|| gst_adaptive_demux_stream_has_next_fragment (demux, stream)) {ret = klass->stream_advance_fragment (stream);} else {ret = GST_FLOW_EOS;}/* 下一个片断开始下载时间 */stream->download_start_time =GST_TIME_AS_USECONDS (gst_adaptive_demux_get_monotonic_time (demux));if (ret == GST_FLOW_OK) {/* 数据下载的时候会有probe函数去查看下载的数据，由此计算出bitrate *//* gst_dash_demux_stream_select_bitrate根据当前下载数据的bitrate，计算出带宽。* 然后将当前active_stream的cur_representation更换成新的representation，* segments信息会根据新的representation重建。* 成功之后会重新要求清理adapter，构建索引表这些信息。* 注意，representation的切换并没有切流，所以不会引起pad的释放与重建。*/if (gst_adaptive_demux_stream_select_bitrate (demux, stream,gst_adaptive_demux_stream_update_current_bitrate (demux, stream))) {stream->need_header = TRUE;ret = (GstFlowReturn) GST_ADAPTIVE_DEMUX_FLOW_SWITCH;}/* the subclass might want to switch pads，* 如果dash或者hls要求切换pad，那么，当前流下载完成或者被取消以后，切换。*/if (G_UNLIKELY (demux->next_streams)) {GList *iter;gboolean can_expose = TRUE;gst_task_stop (stream->download_task);ret = GST_FLOW_EOS;for (iter = demux->streams; iter; iter = g_list_next (iter)) {/* Only expose if all streams are now cancelled or finished downloading */GstAdaptiveDemuxStream *other = iter->data;if (other != stream) {g_mutex_lock (&other->fragment_download_lock);can_expose &= (other->cancelled == TRUE|| other->download_finished == TRUE);g_mutex_unlock (&other->fragment_download_lock);}}if (can_expose) {GST_DEBUG_OBJECT (demux, "Subclass wants new pads ""to do bitrate switching");gst_adaptive_demux_prepare_streams (demux, FALSE);gst_adaptive_demux_start_tasks (demux, TRUE);} else {GST_LOG_OBJECT (demux, "Not switching yet - ongoing downloads");}}}return ret;
}

gst_dash_demux_process_manifest

static gboolean
gst_adaptive_demux_sink_event (GstPad * pad, GstObject * parent,GstEvent * event)
{GstAdaptiveDemux *demux = GST_ADAPTIVE_DEMUX_CAST (parent);gboolean ret;switch (event->type) {
……}/* 表示Manifest下载结束,sink eos。媒体数据下载结束的处理函数是_src_event */case GST_EVENT_EOS:{
……/* demux->priv->input_adapter用来保存输入的sinkpad chain的数据,* adaptive_demux输入MPD数据，输出source stream pad */available = gst_adapter_available (demux->priv->input_adapter);if (available == 0) {ret = gst_pad_event_default (pad, parent, event);……return ret;}/* Need to get the URI to use it as a base to generate the fragment's* uris */query = gst_query_new_uri ();query_res = gst_pad_peer_query (pad, query);if (query_res) {gchar *uri, *redirect_uri;gboolean permanent;gst_query_parse_uri (query, &uri);gst_query_parse_uri_redirection (query, &redirect_uri);gst_query_parse_uri_redirection_permanent (query, &permanent);if (permanent && redirect_uri) {demux->manifest_uri = redirect_uri;demux->manifest_base_uri = NULL;g_free (uri);} else {demux->manifest_uri = uri;demux->manifest_base_uri = redirect_uri;}} else {GST_WARNING_OBJECT (demux, "Upstream URI query failed.");}gst_query_unref (query);/* Let the subclass parse the manifest */manifest_buffer =gst_adapter_take_buffer (demux->priv->input_adapter, available);/* 调用了gst_dash_demux_process_manifest */if (!demux_class->process_manifest (demux, manifest_buffer)) {/* In most cases, this will happen if we set a wrong url in the* source element and we have received the 404 HTML response instead of* the manifest */GST_ELEMENT_ERROR (demux, STREAM, DECODE, ("Invalid manifest."),(NULL));ret = FALSE;} else {g_atomic_int_set (&demux->priv->have_manifest, TRUE);}gst_buffer_unref (manifest_buffer);
……if (ret) {/* Send duration message */if (!gst_adaptive_demux_is_live (demux)) {GstClockTime duration = demux_class->get_duration (demux);if (duration != GST_CLOCK_TIME_NONE) {GST_DEBUG_OBJECT (demux,"Sending duration message : %" GST_TIME_FORMAT,GST_TIME_ARGS (duration));gst_element_post_message (GST_ELEMENT (demux),gst_message_new_duration_changed (GST_OBJECT (demux)));} else {GST_DEBUG_OBJECT (demux,"media duration unknown, can not send the duration message");}}if (demux->next_streams) {/* prepared_streams为空 */gst_adaptive_demux_prepare_streams (demux,gst_adaptive_demux_is_live (demux));gst_adaptive_demux_start_tasks (demux, TRUE);gst_adaptive_demux_start_manifest_update_task (demux);} else {/* no streams */GST_WARNING_OBJECT (demux, "No streams created from manifest");GST_ELEMENT_ERROR (demux, STREAM, DEMUX,(_("This file contains no playable streams.")),("No known stream formats found at the Manifest"));ret = FALSE;}}GST_MANIFEST_UNLOCK (demux);GST_API_UNLOCK (demux);gst_event_unref (event);return ret;}……}return gst_pad_event_default (pad, parent, event);
}static gboolean
gst_dash_demux_process_manifest (GstAdaptiveDemux * demux, GstBuffer * buf)
{GstDashDemux *dashdemux = GST_DASH_DEMUX_CAST (demux);gboolean ret = FALSE;gchar *manifest;GstMapInfo mapinfo;/* client是GstMPDClient结构，包括了MPD uri，active_streams,downloader，mpd_root_node等成员，* mpd_root_node包含所有的MPD信息，一并释放，gst_mpdparser_free_active_stream似乎并没有将成员置空。*/if (dashdemux->client)gst_mpd_client_free (dashdemux->client);dashdemux->client = gst_mpd_client_new ();gst_mpd_client_set_uri_downloader (dashdemux->client, demux->downloader);dashdemux->client->mpd_uri = g_strdup (demux->manifest_uri);dashdemux->client->mpd_base_uri = g_strdup (demux->manifest_base_uri);
……if (gst_buffer_map (buf, &mapinfo, GST_MAP_READ)) {manifest = (gchar *) mapinfo.data;if (gst_mpd_client_parse (dashdemux->client, manifest, mapinfo.size)) {/* 根据mpd_root_node包含所有的MPD信息，重建period等信息 */if (gst_mpd_client_setup_media_presentation (dashdemux->client, 0, 0,NULL)) {ret = TRUE;} else {GST_ELEMENT_ERROR (demux, STREAM, DECODE,("Incompatible manifest file."), (NULL));}}gst_buffer_unmap (buf, &mapinfo);} else {GST_WARNING_OBJECT (demux, "Failed to map manifest buffer");}/* * 1. 清除所有的active stream 链表* 2. 获取新的adaptation_set。* 3. 建立GstActiveStream，配置GstActiveStream的cur_adapt_set，选取最低码率的representation* 4. 配置GstActiveStream的representation一系列参数。* 5. 重构新的source流。*/if (ret)ret = gst_dash_demux_setup_streams (demux);return ret;
}

gst_system_clock_async_thread任务

作为定时任务，时间到了调用gst_adaptive_demux_clock_callback

DASH直播推流方式

写成bat文件，内容如下：

start mp4box -profile live -dash-live 10000 -frag 5000 H264_AAC_0.MP4 H264_AAC_1.MP4 H264_AAC_2.MP4 -out MP4_LIVE_SegmentTemplate -segment-name segment\%s_ -mpd-refresh 10 -min-buffer 10000 pause
del *.mpd
del MP4_LIVE_SegmentTemplate*
del /s /q segment\*
pause

参数详细分析参考链接：https://www.jianshu.com/p/5e97f93b00f1