深入浅出mediasoup—通信框架

libuv 是一个跨平台的异步事件驱动库,用于构建高性能和可扩展的网络应用程序。mediasoup 基于 libuv 构建了包括管道、信号和 socket 在内的一整套通信框架,具有单线程、事件驱动和异步的典型特征,是构建高性能 WebRTC 流媒体服务器的重要基础,本文主要分析 mediasoup 对 libuv 的封装。

1. Pipe 通信

Node.js 进程与 worker 进程之间使用管道通信,而且是双向通信。node.js 进程通过管道向 worker 进程发送请求,并接收响应。worker 进程也可以主动向 node.js 进程发送通知消息。

1.1. 文件描述符

管道通信需要使用两个文件描述符,node.js 进程的文件描述符定义如下:

this.#channel = new Channel({producerSocket: this.#child.stdio[3],consumerSocket: this.#child.stdio[4],pid: this.#pid,
});

worker 进程的文件描述符定义如下:

static constexpr int ConsumerChannelFd{ 3 };
static constexpr int ProducerChannelFd{ 4 };

1.2. 静态结构

worker 进程对管道通信的封装看起来比较复杂,涉及到多个类,如下图所示。由于这里面糅合了几个逻辑,拆解以后会更好理解:

1)UnixStreamSocketHandle 封装了基于 libuv 的 pipe 通信能力,内部包含 libuv 句柄。

2)ChannelSocket 内部包含的 ConsumerSocket 和 ProducerSocket 对应管道通信的读和写两个方向。ChannelSocket 继承了 ConsumerSocket::Listener,从 ConsumerSocket 收到的管道消息,都会回调到 ChannelSocket。

3)全局只有一个 ChannelSocket 对象,被 Worker 持有。Worker 继承了 ChannelSocekt::Listener,ChannelSocket 收到的所有管道消息都会回调 Worker。

4)Worker 包含了一个 Shared 对象,从名字上能看出,这是一个“共享对象”,通过传参的方式共享给各个对象,本质上就是一个全局对象。

5)Shared 内部包含两个对象:ChannelMessageRegistor 和 ChannelNotifier。ChannelMessageRegistor 用来管理管道消息处理器,因为全局就一个 ChannelSocket 对象,所有需要处理管道消息的对象都要把自己注册到 ChannelMessageRegistor,Worker 根据注册信息把管道消息分发给各个处理器。ChannelNotifier 用来发送管道消息,其内部也是使用 ChannelSocket 来发送消息,所有对象需要向 Node.js 进程发送管道消息调用 ChannelNotifier 接口即可。

1.3. 数据流

管道通信的数据流如下图所示。接收到的管道消息会一层层回调到 Worker 对象,Worker 先对消息进行过滤,如果是 Worker 自己关注的消息,自己先处理,其他消息则根据“注册表”进行路由。发送管道消息,调用 ChannelNotifier::Emit 接口,最终通过 libuv 发送出去。

2. Socket 通信

Socket 通信主要用来处理 mediasoup worker 与 WebRTC 客户端之间的媒体通信,支持 TCP 和 UDP。

2.1. 静态结构

2.1.1. UDP

1)UdpSocketHandle 封装了基于 libuv 的 UDP 通信能力,内部包含 libuv 句柄。

2)UdpSocket 继承自 UdpSocketHandle,内部包含了一个数据监听对象,用来接收 UDP 消息。

2)PipeTransport、PlainTransport、WebRtcTransport 和 WebRtcServer 都支持 UDP 通信,它们内部都包含一个指向 UdpSocket 的指针,用来发送 UDP 消息。

【注】这里的 PipeTransport 并不是使用管道通信的 transport。

2.1.2. TCP

1)TcpServerHandle 封装了基于 libuv 的 TCP 监听能力,内部包含 libuv 句柄。

2)TcpConnectionHandle 封装了基于 libuv 的 TCP 通信能力,内部包含 libuv 句柄。TCP 连接中断会通过 OnTcpConnectionClosed 通知 TcpServerHandle。

