WebRTC服务质量(10)- Pacer机制(02) RoundRobinPacketQueue

news2025/3/17 1:17:23

WebRTC服务质量(01)- Qos概述
WebRTC服务质量(02)- RTP协议
WebRTC服务质量(03)- RTCP协议
WebRTC服务质量(04)- 重传机制(01) RTX NACK概述
WebRTC服务质量(05)- 重传机制(02) NACK判断丢包
WebRTC服务质量(06)- 重传机制(03) NACK找到真正的丢包
WebRTC服务质量(07)- 重传机制(04) 接收NACK消息
WebRTC服务质量(08)- 重传机制(05) RTX机制
WebRTC服务质量(09)- Pacer机制(01) 流程概述
WebRTC服务质量(10)- Pacer机制(02) RoundRobinPacketQueue
WebRTC服务质量(11)- Pacer机制(03) IntervalBudget
WebRTC服务质量(12)- Pacer机制(04) 向Pacer中插入数据

一、前言:

RoundRobinPacketQueue 是Pacer模块当中一个非常重要的循环队列模块,主要目的是在多流场景中,根据每个流的 优先级 以及流内的 RTP 包的 入队顺序类型 和其他因素,动态决定数据包的发送顺序。

二、类关系图:

2.1、类图:

在这里插入图片描述

关键部分加了注释,便于理解:

  1. 多流支持: 每个流按 SSRC(Synchronization Source Identifier,同步信源标识符)进行区分,每个流维护独立的优先级队列。
  2. 多级优先级调度: 不同流间按优先级调度(基于 StreamPrioKey),单个流内的包按严格的队列顺序管理。
  3. 静态与动态属性结合: 例如,属性包括每个包的优先级(priority)、入队时间(enqueue_time)、重传标志、带宽开销等。
  4. 分组调度: 使用轮询(Round-Robin)机制,从不同流根据流的权重和优先级选取包发送,提高音频、视频和反馈等多种类型数据的实时性。

2.2、重要成员关系:

在这里插入图片描述

  1. stream_priorities当中存放StreamPrioKey和ssrc对应关系;
  2. streams当中存放ssrc和Stream的对应关系;
  3. 这样就建立了StreamPrioKey和Stream之间的关系;
  4. 反过来,也可以通过Stream当中的priority_it找到stream_priorities当中的某一项;

小结下:重要成员变量的功能

成员变量功能
streams_保存所有流,键是流的 ssrc,每条流存有独立队列和优先级信息。
stream_priorities_将所有流按照 StreamPrioKey 排列,用于实现流间优先级调度。
enqueue_times_一个多重集合,保存所有包的入队时间,便于快速找到最早入队的包时间。
size_packets_size_分别记录队列中包的个数和总字节数,动态调整。
transport_overhead_per_packet_计算包传输的额外开销(如包头)。
time_last_updated_用于统计队列的更新时间,辅助计算入队和等待时间等。

三、重要函数:

3.1、Push

Push 会根据包的优先级、流的权重、总队列大小,以及包的类型等,将包插入到对应的流(Stream)队列,并更新其他与队列状态关联的元数据。总体思路如下:

处理一个新包时:

  1. 确定该包所属的流,如果不存在该流,则创建一个 Stream 对象。
  2. 将包包装成为 QueuedPacket 并插入到流的优先级队列。
  3. 更新流的优先级键 StreamPrioKey,并在 stream_priorities_ 中重新排序。
  4. 更新队列元数据(如队列包大小、队列时间等)。
void RoundRobinPacketQueue::Push(int priority,
                                 Timestamp enqueue_time,
                                 uint64_t enqueue_order,
                                 std::unique_ptr<RtpPacketToSend> packet) {
  RTC_DCHECK(packet->packet_type().has_value());
  if (size_packets_ == 0) {
    // Single packet fast-path.
    single_packet_queue_.emplace(
        QueuedPacket(priority, enqueue_time, enqueue_order,
                     enqueue_times_.end(), std::move(packet)));
    UpdateQueueTime(enqueue_time);
    single_packet_queue_->SubtractPauseTime(pause_time_sum_);
    size_packets_ = 1;
    size_ += PacketSize(*single_packet_queue_);
  } else {
    MaybePromoteSinglePacketToNormalQueue();
    Push(QueuedPacket(priority, enqueue_time, enqueue_order,
                      enqueue_times_.insert(enqueue_time), std::move(packet)));
  }
}


