音视频入门基础：RTP专题（9）——FFmpeg接收RTP流的原理和内部实现
一、引言
由《音视频入门基础：RTP专题（2）——使用FFmpeg命令生成RTP流》可以知道，推流端通过下面FFmpeg命令可以将一个媒体文件转推RTP，生成RTP流：
ffmpeg -re -stream_loop -1 -i input.mp4 -vcodec copy -an -f rtp rtp://192.168.0.103:6005 -acodec copy -vn -sdp_file XXX.sdp -f rtp rtp://192.168.0.103:7005接收端通过命令：ffmpeg -protocol_whitelist "file,rtp,udp" -i XXX.sdp 可以查看生成的RTP流的信息：
由《音视频入门基础：RTP专题（8）——使用Wireshark分析RTP》可以知道，上述推流端的本质是创建了一个UDP客户端，将媒体文件input.mp4的视频数据发送到IP为192.168.0.103的UDP服务器的6005端口，音频数据发送到该UDP服务器的7005端口。与之对应，接收端的本质是创建了一个UDP服务器来接收推流端发送的基于UDP的RTP数据。下面讲述接收端的FFmpeg，其源码中接收RTP流的内部实现。
二、FFmpeg接收RTP流的内部实现
（一）获取端口
由《音视频入门基础：RTP专题（3）——SDP简介》可以知道，SDP中某行<type>的值为'm'时，<value>会包含媒体描述信息，此时该行SDP的格式为：m=<media> <port> <proto> <fmt> ...。如果音视频数据为RTP，那<port>为接收端需要创建的UDP服务器的端口号，该端口用于接收推流端发送的基于UDP的RTP数据。
由《音视频入门基础：RTP专题（5）——FFmpeg源码中，解析SDP的实现》可以知道，FFmpeg源码中通过ff_sdp_parse函数解析SDP，而ff_sdp_parse函数中又会通过sdp_parse_line函数解析SDP中的一行数据，所以sdp_read_header函数中执行完ff_sdp_parse函数后，变量rtsp_st->sdp_port会得到<port>的信息。然后sdp_read_header函数中会通过语句：ff_url_join(url, sizeof(url), "rtp", NULL,namebuf, rtsp_st->sdp_port,"?localport=%d&ttl=%d&connect=%d&write_to_source=%d",rtsp_st->sdp_port, rtsp_st->sdp_ttl,rt->rtsp_flags & RTSP_FLAG_FILTER_SRC ? 1 : 0,rt->rtsp_flags & RTSP_FLAG_RTCP_TO_SOURCE ? 1 : 0) 将包含<port>信息的字符串拷贝到url数组中：
static int sdp_read_header(AVFormatContext *s)
{
RTSPState *rt = s->priv_data;
char url[MAX_URL_SIZE]
//...
err = ff_sdp_parse(s, bp.str);
av_bprint_finalize(&bp, NULL);
if (err) goto fail;
/* open each RTP stream */
for (i = 0; i < rt->nb_rtsp_streams; i++) {
char namebuf[50];
rtsp_st = rt->rtsp_streams[i];
if (!(rt->rtsp_flags & RTSP_FLAG_CUSTOM_IO)) {
AVDictionary *opts = map_to_opts(rt);
char buf[MAX_URL_SIZE];
const char *p;
err = getnameinfo((struct sockaddr*) &rtsp_st->sdp_ip,
sizeof(rtsp_st->sdp_ip),
namebuf, sizeof(namebuf), NULL, 0, NI_NUMERICHOST);
if (err) {
av_log(s, AV_LOG_ERROR, "getnameinfo: %s\n", gai_strerror(err));
err = AVERROR(EIO);
av_dict_free(&opts);
goto fail;
}
ff_url_join(url, sizeof(url), "rtp", NULL,
namebuf, rtsp_st->sdp_port,
"?localport=%d&ttl=%d&connect=%d&write_to_source=%d",
rtsp_st->sdp_port, rtsp_st->sdp_ttl,
rt->rtsp_flags & RTSP_FLAG_FILTER_SRC ? 1 : 0,
rt->rtsp_flags & RTSP_FLAG_RTCP_TO_SOURCE ? 1 : 0);
p = strchr(s->url, '?');
if (p && av_find_info_tag(buf, sizeof(buf), "localaddr", p))
av_strlcatf(url, sizeof(url), "&localaddr=%s", buf);
else if (rt->localaddr && rt->localaddr[0])
av_strlcatf(url, sizeof(url), "&localaddr=%s", rt->localaddr);
append_source_addrs(url, sizeof(url), "sources",
rtsp_st->nb_include_source_addrs,
rtsp_st->include_source_addrs);
append_source_addrs(url, sizeof(url), "block",
rtsp_st->nb_exclude_source_addrs,
rtsp_st->exclude_source_addrs);
err = ffurl_open_whitelist(&rtsp_st->rtp_handle, url, AVIO_FLAG_READ,
&s->interrupt_callback, &opts, s->protocol_whitelist, s->protocol_blacklist, NULL);
av_dict_free(&opts);
if (err < 0) {
err = AVERROR_INVALIDDATA;
goto fail;
}
}
if ((err = ff_rtsp_open_transport_ctx(s, rtsp_st)))
goto fail;
}
//..
}然后sdp_read_header中会执行语句：err = ffurl_open_whitelist(&rtsp_st->rtp_handle, url, AVIO_FLAG_READ,&s->interrupt_callback, &opts, s->protocol_whitelist, s->protocol_blacklist, NULL)。ffurl_open_whitelist函数中会通过语句：int ret = ffurl_alloc(puc, filename, flags, int_cb)将上述url数组中的字符串拷贝到(*puc)->filename中。然后ffurl_open_whitelist函数中会调用ffurl_connect函数：
int ffurl_open_whitelist(URLContext **puc, const char *filename, int flags,
const AVIOInterruptCB *int_cb, AVDictionary **options,
const char *whitelist, const char* blacklist,
URLContext *parent)
{
//...
int ret = ffurl_alloc(puc, filename, flags, int_cb);
//...
ret = ffurl_connect(*puc, options);
//...
}ffurl_connect函数中会执行函数指针url_open指向的回调函数。当音视频流为RTP流时，对应的回调函数就是rtp_open函数：
int ffurl_connect(URLContext *uc, AVDictionary **options)
{
//...
err =
uc->prot->url_open2 ? uc->prot->url_open2(uc,
uc->filename,
uc->flags,
options) :
uc->prot->url_open(uc, uc->filename, uc->flags);
//...
}rtp_open函数的底层又会调用udp_open函数，udp_open函数中会通过下面语句将上述字符串中的<port>信息提取出来，赋值给变量s->local_port：
static int udp_open(URLContext *h, const char *uri, int flags)
{
//...
p = strchr(uri, '?');
if (p) {
//...
if (av_find_info_tag(buf, sizeof(buf), "localport", p)) {
s->local_port = strtol(buf, NULL, 10);
//...
}
}
//...
}总之，经过一系列繁琐的赋值和拷贝操作后，变量s->local_port最终得到接收端需要创建的UDP服务器的端口号，该端口来源于推流端的FFmpeg命令生成的SDP文件中的端口信息。
（二）给addrinfo结构赋值
获取完端口后，udp_open函数内部会调用udp_socket_create函数：
/* put it in UDP context */
/* return non zero if error */
static int udp_open(URLContext *h, const char *uri, int flags)
{
//...
udp_fd = udp_socket_create(h, &my_addr, &len, s->localaddr);
//...
}udp_socket_create函数内部会通过语句：
res0 = ff_ip_resolve_host(h, (localaddr && localaddr[0]) ? localaddr : NULL,s->local_port,SOCK_DGRAM, family, AI_PASSIVE) 根据给定的主机名和需要创建的UDP服务器的端口，返回一个struct addrinfo结构，从而给addrinfo结构赋值。其中localaddr包含主机名，s->local_port为上述讲到的需要创建的UDP服务器的端口：
static int udp_socket_create(URLContext *h, struct sockaddr_storage *addr,
socklen_t *addr_len, const char *localaddr)
{
UDPContext *s = h->priv_data;
int udp_fd = -1;
struct addrinfo *res0, *res;
int family = AF_UNSPEC;
if (((struct sockaddr *) &s->dest_addr)->sa_family)
family = ((struct sockaddr *) &s->dest_addr)->sa_family;
res0 = ff_ip_resolve_host(h, (localaddr && localaddr[0]) ? localaddr : NULL,
s->local_port,
SOCK_DGRAM, family, AI_PASSIVE);
if (!res0)
goto fail;
for (res = res0; res; res=res->ai_next) {
if (s->udplite_coverage)
udp_fd = ff_socket(res->ai_family, SOCK_DGRAM, IPPROTO_UDPLITE, h);
else
udp_fd = ff_socket(res->ai_family, SOCK_DGRAM, 0, h);
if (udp_fd != -1) break;
ff_log_net_error(h, AV_LOG_ERROR, "socket");
}
if (udp_fd < 0)
goto fail;
memcpy(addr, res->ai_addr, res->ai_addrlen);
*addr_len = res->ai_addrlen;
freeaddrinfo(res0);
return udp_fd;
fail:
if (udp_fd >= 0)
closesocket(udp_fd);
if(res0)
freeaddrinfo(res0);
return -1;
}ff_ip_resolve_host函数定义在libavformat/ip.c中，可以看到其内部通过getaddrinfo函数根据给定的主机名和服务名，返回一个struct addrinfo结构。关于getaddrinfo函数的用法可以参考：《百度百科——getaddrinfo》：
struct addrinfo *ff_ip_resolve_host(void *log_ctx,
const char *hostname, int port,
int type, int family, int flags)
{
struct addrinfo hints = { 0 }, *res = 0;
int error;
char sport[16];
const char *node = 0, *service = "0";
if (port > 0) {
snprintf(sport, sizeof(sport), "%d", port);
service = sport;
}
if ((hostname) && (hostname[0] != '\0') && (hostname[0] != '?')) {
node = hostname;
}
hints.ai_socktype = type;
hints.ai_family = family;
hints.ai_flags = flags;
if ((error = getaddrinfo(node, service, &hints, &res))) {
res = NULL;
av_log(log_ctx, AV_LOG_ERROR, "getaddrinfo(%s, %s): %s\n",
node ? node : "unknown",
service,
gai_strerror(error));
}
return res;
}执行完上述操作后，udp_open函数中的变量my_addr会得到主机名和需要创建的UDP服务器的端口信息：
/* put it in UDP context */
/* return non zero if error */
static int udp_open(URLContext *h, const char *uri, int flags)
{
//...
udp_fd = udp_socket_create(h, &my_addr, &len, s->localaddr);
//...
}（三）创建套接字
给addrinfo结构赋值后，udp_socket_create函数内部会执行语句：udp_fd = ff_socket(res->ai_family, SOCK_DGRAM, 0, h) 来创建套接字，SOCK_DGRAM表示是基于UDP：
static int udp_socket_create(URLContext *h, struct sockaddr_storage *addr,
socklen_t *addr_len, const char *localaddr)
{
//...
res0 = ff_ip_resolve_host(h, (localaddr && localaddr[0]) ? localaddr : NULL,
s->local_port,
SOCK_DGRAM, family, AI_PASSIVE);
//...
for (res = res0; res; res=res->ai_next) {
if (s->udplite_coverage)
udp_fd = ff_socket(res->ai_family, SOCK_DGRAM, IPPROTO_UDPLITE, h);
else
udp_fd = ff_socket(res->ai_family, SOCK_DGRAM, 0, h);
if (udp_fd != -1) break;
ff_log_net_error(h, AV_LOG_ERROR, "socket");
}
//...
}ff_socket函数定义在libavformat/network.c中，可以看到该函数内部通过socket函数创建了套接字。音视频流为RTP的情况下，形参type的值为SOCK_DGRAM，所以此时创建的是UDP套接字：
int ff_socket(int af, int type, int proto, void *logctx)
{
int fd;
#ifdef SOCK_CLOEXEC
fd = socket(af, type | SOCK_CLOEXEC, proto);
if (fd == -1 && errno == EINVAL)
#endif
{
fd = socket(af, type, proto);
#if HAVE_FCNTL
if (fd != -1) {
if (fcntl(fd, F_SETFD, FD_CLOEXEC) == -1)
av_log(logctx, AV_LOG_DEBUG, "Failed to set close on exec\n");
}
#endif
}
#ifdef SO_NOSIGPIPE
if (fd != -1) {
if (setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &(int){1}, sizeof(int))) {
av_log(logctx, AV_LOG_WARNING, "setsockopt(SO_NOSIGPIPE) failed\n");
}
}
#endif
return fd;
}（四）绑定套接字
创建完UDP套接字后，udp_open函数中会通过bind函数绑定上述创建的UDP套接字到my_addr上。从上面我们已经可以知道，my_addr包含需要创建的UDP服务器的端口信息：
/* put it in UDP context */
/* return non zero if error */
static int udp_open(URLContext *h, const char *uri, int flags)
{
//...
udp_fd = udp_socket_create(h, &my_addr, &len, s->localaddr);
//...
/* bind to the local address if not multicast or if the multicast
* bind failed */
/* the bind is needed to give a port to the socket now */
if (bind_ret < 0 && bind(udp_fd,(struct sockaddr *)&my_addr, len) < 0) {
ff_log_net_error(h, AV_LOG_ERROR, "bind failed");
ret = ff_neterrno();
goto fail;
}
//...
}
（五）监视文件描述符是否可读并接收数据
绑定完套接字后，FFmpeg的avformat_find_stream_info函数底层会调用rtp_read函数，而rtp_read函数内部会通过poll函数监视文件描述符（上述创建的UDP套接字）是否可读。如果可读，那就通过recvfrom函数接收UDP数据（基于UDP的RTP音视频数据），将接收到的数据存放到形参buf指向的缓冲区中：
static int rtp_read(URLContext *h, uint8_t *buf, int size)
{
RTPContext *s = h->priv_data;
int len, n, i;
struct pollfd p[2] = {{s->rtp_fd, POLLIN, 0}, {s->rtcp_fd, POLLIN, 0}};
int poll_delay = h->flags & AVIO_FLAG_NONBLOCK ? 0 : POLLING_TIME;
struct sockaddr_storage *addrs[2] = { &s->last_rtp_source, &s->last_rtcp_source };
socklen_t *addr_lens[2] = { &s->last_rtp_source_len, &s->last_rtcp_source_len };
int runs = h->rw_timeout / 1000 / POLLING_TIME;
for(;;) {
if (ff_check_interrupt(&h->interrupt_callback))
return AVERROR_EXIT;
n = poll(p, 2, poll_delay);
if (n > 0) {
/* first try RTCP, then RTP */
for (i = 1; i >= 0; i--) {
if (!(p[i].revents & POLLIN))
continue;
*addr_lens[i] = sizeof(*addrs[i]);
len = recvfrom(p[i].fd, buf, size, 0,
(struct sockaddr *)addrs[i], addr_lens[i]);
if (len < 0) {
if (ff_neterrno() == AVERROR(EAGAIN) ||
ff_neterrno() == AVERROR(EINTR))
continue;
return AVERROR(EIO);
}
if (ff_ip_check_source_lists(addrs[i], &s->filters))
continue;
return len;
}
} else if (n == 0 && h->rw_timeout > 0 && --runs <= 0) {
return AVERROR(ETIMEDOUT);
} else if (n < 0) {
if (ff_neterrno() == AVERROR(EINTR))
continue;
return AVERROR(EIO);
}
if (h->flags & AVIO_FLAG_NONBLOCK)
return AVERROR(EAGAIN);
}
}三、总结
1.从上面的代码分析可以看出来，接收端的FFmpeg接收RTP数据，其原理就是创建了一个UDP服务器来接收推流端发送的基于UDP的RTP数据，本质就是接收UDP数据。该UDP服务器的实现原理跟《Linux下使用poll函数编写UDP客户端、服务器程序》中展示的UDP服务器是一样的，都是调用了socket、bind、poll、recvfrom这几个函数。
2.跟普通的UDP服务器相比，FFmpeg接收RTP流时创建的UDP服务器，其端口来源于推流端的FFmpeg命令生成的SDP文件中的端口信息。也就是说此时的流程是推流端的UDP客户端先被创建，然后通过SDP把需要创建的UDP服务器的端口号发送给接收端，最后接收端才根据这个端口号创建UDP服务器。所以可以实现推流端先推流（UDP客户端先发送音视频数据，不管服务器有没有接收到），接收端再接收RTP音视频数据。