3)TcpConnection 继承自 TcpConnectionHandle,收到 TCP 报文会回调连接监听者。

4)当前只有 WebRtcServer 和 WebRtcTransport 支持 TCP 通信。

【注】WebRtcServer 用来实现端口聚合,其上可以承载多个 WebRtcTransport。

2.2. Socket 创建

2.2.1. WebRtcServer

WebRtcServer 用来实现 WebRTC 连接的端口聚合,WebRtcTransport 可以运行在 WebRtcServer 之上,共享 WebRtcServer 的端口。

WebRtcServer 根据传入的参数,决定创建 UdpSocket 还是 TcpServer,支持指定端口或端口范围。

WebRtcServer::WebRtcServer(RTC::Shared* shared, const std::string& id,const flatbuffers::Vector<flatbuffers::Offset<Transport::ListenInfo>>* listenInfos): id(id), shared(shared)
{...// 遍历所有地址for (const auto* listenInfo : *listenInfos){auto ip = listenInfo->ip()->str();...// UDP 协议if (listenInfo->protocol() == FBS::Transport::Protocol::UDP){RTC::UdpSocket* udpSocket;// 指定端口范围,从中选择一个if (listenInfo->portRange()->min() != 0 && listenInfo->portRange()->max() != 0){uint64_t portRangeHash{ 0u };udpSocket = new RTC::UdpSocket(this,ip,listenInfo->portRange()->min(),listenInfo->portRange()->max(),flags,portRangeHash);}// 指定端口else if (listenInfo->port() != 0){udpSocket = new RTC::UdpSocket(this, ip, listenInfo->port(), flags);}// 未指定端口,使用配置中的端口else{uint64_t portRangeHash{ 0u };udpSocket = new RTC::UdpSocket(this,ip,Settings::configuration.rtcMinPort,Settings::configuration.rtcMaxPort,flags,portRangeHash);}...}// TCP 协议else if (listenInfo->protocol() == FBS::Transport::Protocol::TCP){RTC::TcpServer* tcpServer;// 指定端口范围if (listenInfo->portRange()->min() != 0 && listenInfo->portRange()->max() != 0){uint64_t portRangeHash{ 0u };tcpServer = new RTC::TcpServer(this,this,ip,listenInfo->portRange()->min(),listenInfo->portRange()->max(),flags,portRangeHash);}// 指定端口else if (listenInfo->port() != 0){tcpServer = new RTC::TcpServer(this, this, ip, listenInfo->port(), flags);}// 未指定端口,使用配置中的端口else{uint64_t portRangeHash{ 0u };tcpServer = new RTC::TcpServer(this,this,ip,Settings::configuration.rtcMinPort,Settings::configuration.rtcMaxPort,flags,portRangeHash);}...}}...
}

2.2.2. WebRtcTransport

如果 WebRtcTransport 运行在 WebRtcServer 之上,则 WebRtcTransport 不会再创建 Socket。

WebRtcTransport::WebRtcTransport(...)
{...// 将 WebRtcTransport 加入到 WebRtcServer 的转发列表this->webRtcTransportListener->OnWebRtcTransportCreated(this);...
}

否则,还需自食其力,WebRtcTransport  创建 Socket 的逻辑与 WebRtcServer 类似,不再赘述。

2.2.3. PlainTransport

PlainTransport 用来对接像 FFMPEG 这种第三方编码器和工具的推拉流, 只支持 UDP 协议,创建逻辑类似,也支持指定端口或端口范围。