void RoundRobinPacketQueue::Push(QueuedPacket packet) {
  auto stream_info_it = streams_.find(packet.Ssrc());
  if (stream_info_it == streams_.end()) {
    stream_info_it = streams_.emplace(packet.Ssrc(), Stream()).first;
    stream_info_it->second.priority_it = stream_priorities_.end();
    stream_info_it->second.ssrc = packet.Ssrc();
  }

  Stream* stream = &stream_info_it->second;

  if (stream->priority_it == stream_priorities_.end()) {
    RTC_CHECK(!IsSsrcScheduled(stream->ssrc));
    stream->priority_it = stream_priorities_.emplace(StreamPrioKey(packet.Priority(), stream->size), packet.Ssrc());
  } else if (packet.Priority() < stream->priority_it->first.priority) {
    stream_priorities_.erase(stream->priority_it);
    stream->priority_it = stream_priorities_.emplace(StreamPrioKey(packet.Priority(), stream->size), packet.Ssrc());
  }

  if (packet.EnqueueTimeIterator() == enqueue_times_.end()) {
    packet.UpdateEnqueueTimeIterator(enqueue_times_.insert(packet.EnqueueTime()));
  } else {
    UpdateQueueTime(packet.EnqueueTime());
    packet.SubtractPauseTime(pause_time_sum_);

    size_packets_ += 1;
    size_ += PacketSize(packet);
  }

  stream->packet_queue.push(packet);
}
  • 从streams中找packet所属的Ssrc的stream,如果没有,则在streams中插入一项;
  • 查看stream的priority_it是否等于stream_priorities_的end()
    • 如果相等,则在stream_priorities_中插入一项;
    • 否则,如果新包的优先高(注意,优先级数值越小表示优先级越高),则更新其ssrc对应队列的优先级;
  • 更新队列总时长;
  • 入队时间减去暂停时间(一般不会有暂停);
  • 队列总包数+1;
  • 队列总字节大小 = 包的负载大小+Padding大小;
  • 插入到stream中的packet_queue中;

3.2、Pop

Pop 会轮询不同的流并从当前优先级最高的流中取出一个包发送,同时维护包的发送顺序。总体思路如下:

  1. 调用 GetHighestPriorityStream 获取当前优先级最高的流。
  2. 从该流的优先级队列(PriorityPacketQueue)中取出队首包,并更新流的状态(如剩余大小、时间等)。
  3. 如果该流没有剩余包,删除对应的流优先级键。
