gstreamer pipeline梳理及插件新增方案-北京尧图网络

一整体播放的pipeline1,playbin 主要做两件事情收到 uridecodebin 的 raw padpad-added→ 接入 input-selector 的 sink_%upad_added_cb收到 uridecodebin 的 no-more-pads → request playsink pad 并链接 combiner src → playsink sinkno_more_pads_cb然后 unblock combinerplaybin (GstPipeline)├─ playsink├─ uridecodebin(main)├─ uridecodebin(sub1) / sub2 ... (TCL_PATCH 多外挂字幕)├─ input-selector(audio) ─┐├─ input-selector(video) ─┼─ playsink (request pad)└─ input-selector(text) ─┘2pipeline 详细流程playbin├─ uridecodebin (有时叫 sourcebin/uridecodebin3取决于实现)│ ├─ source (按 URI scheme 选择)│ │ ├─ filesrc / souphttpsrc / rtspsrc / ...│ │ └─ (可选) queue2 / downloadbuffer / retry 等│ ├─ typefind│ ├─ demux (按容器选择)│ │ ├─ qtdemux / matroskademux / tsdemux / ...│ │ └─ 输出audio_%u / video_%u / subtitle_%u (压缩流 pad)│ └─ decodebin (GstDecodeBin / decodebin3)│ ├─ (每条流一条 decode chain)│ │ ├─ queue│ │ ├─ parser (可选h264parse/aacparse/…)│ │ ├─ decoder (avdec_h264/faad/… 或硬解)│ │ ├─ convert (audioconvert/videoconvert)│ │ ├─ resample/scale (audioresample/videoscale可选)│ │ └─ 输出audio/x-raw、video/x-raw、text/x-raw│ └─ 暴露 padsrc_%u (raw)│├─ audio-bin (playbin 内部组的音频输出链)│ ├─ queue│ ├─ audioconvert│ ├─ audioresample│ ├─ (可选) volume / level / capsfilter│ └─ audiosink (autoaudiosink / pulsesink / alsasink / openslessink / ... )│├─ video-bin│ ├─ queue│ ├─ videoconvert│ ├─ videoscale│ ├─ (可选) subtitleoverlay / videofilter│ └─ videosink (autovideosink / glimagesink / waylandsink / ... )│└─ text/subtitle-bin (有的版本并入 video-bin)├─ subparse / ttmlparse / ...└─ textoverlay(叠到视频) 或 appsink(单独输出)3流程描述1. 初始化和URI设置应用程序通过g_object_set(playbin, uri, ...)设置URI。gst_play_bin_set_uri()将URI存入next_group标记valid TRUE。2. 状态变化到PAUSEDgst_play_bin_change_state处理READY→PAUSED时调用setup_next_source()。setup_next_source()检查next_group是否有效若有效则如果当前有正在播放的curr_group先解除其与playsink的连接deactivate_group。激活next_groupactivate_group创建/获取uridecodebin和suburidecodebin。设置其属性URI、缓冲参数等。连接pad-added、no-more-pads等信号。将uridecodebin加入bin并设置状态为目标状态PAUSED。将curr_group指向next_group原curr_group变为新的next_group双缓冲交换。3. 解码流水线构建当uridecodebin发现新pad时触发pad_added_cb。回调中根据pad的caps判断媒体类型音频/视频/字幕找到对应的GstSourceCombine。如果该类型的组合器尚未创建则创建input-selector或使用用户自定义组合器。从组合器请求一个sink pad如sink_%u并链接uridecodebin的src pad到此sink pad。将sink pad存入对应的channels数组。如果组合器支持active-pad属性连接notify::active-pad信号以便在流切换时更新current-*属性。当所有pad添加完毕no-more-pads_cb被调用为每个有数据的媒体类型向playsink请求一个对应的sink pad如音频sink pad并与组合器的src pad链接。解阻塞组合器的src pad之前被BLOCK探针阻塞防止数据过早流出。发送async-done表示管道已准备好。4. 流选择与切换应用程序通过设置current-video/current-audio/current-text属性切换流。gst_play_bin_set_current_*_stream()获取目标流的sink pad然后向组合器发送自定义事件如playsink-custom-video-flush以触发刷新。设置组合器的active-pad属性为目标pad。组合器的notify::active-pad回调会更新playbin的current-*属性并可能发送刷新完成事件。对于字幕流还会处理特殊的阻塞/解阻塞逻辑因为字幕可能来自独立的suburidecodebin。5. 缓冲与网络流当uridecodebin内部队列缓冲时会向上游发送GST_MESSAGE_BUFFERING消息。gst_play_bin_handle_message中特殊处理此类消息如果当前正在切换组stream_changed_pending为TRUE则将缓冲消息暂存待切换完成后重新发送。否则直接传递给父类处理最终由应用程序根据百分比控制状态PAUSED/PLAYING。6. 标签处理当组合器的sink pad上的tags属性变化时notify_tags_cb被调用发出*-tags-changed信号。应用程序可通过get-video-tags等action信号获取具体标签。7. 无缝切换Gapless当当前URI即将结束时uridecodebin发出drained信号触发drained_cb。drained_cb发出about-to-finish信号通知应用程序设置下一个URI通过uri属性。然后调用setup_next_source()若新URI已设置则开始预滚下一个片段实现无缝衔接。二uridecodebin 的pipeline1第一类都是raw 源文件不需要创建decodebin和typefind 链路setup_source() → analyse_source() → expose_decoded_pad() → new_decoded_pad_added_cb()uridecodebin└─ source (例如 v4l2src / audiotestsrc / 某些自带解码输出的源)└─ (optional) queue2 # decoder-need_queue 时插入└─ expose_decoded_pad()└─ ghost src_%u2非stream 的普通文件或者本地文件setup_source() → make_decoder() → gst_element_link_pads(source, NULL, decodebin, sink),decodebin pad-added → new_decoded_pad_added_cb() 产生 ghost src_%uuridecodebin└─ source (filesrc/fdsrc/isosrc/...)└─ decodebin└─ (内部 autoplug demux/decoder/convert...)└─ raw src pad(s)└─ ghost src_%u3,stream(http/hls)setup_source() → setup_streaming()创建 typefind并 link source→typefindtypefind have-type → type_found()make_decoder()非 adaptive插 queue2 或 downloadbuffer然后 link typefind→queue→decodebinadaptive不插 queue直接 link typefind→decodebindecodebin pad-added → new_decoded_pad_added_cb() 暴露 ghost src_%uuridecodebin└─ source (souphttpsrc/httppullsrc/...)└─ typefind└─ (queue2 或 downloadbuffer) # 由 type_found() 决定└─ decodebin└─ raw src pads└─ ghost src_%u三decodebin 的详细流程decodebin└─ typefind (固定存在)└─ (动态 autoplug)├─ parser├─ demuxer│ └─ multiqueue│ ├─ chain(audio): parser - decoder - ... - raw endpad│ ├─ chain(video): parser - decoder - ... - raw endpad│ └─ chain(text) : subparse/dec - ... - raw endpad└─ (最终) expose src_%u (GstDecodePadghostpad) # 注意expose 后才对外可见2.1 decodebin READY→PAUSED开始异步等 typefind have-typegst_decode_bin_change_state(READY_TO_PAUSED)清理旧 decode_chaindo_async_start()connect(typefind, have-type, type_found)所以 decodebin 的根永远从 typefind 开始2.1 decodebin READY→PAUSED开始异步等 typefind have-typegst_decode_bin_change_state(READY_TO_PAUSED)清理旧 decode_chaindo_async_start()connect(typefind, have-type, type_found)所以 decodebin 的根永远从 typefind 开始3) analyze_new_pad()这是 decodebin autoplug 的“总调度器”对一个新 pad可能来自 typefind、demuxer、parser、decoder它做autoplug-continue只对 fixed caps如果返回 FALSE → 认为“到此为止”直接 exposeplaybin 在上层可以用它“让字幕直接暴露/阻止继续解码”等检查是否已是 final capsare_final_caps()默认 final caps 是 dbin-caps通常是 raw capsaudio/video/x-raw如果 caps 非 fixed → 挂 notify::caps 延迟处理pending_pads获取可用 factories发出 autoplug-factories发出 autoplug-sort可选进入 connect_pad()尝试一个个 factory创建 element 并 link递归向下。关键映射到上层回调链decodebin autoplug- → uridecodebin proxy → playbin autoplug__cb**你前面那套 factories/select/continue/query 就是在这里生效的。4) connect_pad()真正“创建 element 并把树接起来”的地方4.1 demux 分叉识别 demuxer → 插 multiqueue → 为每路流创建子 chain当 is_demuxer即“第一次从 demuxer 出来的 pad”时默认会把 demuxer src pad 先接到 gst_decode_group_control_demuxer_pad()multiqueue.request_pad(sink_%u)demuxer_pad → mq sinkpad取对应 mq srcpad通过 internal-links然后后续对这个 mq srcpad 再继续 autoplug4.2 对每个候选 factoryautoplug-select 决定 TRY/EXPOSE/SKIP在创建 element 前g_signal_emit(SIGNAL_AUTOPLUG_SELECT, dpad, caps, factory, ret);TRY继续创建并尝试 linkEXPOSE不插 element直接把当前 pad expose常用于 sink 直接能接收、或字幕/特殊流希望原样输出SKIP跳过这个 factory在 playbin 里 autoplug_select_cb() 你们实现了如果 factory 是 sink会尝试创建并 activate_sink() 验证能 READY 且接受 caps然后返回 EXPOSE让 decodebin 直接暴露 raw pad 给上层由 playbin 自己接到 playsink/selector。✅如果 factory 是 decoder则会按你们的 decoder/sink 优先级列表aelements/velements做兼容性筛选4.3 element 创建与“探测可用性”decodebin 在 TRY 路径下会做一套非常稳的“试用流程”gst_element_factory_create(factory)gst_element_set_locked_state(element, TRUE)先不让 bin 控制状态gst_bin_add(dbin, element)找 sink pad → link 上游 pad → sinkpadgst_element_set_state(element, READY)gst_pad_query_accept_caps(sinkpad, caps) 再确认失败就 remove 并继续尝试下一个 factory。这解释了为什么 decodebin 看起来“尝试了很多插件”但最终只留一条成功链路4.4 connect_element() 递归 analyze_new_pad()element 连接完成后connect_element() 找 element 的 src padsALWAYS/SOMETIMES 的已存在 pad 立刻纳入 to_connect如果有动态 padsSOMETIMES 未出现就连 pad-added/no-more-pads 回调对每个 output pad 再回到 analyze_new_pad()递归长树所以 decodebin 的“树生长”是一个典型 DFS/递归过程四整体逻辑梳理:playbin (pipeline)├─ playsink├─ uridecodebin(main)│ └─ source - (typefind - queue2/downloadbuffer)? - decodebin - ghost src_%u(audio/video/text)├─ uridecodebin(sub1)│ └─ source - ... - decodebin - ghost src_%u(text/subpicture)├─ uridecodebin(sub2)│ └─ ...├─ input-selector (audio combiner)│ ├─ sink_0 - main src_%u(audio/x-raw) # 多音轨则多个 sink_n│ └─ src - playsink audio pad├─ input-selector (video combiner)│ ├─ sink_0 - main src_%u(video/x-raw)│ └─ src - playsink video pad└─ input-selector (text combiner)├─ sink_0 - main 内嵌字幕 src_%u(text/subpicture)├─ sink_1 - sub1 外挂字幕 src_%u(...)└─ src - playsink text pad以mp4播放为例子梳理播放pipelineplaybin (GstPipeline)├─ playsink│ ├─ (内部 audio chain) ... - audiosink│ ├─ (内部 video chain) ... - videosink│ └─ (内部 text/sub overlay chain) ...│├─ uridecodebin(main) # activate_group() 创建│ └─ (内部见第2节)│├─ input-selector (audio combiner) # playbin pad_added_cb() 首次见到音频创建│ ├─ sink_0 --- uridecodebin(main):src_0 (audio/x-raw)│ ├─ sink_1 --- uridecodebin(main):src_3 (audio/x-raw) # 多音轨时│ └─ src --- playsink:audio_sinkpad (request)│├─ input-selector (video combiner)│ ├─ sink_0 --- uridecodebin(main):src_1 (video/x-raw)│ └─ src --- playsink:video_sinkpad (request)│└─ input-selector (text combiner) # 有字幕流时├─ sink_0 --- uridecodebin(main):src_2 (text/x-raw 或 subpicture/..)├─ sink_1 --- uridecodebin(sub1):src_0 (...)└─ src --- playsink:text_sinkpad (request)以hls 协议为例子看问题的pipelineplaybin└─ uridecodebin(main)└─ source(httpsrc/httppullsrc/...)└─ typefind (capsapplication/x-hls)└─ decodebin└─ hlsdemux (Adaptive Demux)├─ (可能) tsdemux / qtdemux / fmp4demux # 取决于分片格式TS or fMP4│ ├─ audio pad - aacparse - aac decoder - audio/x-raw - decodebin:src_%u│ └─ video pad - h264parse - h264 decoder - video/x-raw - decodebin:src_%u└─ (有字幕则多一支路)decodebin pipeline:decodebin└─ typefind (typefind) # gst_decode_bin_init() 固定创建并 ghost sink└─ qtdemux # connect_pad() 选 factory 后 createlink└─ (multiqueue) # gst_decode_group_new() 创建demuxer_pad - mq├─ queue branch A (audio pad from demux)│ └─ aacparse (optional)│ └─ faad/avdec_aac/硬解aac...│ └─ (可能还有 audioconvert 等取决于插件链)│ └─ [ENDPAD] audio/x-raw # GstDecodePad (ghost) 被 expose 成 decodebin:src_%u│└─ queue branch B (video pad from demux)└─ h264parse (optional)└─ avdec_h264/硬解h264...└─ (可能还有 videoconvert 等)└─ [ENDPAD] video/x-raw # expose 成 decodebin:src_%u1:gst_decode_bin_change_state(READY→PAUSED)connect typefind::have-type → type_found()2:type_found()创建 top chaingst_decode_chain_new(dbin, NULL, typefind.srcpad)analyze_new_pad(dbin, typefind, typefind.srcpad, caps, chain, NULL)3:analyze_new_pad() 发现 caps 不是 final不是 raw发出 autoplug-factories默认返回所有可接 caps 的 factory但你们上层 playbin/uridecodebin 会插手重排/筛选进入 connect_pad() 尝试 factory4:connect_pad() 选中 qtdemux 并创建 element、linkgst_element_factory_create(qtdemux)link typefind.srcpad - qtdemux.sinkconnect_element() 给 qtdemux 连接 pad-added/no-more-pads5.qtdemux::pad-added → decodebin: pad_added_cb()对每个 demux pad 再做一次 analyze_new_pad()因为这是 demuxer 的分叉点会创建 GstDecodeGroup并插入 multiqueuegst_decode_group_new() 创建 multiqueue并 decodebin_set_queue_size_full(preroll...)gst_decode_group_control_demuxer_pad()demux pad - mq.sink_%u得到 mq.srcpad后续对 mq.srcpad 继续 autoplug parser/decoder6.当某条分支最终输出 raware_final_capsTRUE 或 autoplug-continueFALSEexpose_pad()创建/激活 GstDecodePadghost并 block target pad若整棵树 completegst_decode_bin_expose()gst_element_add_pad(decodebin, src_%u) gst_element_no_more_pads(decodebin)gst_decode_pad_unblock() 让流开始走curlhttpsrc 下载总体框架1两个队列src-queue数据队列装 GstBuffersrc-downloadMsgQueue消息队列装 GstCurlHttpSrcMsg包含 event、buf、seek startPos 等2同步手段queue_lock queue_cond控制数据队列的“有数据/等待”download_queuelock保护消息队列select_lock select_cond用于 seek 同步create 等待 seek 完成stop_lock stop_cond用于 stop/clean request 的等待协调request_lock保护对 dl_handle 的 Curl_Create/Clean/Send 等调用避免并发调用3生命周期2.1 start启动下载线程 Curl_Open 发起请求gst_curl_http_src_start()gst_curl_http_src_reset() 清零状态consume_len/download_len/total_len 等src-downloadloopRun TRUE; gst_task_start(download_task);Curl_Open(src-dl_handle, DL_MULTI_MODE);gst_curl_http_src_do_request(src, FALSE); 立即开始一次 GET或 Range GET这里说明download_task 只是负责“处理回调消息并推进状态”真正下载由 Curl 库内部异步执行下载结果通过 gst_curl_http_src_request_buffer_cb() 回调进来。2.2 Curl 回调把网络数据转成 GstBuffer 并入队gst_curl_http_src_request_buffer_cb() 是 curl 库回调入口分两大类A) 数据到达DL_EVT_DATA_SUCCESS或者 ginga 的 DL_EVT_SUCCESS逻辑msg malloc(GstCurlHttpSrcMsg)msg-event DL_EVT_DATA_SUCCESS分配 msg-buf gst_buffer_new_allocate(NULL, data_len, NULL)把 *data memcpy 进去src-download_len data_len如果还不知道总长度用 cb_data-content_len 推断 src-total_len还会更新 basesrc-segment.duration src-total_len发送 DURATION_CHANGED message发送 GST_MESSAGE_MSG_DOWNLOAD_BEGIN 自定义 message节流关键缓冲机制之一当 download_len - consume_len MSG_QUEUE_MAX_SIZE(40MB) 时回调会 sleep 等待消费推进这是一种“下载端自限速”防止内存无限增长把 msg 入 downloadMsgQueueB) 非数据事件停止/完成/失败/重定向等对 DL_EVT_STOP_SUCCESS/DL_EVT_DOWNLOAD_FINISH如果正在 seek把 msg 转成 SEEK_EVENT startPos否则标记 download finish并记录 dl_handle对失败类DL_EVT_FAILED / DL_EVT_DOWNLOAD_FAILED处理重定向cb_data-redirect_url → 写入 src-redirection_uri最后同样把 msg 入 downloadMsgQueue除非 stoped2.3 download_loop消费消息队列真正把 buf 推到数据队列gst_curl_http_src_download_loop() 主循环锁 download_queuelock如果 downloadMsgQueue 非空pop 一个 msg按 msg-event 分支(1) SEEK_EVENT重置队列与偏移并重新发请求 gst_curl_http_src_clear_msgqueue()、gst_curl_http_src_clear_queue() 清空旧数据consume_len startPos; download_len startPos;src-ret GST_FLOW_OK; src-is_seek FALSE; signal(select_cond)调用 gst_curl_http_src_do_request(src, FALSE) 发 Range 请求从 startPos 继续这段实现了“seek 之后 create() 线程继续工作”的同步点create 在等 is_seek 变 FALSE。(2) DL_EVT_DATA_SUCCESS把 msg-buf 放入数据队列 ✅GST_BUFFER_OFFSET(msg-buf) src-consume_len注意这里 offset 用的是当前 consume_len严格来说更像“当前读指针”不是下载位置gst_curl_http_src_push_data_to_queue(src, msg-buf)push 到 src-queuesignal(queue_cond)此外还有 end_offset 逻辑如果设置了 end_offset 且 download_len end_offset有些情况下会 Curl_CleanRequest() 结束下载当 consume_len end_offset会制造一个 DL_EVT_DOWNLOAD_FINISH msg 进入队列触发后续 EOS(3) 失败DL_EVT_FAILED / DL_EVT_DOWNLOAD_FAILED重试或报错先 Curl_CleanRequest(dl_handle)带 request_lock避免并发进入 retry 策略reload_times满足reload_times 3 !disable_retry response_code 不是 4xx/5xx清空队列download_len consume_len从已消费位置续传gst_curl_http_src_do_request(src, TRUE) 再发起 Range 请求否则跳到 download_errorGST_ELEMENT_ERROR(...) 按 response_code 分类src-ret GST_FLOW_CUSTOM_SUCCESS注意这是一个比较奇怪的返回码可能你们用来让上层识别“下载失败但不直接 ERROR”signal(queue_cond) 唤醒 create 退出(4) DL_EVT_DOWNLOAD_FINISH最终 EOS 触发如果仍觉得没下完且可重试走 Curl_CreateResumeRequest() 续传最多 3 次否则post 自定义 messageGST_MESSAGE_MSG_DOWNLOAD_EOS等待消费线程把队列读到尾while consume_len total_len … sleep设置 src-ret GST_FLOW_EOS; signal(queue_cond)create() 之后返回 EOS2.4 create从数据队列取一个 buffer消费者️gst_curl_http_src_create()如果 cancel_download返回 GST_FLOW_FLUSHING等待 seek 完成while(src-is_seek) wait(select_cond)gst_curl_http_src_pop_data_from_queue()如果队列为空且 src-retGST_FLOW_OKwait(queue_cond)pop 一个 buffersrc-consume_len gst_buffer_get_size(buf)如果 src-ret ! OK 或 stoped释放 outbuf返回对应 flowEOS/FLUSHING/CUSTOM等2.5 stop停止下载、清队列、让 create 退出 gst_curl_http_src_stop()stoped TRUE; 清掉 start/end offsetpause 下载 task并 signal stop_condrequest_lock 下 Curl_CleanRequest()并 STOP_COND_WAIT 等待回调确认Curl_Close(dl_handle)清空 data queue / msg queuesrc-ret GST_FLOW_EOS; signal(queue_cond) 让 create 线程退出2.6下载的数据如果传到后端组件例如decodebinlibcurl callback thread└─ 写 src-buffer signal buffer_condgst_curl_http_src_create()├─ 等待 buffer_cond├─ gst_buffer_new_allocate()├─ memcpy HTTP 数据└─ return GST_FLOW_OK outbuf↓GstBaseSrc (内部)└─ gst_pad_push(srcpad, outbuf)↓typefind└─ demux / decodebin↓audio/video raw pads2.7 curl 数据从PLAYING到首帧的逻辑通路2.7.1:playbin → uridecodebin → curlhttpsrcgst_base_src_start() 被调用curlhttpsrc 内部创建 curl easy handle请求 m3u8 / mpdHLS/DASHdownload 线程启动2.7.2:curlhttpsrc 开始产生首批 buffer,在你的 gst_curl_http_src_create() 中while (src-buffer_len 0 state OK)g_cond_wait(buffer_cond);一旦 curl 回调收到数据HTTP response bodym3u8 / mpd / segmentmemcpy → src-buffersignal(buffer_cond)然后*outbuf gst_buffer_new_allocate(...)return GST_FLOW_OK;2.7.3:typefind 介入,缓存前 N 个 buffer,分析 magic/header,决定 caps2.7.4:decodebin 进入 adaptive 模式HLS / DASH)2.7.5:Manifest 解析m3u8 / mpd,hlsdemux / dashqtdemux消费 curlhttpsrc 提供的 buffer解析 manifest决定variant / representationsegment URL初始化 segmentinit.mp4这个阶段不会出 audio/video pad只是“控制面解析”所以你会看到有网络流量但没首帧2.7.6::开始请求第一个 media segment,curlhttpsrc → 新的 HTTP 请求segment,curlhttpsrc 又开始吐 segment 的二进制 buffer,这些 buffer 再次走curlhttpsrc → typefind这次是 video/quicktime / video/mpegts2.7.7:二级 demux parser decoder 插入hlsdemux└─ tsdemux├─ h264parse → avdec_h264 / v4l2h264dec└─ aacparse → faad / avdec_aac2.7.8:decodebin 的 expose 条件首帧关键点:decodebin 不会立刻 expose pad它要满足每个 group所有“必需流”audio/video已发现decoder 完成 prerollpad 不再 block2.7.9:pad expose → playbin selector → playsinkuridecodebin 把 pad 转交给 playbinplaybin接入 input-selector连接 playsinkunblock selector.src3, httppullsrccurl_interface 下载流程GStreamer上层拉取└── gst_http_pull_src_fill()├── 按 offset/length 读 queue├── 不够则等待 queue_cond└── 超范围则触发 seek/re-request中间调度层└── gst_http_pull_src_download_loop()├── 处理下载回调消息 downloadMsgQueue├── DATA_SUCCESS - push 到 data queue├── FINISH - EOS / resume├── FAILED - retry / error└── SEEK_EVENT - 清队列并重发请求底层下载层└── curl_interface.c├── Curl_Open() 创建 multi 下载线程├── Curl_CreateRequest() 配 easy/multi├── Curl_SendRequest() 启动请求├── CurlDownload_DLTask() 驱动 curl_multi_perform/select├── CurlDownload_WriteData() 收数据├── CurlDownload_WriteHeader() 解析头└── 通过 gst_http_pull_src_request_buffer_cb() 回调给 httppullsrc三个线程的流转过程线程ACurlDownload_DLTask职责驱动 libcurl multiselect/socket 等待接收 header/data 回调在完成/失败时回调上层线程Bgst_http_pull_src_download_loop职责消费回调生成的消息把数据塞到 src-queue处理 seek/retry/eos/error线程CGStreamer pull 线程职责调用 gst_http_pull_src_fill()按 offset 从 src-queue 拷数据返回 demux这个“三线程模型”是理解整套逻辑的关键3.1下载主流程,对应线程Bgst_http_pull_src_change_state()开启下载gst_http_pull_src_download_start()线程下载线程主要做如下几件事儿3.1.1gst_http_pull_src_reset(src);3.1.2启动内部下载任务gst_task_start(src-download_task);这个线程不是直接下载网络而是处理下载消息队列 downloadMsgQueue。3.1.3:打开 curl 下载句柄 Curl_Open(src-dl_handle, DL_MULTI_MODE);3.2:真正发起http 请求gst_http_pull_src_do_request3.2.1:先构造DL_HTTPPARAM_t3.2.2:根据 consume_len / end_offset / total_len 拼 Range3.2.3:真正创建并发起请求:Curl_CreateRequest(src-dl_handle, gst_http_pull_src_request_buffer_cb, src, http_param);Curl_SendRequest(src-dl_handle, DL_GET_METHOD);3.3:curl_interface.c 这一层的下载流程3.3.1Curl_Open()初始化 DL_CONTEXT_t启动curl_download 下载线程对应线程ATask_Create(CurlDownload_DLTask,dl_context,dl_context-task_id);3.3.2:Curl_CreateRequest(),保存 URL / callback / 状态清零,创建或 reset easy handle,设置 libcurl 参数,multi 模式加入 multi handledl_context-multi_handle curl_multi_init();curl_multi_add_handle(dl_context-multi_handle, dl_context-easy_handle);3.3.3:Curl_SendRequest(),CurlDownload_SetTaskStatus(dl_context,DL_STATE_START);3.4curl 后台线程 CurlDownload_DLTask() 是下载引擎核心3.4.1状态机DL_STATE_INIT DL_STATE_START DL_STATE_LOADING DL_STATE_FINISHIED DL_STATE_STOP DL_STATE_CLOSED3.4.2curl_multi_perform(dl_context-multi_handle, stillRunning)驱动curl网络超时会清理 curl 资源并回调失败3.4.3收到数据会怎么走CurlDownload_WriteHeader() header 回调CurlDownload_WriteData() 数据回调3.4.3.1header 回调主要做如下动作Content-LengthContent-RangeTransfer-Encoding: chunkedLocationAccept-RangesHTTP 状态码3.4.3.2data 回调主要做如下动作{if(data_size 0){DL_CB_DATA_t cb_data {0};dl_context-total_len data_size;dl_context-response_data_len dl_context-total_len;cb_data.content_len dl_context-content_len;dl_context-respose_callback(data,data_size,(dl_context-callback_para),DL_EVT_DATA_SUCCESS,cb_data);}}libcurl 每收到一块网络数据就立即通过 gst_http_pull_src_request_buffer_cb() 回调给 httppullsrc。这里不是先缓存到 curl_interface 再一次性交付而是 streaming callback3.5httppullsrc 的下载回调gst_http_pull_src_request_buffer_cb()3.5.1收到 DL_EVT_DATA_SUCCESSGstBuffer* tmpbuff gst_buffer_new_allocate(NULL, data_len, NULL);memcpy(info.data, *data, data_len);src-partial_spu gst_buffer_append(src-partial_spu, tmpbuff)3.5.2:更新下载进度src-download_len data_len;3.5.3首次拿到 content_len 时更新 total_len3.5.4限流防止 queue 太大3.5.5聚合到 300KB 再投递消息if (gst_buffer_get_size(src-partial_spu) 300*1024 src-download_len src-total_len) {push_msg false;}每次 curl 回调的数据不直接进主 queue而是先聚合约到 300KB 再封成消息推给 downloadMsgQueue。这样做的目的源码注释已经写了避免每个小块都发消息导致 download loop 来不及处理影响 source 性能3.6download_loop 如何把消息转成真正可读 queue3.6.1收到 DL_EVT_DATA_SUCCESSgst_http_pull_src_push_data_to_queue(src, msg-buf);g_queue_push_tail(src-queue,data);src-queue_len gst_buffer_get_size(data);g_cond_signal(queue_cond);3.6.2收到 DL_EVT_FAILED / DL_EVT_DOWNLOAD_FAILED调 Curl_CleanRequest()判断是否允许重试重试次数 3不是 4xx/5xx清消息队列/数据队列把 download_len / queue_len 回退到 consume_len再调 gst_http_pull_src_do_request(src, TRUE)3.6.3收到 DL_EVT_DOWNLOAD_FINISH如果数据没下完整还能继续3.6.3.1数据真的下完了post 一个 download eos message等待已下载数据被上游消费最后塞一个特殊的 EOS buffer3.6.4收到 seek 触发的 SEEK_EVENT清消息队列清数据队列把 consume_len/download_len/queue_len 都改成目标 startPossrc-is_seek FALSE再次调用 gst_http_pull_src_do_request(src, FALSE)3.7上层真正取数据gst_http_pull_src_fill()3.7.1请给我从 offset 开始的 length 字节先判断是否能从 backup 读这是为了处理某些 seek/backward 场景避免立刻重新下载3.7.2判断 offset 是否超出当前缓存窗口fill() 如果发现请求位置不在当前缓存窗口内会阻塞等待新 seek 请求重新建链并下载数据。3.7.3如果 offset 在缓存窗口内就直接从 queue 拷贝遍历 queue 中的多个 GstBuffer找到 offset 对应位置按需丢掉前缀 drop_len连续 memcpy 到目标 buf把网络下载分片 queue 拼接成上游想要的任意 offset/length 数据块。3.7.4如果数据还不够就等待更多 buffer fill() 是一个典型的消费者阻塞模型queue 有数据 - 直接拷贝queue 不够 - 等下载线程继续塞3.7.5什么时候返回 EOS读到特殊 EOS buffer或者外部stop/cancel总结1. READY-PAUSED 时httppullsrc 启动 download_task2. 调 Curl_Open 创建 dl_handle 和 curl 后台线程3. do_request() 按 consume_len 拼 Range4. Curl_CreateRequest Curl_SendRequest 发起 GET5. curl 线程执行 multi_perform/select6. 收到 header解析 Content-Length / Content-Range / Accept-Ranges7. 收到 dataWriteData 以 PART 模式回调给 httppullsrc8. httppullsrc 回调里把小块数据拼成 partial_spu约300KB9. partial_spu 封成消息放入 downloadMsgQueue10. download_loop 消费消息把 GstBuffer push 到 data queue11. fill(offset,length) 从 data queue 拷贝数据给上游 demux12. 下载结束时投递 EOS 标记 bufferfill 返回 GST_FLOW_EOSgstculhttpsrc.c 跟gsthttppullsrc.c 的模式的区别gstcurlhttpsrc 是 PUSH 模式核心函数是gst_curl_http_src_create (GstPushSrc * psrc, GstBuffer ** outbuf)source 自己不断产出下一块数据”下游通常按顺序消费更适合顺序 HTTP body 流式传输gsthttppullsrc 是 GstBaseSrc的pull/fill 模式核心函数gst_http_pull_src_fill (GstBaseSrc * src, guint64 offset, guint length, GstBuffer * buf)上游按 offset 指定读取source 必须回答“这个 offset 开始的 length 字节是什么”支持随机访问 / seek / range 更自然两种不同的模块跟libcurl的耦合层级不一样1gstcurlhttpsrc直接耦合 libcurlgstcurlhttpsrc.c 里直接curl_easy_initcurl_easy_setoptcurl_multi_initcurl_multi_add_handlecurl_multi_performcurl_multi_info_read2gsthttppullsrc通过 curl_interface.c 间接耦合不直接调用 curl而是调用Curl_OpenCurl_CreateRequestCurl_SendRequestCurl_CleanRequestCurl_CloseCurl_CheckHeader

gstreamer pipeline梳理及插件新增方案

相关推荐

C语言通讯录系统：从入门到工程实践的完整图谱

年底还在招的岗位，先确认缺口和到岗要求

Sue E. Goodman. Beginning Topology

更多资讯

AI音乐如何成为可交付的真实作品

2.8万亿参数开源大模型上架亚马逊云：MoE架构与推理成本全解析

DeepSeek涨价后，五个实测可用的低价替代方案与降本组合拳

LINGO建模求解实战：从安装配置到Excel数据对接与避坑指南

链表回文判断：快慢指针与反转链表技术详解

Kubernetes核心组件与架构深度解析

今日更新

AI辅助技术栈评估：提升研发决策效率60%

从生成Token到直接决策：Jev如何重塑AI Agent工程实践

32G统一内存跑大模型：内存账、TPS实测与端云分流

本周精选

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

读完文章，想聊聊您的网站？