PipeTransport::PipeTransport(RTC::Shared* shared,const std::string& id,RTC::Transport::Listener* listener,const FBS::PipeTransport::PipeTransportOptions* options): RTC::Transport::Transport(shared, id, listener, options->base())
{...// 指定端口范围if (this->listenInfo.portRange.min != 0 && this->listenInfo.portRange.max != 0){uint64_t portRangeHash{ 0u };this->udpSocket = new RTC::UdpSocket(this,this->listenInfo.ip,this->listenInfo.portRange.min,this->listenInfo.portRange.max,this->listenInfo.flags,portRangeHash);}// 指定端口else if (this->listenInfo.port != 0){this->udpSocket = new RTC::UdpSocket(this, this->listenInfo.ip, this->listenInfo.port, this->listenInfo.flags);}// 未指定端口,使用配置else{uint64_t portRangeHash{ 0u };this->udpSocket = new RTC::UdpSocket(this,this->listenInfo.ip,Settings::configuration.rtcMinPort,Settings::configuration.rtcMaxPort,this->listenInfo.flags,portRangeHash);}...
}

2.2.4. PipeTransport

PipeTransport 的设计目的是为了使位于同一主机上或不同主机上的两个Router实例之间进行通信,只支持 UDP 协议,创建逻辑类似,也支持指定端口或端口范围。

PipeTransport::PipeTransport(RTC::Shared* shared,const std::string& id,RTC::Transport::Listener* listener,const FBS::PipeTransport::PipeTransportOptions* options): RTC::Transport::Transport(shared, id, listener, options->base())
{...if (this->listenInfo.portRange.min != 0 && this->listenInfo.portRange.max != 0){uint64_t portRangeHash{ 0u };this->udpSocket = new RTC::UdpSocket(this,this->listenInfo.ip,this->listenInfo.portRange.min,this->listenInfo.portRange.max,this->listenInfo.flags,portRangeHash);}else if (this->listenInfo.port != 0){this->udpSocket = new RTC::UdpSocket(this, this->listenInfo.ip, this->listenInfo.port, this->listenInfo.flags);}else{uint64_t portRangeHash{ 0u };this->udpSocket = new RTC::UdpSocket(this,this->listenInfo.ip,Settings::configuration.rtcMinPort,Settings::configuration.rtcMaxPort,this->listenInfo.flags,portRangeHash);}...
}

2.3. 数据流

2.3.1. UDP

2.3.1.1. 接收数据

以 WebRtcServer 为例,libuv 收到 UDP 消息会回调 UdpSocketHandle::OnUvRecv,UdpSocketHandle 再回调 UdpSocket::UserOnUdpDatagramReceived,最终将消息回调给数据监听者 WebRtcServer。

2.3.1.2. 发送数据

需要发送 UDP 消息的模块持有 TransportTuple 对象,调用 TransportTuple:: Send 方法,内部调用 UdpSocketHandle::Send,最终通过 libuv 接口将数据发送到网络。

需要注意,UDP 报文发送有一个特殊机制,mediaoup 会先调用 libuv 同步发送接口,如果同步发送接口出错,mediasoup 不是立即返回,而是拷贝发送数据,继续调用 libuv 的异步发送接口。这在某些极端场景下,可能会大量消耗服务器内存。

void UdpSocketHandle::Send(const uint8_t* data, size_t len, const struct sockaddr* addr,   UdpSocketHandle::onSendCallback* cb)
{...// 使用待发送送数据初始化一块uv缓冲区uv_buf_t buffer = uv_buf_init(reinterpret_cast<char*>(const_cast<uint8_t*>(data)), len);// 调用同步接口发送const int sent  = uv_udp_try_send(this->uvHandle, &buffer, 1, addr);// 所有数据都发送完成if (sent == static_cast<int>(len)){// Update sent bytes.this->sentBytes += sent;if (cb){(*cb)(true); // 回调返回成功delete cb;}return;}// 发送了部分数据else if (sent >= 0){this->sentBytes += sent;if (cb){(*cb)(false); // 回调返回失败delete cb;}return;}// 出错了,可能是网络繁忙,使用异步接口uv_udp_send发送else if (sent != UV_EAGAIN){MS_WARN_DEV("uv_udp_try_send() failed, trying uv_udp_send(): %s", uv_strerror(sent));}// 创建一个异步处理数据结构auto* sendData = new UvSendData(len);// 作为自定义数据挂载到uv数据结构中sendData->req.data = static_cast<void*>(sendData);// 拷贝待发送数据std::memcpy(sendData->store, data, len);// 保存回调函数指针sendData->cb = cb;// 使用待发送数据的拷贝初始化uv缓冲区buffer = uv_buf_init(reinterpret_cast<char*>(sendData->store), len);// 调用异步接口发送,设置回调接口onSendconst int err = uv_udp_send(&sendData->req, this->uvHandle, &buffer, 1, addr, static_cast<uv_udp_send_cb>(onSend));if (err != 0){if (cb){(*cb)(false);}delete sendData;}else{this->sentBytes += len;}
}