std::unique_ptr<RtpPacketToSend> RoundRobinPacketQueue::Pop() {
  if (single_packet_queue_.has_value()) {
    RTC_DCHECK(stream_priorities_.empty());
    std::unique_ptr<RtpPacketToSend> rtp_packet(
        single_packet_queue_->RtpPacket());
    single_packet_queue_.reset();
    queue_time_sum_ = TimeDelta::Zero();
    size_packets_ = 0;
    size_ = DataSize::Zero();
    return rtp_packet;
  }

  RTC_DCHECK(!Empty());
  Stream* stream = GetHighestPriorityStream();
  const QueuedPacket& queued_packet = stream->packet_queue.top();

  stream_priorities_.erase(stream->priority_it);

  // Calculate the total amount of time spent by this packet in the queue
  // while in a non-paused state. Note that the |pause_time_sum_ms_| was
  // subtracted from |packet.enqueue_time_ms| when the packet was pushed, and
  // by subtracting it now we effectively remove the time spent in in the
  // queue while in a paused state.
  TimeDelta time_in_non_paused_state =
      time_last_updated_ - queued_packet.EnqueueTime() - pause_time_sum_;
  queue_time_sum_ -= time_in_non_paused_state;

  RTC_CHECK(queued_packet.EnqueueTimeIterator() != enqueue_times_.end());
  enqueue_times_.erase(queued_packet.EnqueueTimeIterator());

  // Update |bytes| of this stream. The general idea is that the stream that
  // has sent the least amount of bytes should have the highest priority.
  // The problem with that is if streams send with different rates, in which
  // case a "budget" will be built up for the stream sending at the lower
  // rate. To avoid building a too large budget we limit |bytes| to be within
  // kMaxLeading bytes of the stream that has sent the most amount of bytes.
  DataSize packet_size = PacketSize(queued_packet);
  stream->size =
      std::max(stream->size + packet_size, max_size_ - kMaxLeadingSize);
  max_size_ = std::max(max_size_, stream->size);

  size_ -= packet_size;
  size_packets_ -= 1;
  RTC_CHECK(size_packets_ > 0 || queue_time_sum_ == TimeDelta::Zero());

  std::unique_ptr<RtpPacketToSend> rtp_packet(queued_packet.RtpPacket());
  stream->packet_queue.pop();

  // If there are packets left to be sent, schedule the stream again.
  RTC_CHECK(!IsSsrcScheduled(stream->ssrc));
  if (stream->packet_queue.empty()) {
    stream->priority_it = stream_priorities_.end();
  } else {
    int priority = stream->packet_queue.top().Priority();
    stream->priority_it = stream_priorities_.emplace(
        StreamPrioKey(priority, stream->size), stream->ssrc);
  }

  return rtp_packet;
}
  • 获得优先级最高的stream;

  • 从stream的packet_queue当中取出第一个包;

  • 将stream在stream_priorities_中的项删除;

    • 思考下虽然包删除了,但是stream还在,为啥要删除?

      答案:因为stream_priorities_ 是multimap,允许出现相同key,也就是说,这个包没有,它的stream优先级全靠它撑着(因为是先pop优先级高的包),所以,现在不应该由它撑着了,换下一个优先级最高的包撑着。

  • 计算Packet入队后距离现在的时间(不包括暂停时间);

  • 将这段时间从队列的总时间减去;

  • enqueue_times_中将Packet的项删除;

  • 总包数减去1;

  • 总字节数减去包的字节数;

  • 将包从stream中的queue中弹出;

  • 如果stream中的队列为空,则令stream的priority_it指向stream_priorities_end();

  • 否则,从stream队列头部取出Packet,将该Packet的priority插入到stream_priorities_中;

四、优先级调度:

  • 每个流有单独的优先级队列(PriorityPacketQueue),保存 QueuedPacket 对象。

  • QueuedPacket 是一个包装类,表示单个 RTP 数据包及其附属信息(例如入队时间、优先级、是否为重传包等)。

  • 列表中的流按 StreamPrioKey 保存在 stream_priorities_ 中,通过此键决定流次序:

    struct StreamPrioKey {
      int priority;    // 流的优先级,数值越低,优先级越高
      DataSize size;   // 数据包大小,用于平衡负载
    };
    

    优先规则:优先级低(priority 值小) > 数据包大小小(size 值小)。

五、轮询调度:

  • 核心逻辑通过 RoundRobin 的方式轮询多个流。但由于流可能有不同的优先级,某些流会被更多次轮询到。
  • 函数 GetHighestPriorityStream 定位当前最高优先级的流,从流对应的队列中取得包然后发送。

六、Stream定义与管理:

每个流由 Stream 类表示,Stream 是该流独有的数据结构,包括:

  • 当前队列状态,如总字节数、包大小和优先级。
  • 内部维护单独的优先级队列(PriorityPacketQueue)。
  • 调度时实时更新优先级,确保新加入高优先级包时能调整队列次序。

七、其他特性:

7.1、重传支持:

  • 标记是否重传的包,在出队时可能依据该标记进行特殊处理。

7.2、时间相关:

  • queue_time_sum_pause_time_sum_ 用于统计包在队列中的存留时间,这对于带宽控制和流量管理很有用。
  • 提供接口如 AverageQueueTime 计算平均队列时间,用于监控流的实时性。

7.3、队列字节限制:

队列有最大可存储的字节数(max_size_),以防止占用过多资源。

7.4、暂停/恢复功能:

可以通过 SetPauseState 暂停或者恢复队列处理。劝你最好别用!!!

八、总结:

RoundRobinPacketQueue 是一个高效的多流、多优先级调度队列,适用于 RTP 媒体数据的分组发送场景。它通过流内、流间的双重调度机制,结合优先级动态提升、统计队列时间和暂停控制等特性,确保在带宽有限的网络环境中最大程度提高数据的实时性和发送效率,是 WebRTC Pacer 模块的核心部分。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/2266627.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

