RTP proxy通过UDP收流，调整udp recv socket buffer size成配置 (#3336)

国标推流有些情况需要UDP方式接收流，端口复用同一个UDP端口可能需要根据服务器性能

conf/config.ini

@@ -332,6 +332,9 @@ gop_cache=1
 #国标发送g711 rtp 打包时，每个包的语音时长是多少，默认是100 ms，范围为20~180ms (gb28181-2016，c.2.4规定)，
 #最好为20 的倍数，程序自动向20的倍数取整
 rtp_g711_dur_ms = 100
+#udp接收数据socket buffer大小配置
+#4*1024*1024=4196304
+udp_recv_socket_buffer=4194304
 
 [rtc]
 #rtc播放推流、播放超时时间

src/Common/config.cpp

@@ -345,6 +345,7 @@ const string kPSPT = RTP_PROXY_FIELD "ps_pt";
 const string kOpusPT = RTP_PROXY_FIELD "opus_pt";
 const string kGopCache = RTP_PROXY_FIELD "gop_cache";
 const string kRtpG711DurMs = RTP_PROXY_FIELD "rtp_g711_dur_ms";
+const string kUdpRecvSocketBuffer = RTP_PROXY_FIELD "udp_recv_socket_buffer";
 
 static onceToken token([]() {
     mINI::Instance()[kDumpDir] = "";
@@ -356,6 +357,7 @@ static onceToken token([]() {
     mINI::Instance()[kOpusPT] = 100;
     mINI::Instance()[kGopCache] = 1;
     mINI::Instance()[kRtpG711DurMs] = 100;
+    mINI::Instance()[kUdpRecvSocketBuffer] = 4 * 1024 * 1024;
 });
 } // namespace RtpProxy
 

src/Common/config.h

@@ -400,6 +400,8 @@ extern const std::string kGopCache;
 //国标发送g711 rtp 打包时，每个包的语音时长是多少，默认是100 ms，范围为20~180ms (gb28181-2016，c.2.4规定)，
 //最好为20 的倍数，程序自动向20的倍数取整
 extern const std::string kRtpG711DurMs;
+// udp recv socket buffer size
+extern const std::string kUdpRecvSocketBuffer;
 } // namespace RtpProxy
 
 /**

src/Rtp/RtpServer.cpp

@@ -174,7 +174,8 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_
     }
 
     //设置udp socket读缓存
-    SockUtil::setRecvBuf(rtp_socket->rawFD(), 4 * 1024 * 1024);
+    GET_CONFIG(int, udpRecvSocketBuffer, RtpProxy::kUdpRecvSocketBuffer);
+    SockUtil::setRecvBuf(rtp_socket->rawFD(), udpRecvSocketBuffer);
 
     TcpServer::Ptr tcp_server;
     _tcp_mode = tcp_mode;
@@ -223,6 +224,7 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_
         //单端口多线程接收多个流，根据ssrc区分流
         udp_server = std::make_shared<UdpServer>(rtp_socket->getPoller());
         (*udp_server)[RtpSession::kOnlyAudio] = only_audio;
+        (*udp_server)[RtpSession::kUdpRecvBuffer] = udpRecvSocketBuffer;
         udp_server->start<RtpSession>(local_port, local_ip);
         rtp_socket = nullptr;
     }

src/Rtp/RtpSession.cpp

@@ -24,6 +24,7 @@ namespace mediakit{
 const string RtpSession::kStreamID = "stream_id";
 const string RtpSession::kSSRC = "ssrc";
 const string RtpSession::kOnlyAudio = "only_audio";
+const string RtpSession::kUdpRecvBuffer = "udp_recv_socket_buffer";
 
 void RtpSession::attachServer(const Server &server) {
     setParams(const_cast<Server &>(server));
@@ -33,6 +34,12 @@ void RtpSession::setParams(mINI &ini) {
     _stream_id = ini[kStreamID];
     _ssrc = ini[kSSRC];
     _only_audio = ini[kOnlyAudio];
+    int udp_socket_buffer = ini[kUdpRecvBuffer];
+    if (_is_udp) {
+        // 设置udp socket读缓存
+        SockUtil::setRecvBuf(getSock()->rawFD(),
+            (udp_socket_buffer > 0) ? udp_socket_buffer : (4 * 1024 * 1024));
+    }
 }
 
 RtpSession::RtpSession(const Socket::Ptr &sock)
@@ -40,10 +47,6 @@ RtpSession::RtpSession(const Socket::Ptr &sock)
     socklen_t addr_len = sizeof(_addr);
     getpeername(sock->rawFD(), (struct sockaddr *)&_addr, &addr_len);
     _is_udp = sock->sockType() == SockNum::Sock_UDP;
-    if (_is_udp) {
-        // 设置udp socket读缓存
-        SockUtil::setRecvBuf(getSock()->rawFD(), 4 * 1024 * 1024);
-    }
 }
 
 RtpSession::~RtpSession() = default;

src/Rtp/RtpSession.h

@@ -25,6 +25,7 @@ public:
     static const std::string kStreamID;
     static const std::string kSSRC;
     static const std::string kOnlyAudio;
+    static const std::string kUdpRecvBuffer;
 
     RtpSession(const toolkit::Socket::Ptr &sock);
     ~RtpSession() override;