UvSendData 定义如下:

struct UvSendData
{uv_udp_send_t req{};uint8_t* store{ nullptr };UdpSocketHandle::onSendCallback* cb{ nullptr };
};

libuv 发送完成后会回调 onSend,在 onSend 函数中处理善后事宜。

inline static void onSend(uv_udp_send_t* req, int status)
{auto* sendData = static_cast<UdpSocketHandle::UvSendData*>(req->data);auto* handle   = req->handle;auto* socket   = static_cast<UdpSocketHandle*>(handle->data);const auto* cb = sendData->cb;if (socket){socket->OnUvSend(status, cb);}// Delete the UvSendData struct (it will delete the store and cb too).delete sendData;
}

2.3.2. TCP

2.3.2.1. 监听连接

1)TcpServer 调用 libuv 接口建立监听。

2)客户端与服务器完成三次握手后,libuv 会回调 TcpServerHandle::OnUvConnection。

3)TcpServerHandle 回调 TcpServer::UserOnTcpConnectionAlloc。

4)TcpServer 创建 TcpConnection 并调用 TcpServerHandle::AcceptTcpConnection 告知要接受这个连接。

5)TcpServerHandle 对 TcpConnection 进行初始化,调用 libuv 的 uv_accpet 方法完成新连接的创建。

6)调用 TcpConnectionHandle::Start 开始接收数据。

2.3.2.2. 接收数据

接收数据逻辑非常简单,以 WebRtcServer 为例,libuv 收到 TCP 数据后会层层回调到 WebRtcServer。

2.3.2.3. 发送数据

发送 TCP 数据的逻辑也很简单,调用 TransportTuple 接口,内部最终调用 libuv 将数据发送到网络。

3. 定时器

定时器在很多地方都会被用到,mediasoup 使用 TimerHanlde 封装 libuv 的定时器能力。需要使用定时器的类需要继承 TimerHandle::Listener,实现 OnTimer 虚拟方法。然后创建一个 TimerHandle 对象,传入 this 指针,调用 TimerHandle::Start 方法启动定时器即可。

4. 信号处理

信号是进程间通信的一种机制,也是操作系统用来通知进程有关系统事件或异常状况的重要手段。信号可以由系统内核发送给进程,也可以由一个进程发送给另一个进程。在 Worker 进程中,Worker 类是唯一处理 signal 的类,它继承 SignalHandle::Listener,实现 OnSignal 虚拟方法,进程接收的所有信号都会回调给 Worker 处理。

mediasoup 当前只处理了 SIGINT 和 SIGTERM 两个信号,用来优雅的关闭 mediasoup 进程。

void Worker::OnSignal(SignalHandle* /*signalHandle*/, int signum)
{if (this->closed){return;}switch (signum){case SIGINT:{if (this->closed){return;}Close();break;}case SIGTERM:{if (this->closed){return;}Close();break;}default:{MS_WARN_DEV("received a non handled signal [signum:%d]", signum);}}
}

5. 总结

熟悉 mediasoup 的底层通信机制,是深入阅读 mediasoup 源码的基础。本文详细描述了 mediasoup 对 libuv 的封装,覆盖了 pipe、socket、signal 等几种通信方式,重点分析了 Socket 通信的静态结构和数据流,补充分析了 UDP 报文的异步发送机制。mediasoup 对 libuv 的封装简洁清晰,是一个优秀的设计方案,值得大家借鉴。

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

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