基于python的电子报实现思路

一种基于PDF生成电子报的思路 需求提出实现思路&#xff1a;技术路线核心代码&#xff1a; 需求提出 最近公司提出了一个电子报的需求&#xff0c;可看网上实现的思路基本上是方正系列的排版软件实现的&#xff0c;公司没必要买这么一套&#xff0c;于是按照自己的思路搞了一个…

【HarmonyOS NEXT】鸿蒙原生应用“上述”

鸿蒙原生应用“上述”已上架华为应用市场&#xff0c;欢迎升级了鸿蒙NEXT系统的用户下载体验&#xff0c;用原生更流畅。 个人CSDN鸿蒙专栏欢迎订阅&#xff1a;https://blog.csdn.net/weixin_44640245/category_12536933.html?fromshareblogcolumn&sharetypeblogcolumn&a…

IntelliJ IDEA中设置激活的profile

在IntelliJ IDEA中设置激活的profile&#xff0c;可以通过以下步骤进行&#xff1a; 通过Run/Debug Configurations设置 打开Run/Debug Configurations对话框&#xff1a; 在IDEA的顶部菜单栏中&#xff0c;选择“Run”菜单&#xff0c;然后点击“Edit Configurations...”或者…

抖去推碰一碰系统技术源码/open SDK转发技术开发

抖去推碰一碰系统技术源码/open SDK转发技术开发 碰一碰智能系统#碰碰卡系统#碰一碰系统#碰一碰系统技术源头开发 碰碰卡智能营销系统开发是一种集成了人工智能和NFC技术的工具&#xff0c;碰碰卡智能营销系统通过整合数据分析、客户关系管理、自动化营销活动、多渠道整合和个…

jumpserver docker安装

#安装jumpserver最新版本&#xff08;当前最新版本v4.5.0-ce&#xff09; curl -sSL https://resource.fit2cloud.com/jumpserver/jumpserver/releases/latest/download/quick_start.sh | bash#登录 http://192.168.31.168/ 默认账号密码 admin/ChangeMe 修改后&#xff1a; ad…

Linux shell脚本用于常见图片png、jpg、jpeg、webp、tiff格式批量转PDF文件

Linux Debian12基于ImageMagick图像处理工具编写shell脚本用于常见图片png、jpg、jpeg、webp、tiff格式批量转PDF文件&#xff0c;”多个图片分开生成多个PDF文件“或者“多个图片合并生成一个PDF文件” BiliBili视频链接&#xff1a; Linux shell脚本对常见图片格式批量转换…

详细介绍Sd-WebUI提示词的语法规则

AI绘画中最大的门槛就是提示词&#xff0c;对英语水平、文学水平、想象力、灵感等要求较高。不能每次一输入正向提示词&#xff08;positive prompt&#xff09;&#xff0c;就只会写a girl, big eyes, red hair。虽然sd-webui软件可以直接翻译&#xff0c;输入一个子母后会立刻…

对Python中队列三种实现方式的测试

一、结论 本程序模拟比较队列的三种情况&#xff1a; 一、deque当作队列使用&#xff1b; 二、queue当作队列使用&#xff1b; 三、list当作队列使用。 结论&#xff1a; Python队列可以用deque、queue、list实现&#xff0c;其中list极慢、queue较慢&#xf…

【VScode】第三方GPT编程工具-CodeMoss安装教程

一、CodeMoss是什么&#xff1f; CodeMoss是一款集编程、学习和办公于一体的高效工具。它兼容多种主流平台&#xff0c;包括VSCode、IDER、Chrome插件、Web和APP等&#xff0c;支持插件安装&#xff0c;尤其在VSCode和IDER上的表现尤为出色。无论你是编程新手还是资深开发者&a…

Unity中如何修改Sprite的渲染网格

首先打开SpriteEditor 选择Custom OutLine,点击Genrate 则在图片边缘会出现边缘线&#xff0c;调整白色小方块可以调整边缘 调整后&#xff0c;Sprite就会按照调整后的网格渲染了。 如何在UI中使用&#xff1f; 只要在UI的Image组件中选择Use Sprite Mesh 即可 结果&#xff1…

【机器学习】探索机器学习与人工智能:驱动未来创新的关键技术

探索机器学习与人工智能&#xff1a;驱动未来创新的关键技术 前言&#xff1a;人工智能的核心技术深度学习&#xff1a;自然语言处理&#xff08;NLP&#xff09;&#xff1a;计算机视觉&#xff1a; 机器学习与人工智能的驱动创新医疗健康领域金融行业智能制造与工业互联网智慧…

Visual Studio Code(VS Code)配置C/C++环境

一、Visual Studio Code安装 Visual Studio Code&#xff0c;下文中简称为VS Code的详细安装方法请参考VSCode安装教程&#xff08;超详细&#xff09;-CSDN博客 二、MinGW编译器下载与配置 1、MinGW介绍 MinGW(Minimalist GNU for Windows)是一款用于Windows 平台的轻…

Postman接口测试01|接口测试基础概念、http协议、RESTful风格、接口文档

目录 一、接口测试基础概念 1、什么是接口 2、接口的类型 3、什么是接口测试 4、为什么要做接口测试 5、接口测试的实现方式 6、什么是自动化接口测试&#xff1f; 二、接口返回的数据格式 1、三种格式 2、Json 三、接口协议 1、webservice协议 2、dubbo协议 3、…

HDR视频技术之十一:HEVCH.265 的 HDR 编码方案

前文我们对 HEVC 的 HDR 编码优化技术做了介绍&#xff0c;侧重编码性能的提升。 本章主要阐述 HEVC 中 HDR/WCG 相关的整体编码方案&#xff0c; 包括不同应用场景下的 HEVC 扩展编码技术。 1 背景 HDR 信号一般意味着使用更多比特&#xff0c;一般的 HDR 信号倾向于使用 10…

ThinkPHP 8开发环境安装

【图书介绍】《ThinkPHP 8高效构建Web应用》-CSDN博客 《ThinkPHP 8高效构建Web应用 夏磊 编程与应用开发丛书 清华大学出版社》【摘要 书评 试读】- 京东图书 1. 安装PHP8 Windows系统用户可以前往https://windows.php.net/downloads/releases/archives/下载PHP 8.0版本&am…

CAN201 Introduction to Networking(计算机网络)Pt.2 传输层

文章目录 3. Transport Layer&#xff08;传输层&#xff09;3.1 Multiplexing and demultiplexing&#xff08;多路复用和多路分解&#xff09;3.2 Connectionless transport&#xff1a;UDP3.3 Principles of reliable data transfer3.4 Pipelined communication3.5 TCP: con…

linux系统上SQLPLUS的重“大”发现

SQL plus版本&#xff1a; [oraclepg-xc2 ~]$ sqlplus -v SQL*Plus: Release 19.0.0.0.0 - Production Version 19.3.0.0.0 操作系统&#xff1a;CentOS Linux 7 (Core) 数据库&#xff1a;Oracle 19c Version 19.3.0.0.0 同样的SQL脚本在windos CMD sqlplus 执行没问题。…

Unity中实现人物残影效果

今天火柴人联盟3公测了&#xff0c;看到一个残影的效果&#xff0c;很有意思&#xff0c;上网查询了一下实现方式&#xff0c; 实现思路&#xff1a; 将角色的网格复制出来&#xff0c;然后放置到新建的物体的MeshFilter组件上&#xff0c;每隔几十毫秒在玩家的位置生成一个&a…

mysql mmm和mha对比

本文简单介绍MySQL的两个high availability方案&#xff0c;MMM和MHA。 MMM MMM(Master-Master replication managerfor Mysql)的基本组成如下&#xff0c; 主节点master1&#xff1a;承载写流量备主节点master2&#xff1a;replicate主节点master1的写流量&#xff0c;在主节…

探秘“香水的 ChatGPT”:AI 开启嗅觉奇幻之旅!

你没有看错&#xff0c;AI也能闻到味道了&#xff01;这是一家名为Osmo公司公布的信息&#xff0c;他们成功创造出了由AI生成的李子味道&#xff0c;快跟着小编一探究竟吧~ 【图片来源于网络&#xff0c;侵删】 Osmo公司的这项技术&#xff0c;通过分析香味的化学成分和人类嗅…