使用 spring MVC 简单的案例 (1)计算器

一、计算器 1.1前端代码 <!DOCTYPE html> <html lang"en"> <head><meta charset"UTF-8"><meta name"viewport" content"widthdevice-width, initial-scale1.0"><title>Document</title> …

Git报错fatal: detected dubious ownership in repository

报错信息 fatal: detected dubious ownership in repository at 解决办法 一行代码解决 git config --global --add safe.directory "*";如何使用git工具初始胡项目并且和远程仓库建立联系 git init–建立一个本地仓库 git add README.md–将README.md文件加入…

MySQL添加索引时会锁表吗?

目录 简介Online DDL概念Online DDL用法总结 简介 在MySQL5.5以及之前的版本&#xff0c;通常更改数据表结构操作&#xff08;DDL&#xff09;会阻塞对表数据的增删改操作&#xff08;DML&#xff09;。 MySQL5.6提供Online DDL之后可支持DDL与DML操作同时执行&#xff0c;降低…

算法通关:005对数器

就是你有优解&#xff0c;但是不知道对不对&#xff0c;或者你遇到了题&#xff0c;但是没有在线网站能跑&#xff0c;无法检查你的思路是否正确。 写一个随机生成符合输入要求的方法。 此时用暴力解法写一个&#xff0c;因为答案肯定是对的&#xff0c;再写一个优解方法。将两…

斐波那契数列的多种解法 C++实现,绘图部分用Python实现

斐波那契数列的多种解法 C实现&#xff0c;绘图部分用Python实现 flyfish 斐波那契数列&#xff08;Fibonacci sequence&#xff09;是一个经典的数列&#xff0c;定义如下&#xff1a; { 0 if n 0 1 if n 1 F ( n − 1 ) F ( n − 2 ) if n > 1 \begin{cases} 0 &…

HackTheBox--Knife

Knife 测试过程 1 信息收集 端口扫描 80端口测试 echo "10.129.63.56 knife.htb" | sudo tee -a /etc/hosts网站是纯静态的&#xff0c;无任何交互功能&#xff0c;检查网页源代码也未发现任何可利用的文件。 检查页面请求时&#xff0c;请求与响应内容&#xff0…

高频面试题-CSS

BFC 介绍下BFC (块级格式化上下文) 1>什么是BFC BFC即块级格式化上下文&#xff0c;是CSS可视化渲染的一部分, 它是一块独立的渲染区域&#xff0c;只有属于同一个BFC的元素才会互相影响&#xff0c;且不会影响其它外部元素。 2>如何创建BFC 根元素&#xff0c;即HTM…

RabbitMQ的学习和模拟实现|sqlite轻量级数据库的介绍和简单使用

SQLite3 项目仓库&#xff1a;https://github.com/ffengc/HareMQ SQLite3 什么是SQLite为什么需要用SQLite官方文档封装Helper进行一些实验 什么是SQLite SQLite是一个进程内的轻量级数据库&#xff0c;它实现了自给自足的、无服务器的、零配置的、事务性的 SQL数据库引擎…

lua 游戏架构 之 LoaderWallet 异步加载

定义了一个名为LoaderWallet class&#xff0c;用于管理资源加载器&#xff08;Loader&#xff09;。这个类封装了资源加载的功能&#xff0c;包括异步加载&#xff0c;以及资源的释放和状态查询。下面是对代码的详细解释&#xff1a; ### 类定义和初始化 这里定义了一个名为…

Inconsistent Query Results Based on Output Fields Selection in Milvus Dashboard

题意&#xff1a;在Milvus仪表盘中基于输出字段选择的不一致查询结果 问题背景&#xff1a; Im experiencing an issue with the Milvus dashboard where the search results change based on the selected output fields. Im working on a RAG project using text data conv…

AndroidStudio 编辑xml布局文件卡死问题解决

之前项目编写的都是正常&#xff0c;升级AndroidStudio后编辑布局文件就卡死&#xff0c;还以为是AndroidStudio文件。 其实不然&#xff0c;我给整个项目增加了版权声明。所以全部跟新后&#xff0c;布局文件也增加了版权声明。估计AndroidStudio在 解析布局文件时候因为有版…

推荐丨SSL证书是什么?该怎么申请,需要准备哪些材料?

SSL证书是什么&#xff1f; SSL证书&#xff08;Secure Sockets Layer Certificate&#xff09;&#xff0c;又称为数字证书&#xff0c;是一种用于在互联网上验证网站身份和加密通信的技术。它遵守SSL协议&#xff0c;由受信任的数字证书颁发机构&#xff08;CA&#xff09;在…

在 CI/CD Pipeline 中实施持续测试的最佳实践!

随着软件开发周期的不断加快&#xff0c;持续集成&#xff08;CI&#xff09;和持续交付/部署&#xff08;CD&#xff09;已经成为现代软件开发的重要组成部分。在这一过程中&#xff0c;持续测试的实施对于确保代码质量、提高发布效率至关重要。本文将详细介绍在CI/CD流水线中…

STM32高级运动控制系统教程

目录 引言环境准备高级运动控制系统基础代码实现&#xff1a;实现高级运动控制系统 4.1 传感器数据采集模块 4.2 数据处理与运动控制模块 4.3 通信与网络系统实现 4.4 用户界面与数据可视化应用场景&#xff1a;运动控制与优化问题解决方案与优化收尾与总结 1. 引言 高级运动…

深入理解Linux网络(五):TCP接收唤醒

深入理解Linux网络&#xff08;五&#xff09;&#xff1a;TCP接收唤醒 TCP接收唤醒由软中断提供服务。 软中断&#xff08;也就是 Linux ⾥的 ksoftirqd 进程&#xff09;⾥收到数据包以后&#xff0c;发现是 tcp 的包的话就会执⾏到 tcp_v4_rcv 函数。接着如果是 ESTABLISH…

mysql JSON特性优化

有朋友问到&#xff0c;mysql如果要根据json中的某个属性过滤&#xff0c;数据量大的话&#xff0c;性能很差&#xff0c;要如何提高性能&#xff1f; 为什么要用json串&#xff1f; 由于一些特定场景&#xff0c;mysql需要用到json串&#xff0c;例如文档&#xff0c;不同的…

【LabVIEW作业篇 - 5】:水仙花数、数组与for循环的连接

文章目录 水仙花数数组与for循环的连接 水仙花数 水仙花数&#xff0c;是指一个3位数&#xff0c;它的每个位上的数字的3次幂之和等于它本身。如371 3^3 7^3 1^3&#xff0c;则371是一个水仙花数。 思路&#xff1a;水仙花数是一个三位数&#xff0c;通过使用for循环&#xf…

RabbitMQ的学习和模拟实现|muduo库的介绍和使用

muduo库 项目仓库&#xff1a;https://github.com/ffengc/HareMQ muduo库 muduo库是什么快速上手搭建服务端快速上手搭建客户端上面搭建的服务端-客户端通信还有什么问题?muduo库中的protobuf基于muduo库中的protobuf协议实现一个服务器 muduo库是什么 Muduo由陈硕大佬开…

ReadAgent,一款具有要点记忆的人工智能阅读代理

人工智能咨询培训老师叶梓 转载标明出处 现有的大模型&#xff08;LLMs&#xff09;在处理长文本时受限于固定的最大上下文长度&#xff0c;并且当输入文本越来越长时&#xff0c;性能往往会下降&#xff0c;即使在没有超出明确上下文窗口的情况下&#xff0c;LLMs 的性能也会随…

pytorch 笔记:torch.optim.Adam

torch.optim.Adam 是一个实现 Adam 优化算法的类。Adam 是一个常用的梯度下降优化方法&#xff0c;特别适合处理大规模数据集和参数的深度学习模型 torch.optim.Adam(params, lr0.001, betas(0.9, 0.999), eps1e-08, weight_decay0, amsgradFalse, *, foreachNone, maximizeFa…