修复webrtc多候选地址无法来回切换的bug (#2266)

最后一个连通的候选地址会被赋值并锁定为_selected_session，如果之前的候选地址再发送数据，将通过_selected_session回复，导致无法切换为旧的候选地址。

webrtc/IceServer.cpp

@@ -198,8 +198,12 @@ namespace RTC
 				// Create a success response.
 				RTC::StunPacket* response = packet->CreateSuccessResponse();
 
+				sockaddr_storage peerAddr;
+				socklen_t addr_len = sizeof(peerAddr);
+				getpeername(tuple->getSock()->rawFD(), (struct sockaddr *)&peerAddr, &addr_len);
+
 				// Add XOR-MAPPED-ADDRESS.
-				response->SetXorMappedAddress(tuple);
+				response->SetXorMappedAddress((struct sockaddr *)&peerAddr);
 
 				// Authenticate the response.
 				if (this->oldPassword.empty())
@@ -260,9 +264,9 @@ namespace RTC
 
 		for (; it != this->tuples.end(); ++it)
 		{
-			RTC::TransportTuple* storedTuple = std::addressof(*it);
+			RTC::TransportTuple* storedTuple = *it;
 
-			if (memcmp(storedTuple, tuple, sizeof (RTC::TransportTuple)) == 0)
+			if (storedTuple == tuple)
 			{
 				removedTuple = storedTuple;
 
@@ -285,9 +289,9 @@ namespace RTC
 		this->selectedTuple = nullptr;
 
 		// Mark the first tuple as selected tuple (if any).
-		if (this->tuples.begin() != this->tuples.end())
+		if (!this->tuples.empty())
 		{
-			SetSelectedTuple(std::addressof(*this->tuples.begin()));
+			SetSelectedTuple(this->tuples.front());
 		}
 		// Or just emit 'disconnected'.
 		else
@@ -477,12 +481,10 @@ namespace RTC
 		MS_TRACE();
 
 		// Add the new tuple at the beginning of the list.
-		this->tuples.push_front(*tuple);
+		this->tuples.push_front(tuple);
-
-		auto* storedTuple = std::addressof(*this->tuples.begin());
 
 		// Return the address of the inserted tuple.
-		return storedTuple;
+		return tuple;
 	}
 
 	inline RTC::TransportTuple* IceServer::HasTuple(const RTC::TransportTuple* tuple) const
@@ -495,15 +497,14 @@ namespace RTC
 			return nullptr;
 
 		// Check the current selected tuple.
-		if (memcmp(selectedTuple, tuple, sizeof (RTC::TransportTuple)) == 0)
+		if (selectedTuple == tuple)
 			return this->selectedTuple;
 
 		// Otherwise check other stored tuples.
 		for (const auto& it : this->tuples)
 		{
-			auto* storedTuple = const_cast<RTC::TransportTuple*>(std::addressof(it));
+			auto& storedTuple = it;
-
+			if (storedTuple == tuple)
-			if (memcmp(storedTuple, tuple, sizeof (RTC::TransportTuple)) == 0)
 				return storedTuple;
 		}
 
@@ -519,6 +520,7 @@ namespace RTC
 			return;
 
 		this->selectedTuple = storedTuple;
+        this->lastSelectedTuple = storedTuple->shared_from_this();
 
 		// Notify the listener.
 		this->listener->OnIceServerSelectedTuple(this, this->selectedTuple);

webrtc/IceServer.hpp

@@ -20,6 +20,7 @@ OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 #define MS_RTC_ICE_SERVER_HPP
 
 #include "StunPacket.hpp"
+#include "Network/Session.h"
 #include "logger.h"
 #include "Utils.hpp"
 #include <list>
@@ -27,11 +28,9 @@ OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 #include <functional>
 #include <memory>
 
-using _TransportTuple = struct sockaddr;
-
 namespace RTC
 {
-    using TransportTuple = _TransportTuple;
+	using TransportTuple = toolkit::Session;
 	class IceServer
 	{
 	public:
@@ -80,9 +79,9 @@ namespace RTC
 		{
 			return this->state;
 		}
-		RTC::TransportTuple* GetSelectedTuple() const
+		RTC::TransportTuple* GetSelectedTuple(bool try_last_tuple = false) const
 		{
-			return this->selectedTuple;
+            return try_last_tuple ? this->lastSelectedTuple.lock().get() : this->selectedTuple;
         }
 		void SetUsernameFragment(const std::string& usernameFragment)
 		{
@@ -100,6 +99,8 @@ namespace RTC
 		// and the given tuple must be an already valid tuple.
 		void ForceSelectedTuple(const RTC::TransportTuple* tuple);
 
+        const std::list<RTC::TransportTuple *>& GetTuples() const { return tuples; }
+
     private:
 		void HandleTuple(RTC::TransportTuple* tuple, bool hasUseCandidate);
 		/**
@@ -125,8 +126,9 @@ namespace RTC
 		std::string oldUsernameFragment;
 		std::string oldPassword;
 		IceState state{ IceState::NEW };
-		std::list<RTC::TransportTuple> tuples;
+		std::list<RTC::TransportTuple *> tuples;
-		RTC::TransportTuple* selectedTuple{ nullptr };
+        RTC::TransportTuple *selectedTuple;
+        std::weak_ptr<RTC::TransportTuple> lastSelectedTuple;
 		//最大不超过mtu
         static constexpr size_t StunSerializeBufferSize{ 1600 };
         uint8_t StunSerializeBuffer[StunSerializeBufferSize];

webrtc/WebRtcPlayer.cpp

@@ -70,21 +70,17 @@ void WebRtcPlayer::onStartWebRTC() {
     }
 }
 void WebRtcPlayer::onDestory() {
-    WebRtcTransportImp::onDestory();
-
     auto duration = getDuration();
     auto bytes_usage = getBytesUsage();
     //流量统计事件广播
     GET_CONFIG(uint32_t, iFlowThreshold, General::kFlowThreshold);
     if (_reader && getSession()) {
-        WarnL << "RTC播放器("
+        WarnL << "RTC播放器(" << _media_info.shortUrl() << ")结束播放,耗时(s):" << duration;
-              << _media_info.shortUrl()
-              << ")结束播放,耗时(s):" << duration;
         if (bytes_usage >= iFlowThreshold * 1024) {
-            NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastFlowReport, _media_info, bytes_usage, duration,
+            NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastFlowReport, _media_info, bytes_usage, duration, true, static_cast<SockInfo &>(*getSession()));
-                                               true, static_cast<SockInfo &>(*getSession()));
         }
     }
+    WebRtcTransportImp::onDestory();
 }
 
 void WebRtcPlayer::onRtcConfigure(RtcConfigure &configure) const {

webrtc/WebRtcPusher.cpp

@@ -118,20 +118,15 @@ void WebRtcPusher::onStartWebRTC() {
 }
 
 void WebRtcPusher::onDestory() {
-    WebRtcTransportImp::onDestory();
-
     auto duration = getDuration();
     auto bytes_usage = getBytesUsage();
     //流量统计事件广播
     GET_CONFIG(uint32_t, iFlowThreshold, General::kFlowThreshold);
 
     if (getSession()) {
-        WarnL << "RTC推流器("
+        WarnL << "RTC推流器(" << _media_info.shortUrl() << ")结束推流,耗时(s):" << duration;
-              << _media_info.shortUrl()
-              << ")结束推流,耗时(s):" << duration;
         if (bytes_usage >= iFlowThreshold * 1024) {
-            NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastFlowReport, _media_info, bytes_usage, duration,
+            NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastFlowReport, _media_info, bytes_usage, duration, false, static_cast<SockInfo &>(*getSession()));
-                                               false, static_cast<SockInfo &>(*getSession()));
         }
     }
 
@@ -142,6 +137,7 @@ void WebRtcPusher::onDestory() {
         auto push_src = std::move(_push_src);
         getPoller()->doDelayTask(_continue_push_ms, [push_src]() { return 0; });
     }
+    WebRtcTransportImp::onDestory();
 }
 
 void WebRtcPusher::onRtcConfigure(RtcConfigure &configure) const {

webrtc/WebRtcSession.cpp

@@ -48,8 +48,6 @@ EventPoller::Ptr WebRtcSession::queryPoller(const Buffer::Ptr &buffer) {
 ////////////////////////////////////////////////////////////////////////////////
 
 WebRtcSession::WebRtcSession(const Socket::Ptr &sock) : Session(sock) {
-    socklen_t addr_len = sizeof(_peer_addr);
-    getpeername(sock->rawFD(), (struct sockaddr *)&_peer_addr, &addr_len);
     _over_tcp = sock->sockType() == SockNum::Sock_TCP;
 }
 
@@ -87,14 +85,12 @@ void WebRtcSession::onRecv_l(const char *data, size_t len) {
             //3、销毁原先的socket和WebRtcSession(原先的对象跟WebRtcTransport不在同一条线程)
             throw std::runtime_error("webrtc over tcp change poller: " + getPoller()->getThreadName() + " -> " + sock->getPoller()->getThreadName());
         }
-
-        transport->setSession(shared_from_this());
         _transport = std::move(transport);
         InfoP(this);
     }
     _ticker.resetTime();
     CHECK(_transport);
-    _transport->inputSockData((char *)data, len, (struct sockaddr *)&_peer_addr);
+    _transport->inputSockData((char *)data, len, this);
 }
 
 void WebRtcSession::onRecv(const Buffer::Ptr &buffer) {
@@ -114,9 +110,13 @@ void WebRtcSession::onError(const SockException &err) {
     if (!_transport) {
         return;
     }
+    auto self = shared_from_this();
     auto transport = std::move(_transport);
-    getPoller()->async([transport] {
+    getPoller()->async([transport, self]() mutable {
         //延时减引用，防止使用transport对象时，销毁对象
+        transport->removeTuple(self.get());
+        //确保transport在Session对象前销毁，防止WebRtcTransport::onDestory()时获取不到Session对象
+        transport = nullptr;
     }, false);
 }
 

webrtc/WebRtcSession.h

@@ -46,7 +46,6 @@ private:
     bool _over_tcp = false;
     bool _find_transport = true;
     Ticker _ticker;
-    struct sockaddr_storage _peer_addr;
     std::weak_ptr<toolkit::TcpServer> _server;
     WebRtcTransportImp::Ptr _transport;
 };

webrtc/WebRtcTransport.cpp

@@ -75,6 +75,17 @@ static void translateIPFromEnv(std::vector<std::string> &v) {
     }
 }
 
+const char* sockTypeStr(Session* session) {
+    if (session) {
+        switch (session->getSock()->sockType()) {
+            case SockNum::Sock_TCP: return "tcp";
+            case SockNum::Sock_UDP: return "udp";
+            default: break;
+        }
+    }
+    return "unknown";
+}
+
 WebRtcTransport::WebRtcTransport(const EventPoller::Ptr &poller) {
     _poller = poller;
     _identifier = "zlm_" + to_string(++s_key);
@@ -109,16 +120,18 @@ void WebRtcTransport::OnIceServerSendStunPacket(
     sendSockData((char *)packet->GetData(), packet->GetSize(), tuple);
 }
 
-void WebRtcTransport::OnIceServerSelectedTuple(const RTC::IceServer *iceServer, RTC::TransportTuple *tuple) {
+void WebRtcTransportImp::OnIceServerSelectedTuple(const RTC::IceServer *iceServer, RTC::TransportTuple *tuple) {
-    InfoL;
+    InfoL << getIdentifier() << " select tuple " << sockTypeStr(tuple) << " " << tuple->get_peer_ip() << ":" << tuple->get_peer_port();
+    tuple->setSendFlushFlag(false);
+    unrefSelf();
 }
 
 void WebRtcTransport::OnIceServerConnected(const RTC::IceServer *iceServer) {
-    InfoL;
+    InfoL << getIdentifier();
 }
 
 void WebRtcTransport::OnIceServerCompleted(const RTC::IceServer *iceServer) {
-    InfoL;
+    InfoL << getIdentifier();
     if (_answer_sdp->media[0].role == DtlsRole::passive) {
         _dtls_transport->Run(RTC::DtlsTransport::Role::SERVER);
     } else {
@@ -127,7 +140,7 @@ void WebRtcTransport::OnIceServerCompleted(const RTC::IceServer *iceServer) {
 }
 
 void WebRtcTransport::OnIceServerDisconnected(const RTC::IceServer *iceServer) {
-    InfoL;
+    InfoL << getIdentifier();
 }
 
 //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -135,7 +148,7 @@ void WebRtcTransport::OnIceServerDisconnected(const RTC::IceServer *iceServer) {
 void WebRtcTransport::OnDtlsTransportConnected(
     const RTC::DtlsTransport *dtlsTransport, RTC::SrtpSession::CryptoSuite srtpCryptoSuite, uint8_t *srtpLocalKey,
     size_t srtpLocalKeyLen, uint8_t *srtpRemoteKey, size_t srtpRemoteKeyLen, std::string &remoteCert) {
-    InfoL;
+    InfoL << getIdentifier();
     _srtp_session_send = std::make_shared<RTC::SrtpSession>(
         RTC::SrtpSession::Type::OUTBOUND, srtpCryptoSuite, srtpLocalKey, srtpLocalKeyLen);
     _srtp_session_recv = std::make_shared<RTC::SrtpSession>(
@@ -153,16 +166,16 @@ void WebRtcTransport::OnDtlsTransportSendData(
 }
 
 void WebRtcTransport::OnDtlsTransportConnecting(const RTC::DtlsTransport *dtlsTransport) {
-    InfoL;
+    InfoL << getIdentifier();
 }
 
 void WebRtcTransport::OnDtlsTransportFailed(const RTC::DtlsTransport *dtlsTransport) {
-    InfoL;
+    InfoL << getIdentifier();
     onShutdown(SockException(Err_shutdown, "dtls transport failed"));
 }
 
 void WebRtcTransport::OnDtlsTransportClosed(const RTC::DtlsTransport *dtlsTransport) {
-    InfoL;
+    InfoL << getIdentifier();
     onShutdown(SockException(Err_shutdown, "dtls close notify received"));
 }
 
@@ -178,7 +191,7 @@ void WebRtcTransport::OnDtlsTransportApplicationDataReceived(
 //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 #ifdef ENABLE_SCTP
 void WebRtcTransport::OnSctpAssociationConnecting(RTC::SctpAssociation *sctpAssociation) {
-    TraceL;
+    TraceL << getIdentifier();
 }
 
 void WebRtcTransport::OnSctpAssociationConnected(RTC::SctpAssociation *sctpAssociation) {
@@ -215,8 +228,9 @@ void WebRtcTransport::sendSockData(const char *buf, size_t len, RTC::TransportTu
     onSendSockData(std::move(pkt), true, tuple ? tuple : _ice_server->GetSelectedTuple());
 }
 
-RTC::TransportTuple *WebRtcTransport::getSelectedTuple() const {
+Session::Ptr WebRtcTransport::getSession() const {
-    return _ice_server->GetSelectedTuple();
+    auto tuple = _ice_server->GetSelectedTuple(true);
+    return tuple ? tuple->shared_from_this() : nullptr;
 }
 
 void WebRtcTransport::sendRtcpRemb(uint32_t ssrc, size_t bit_rate) {
@@ -293,7 +307,7 @@ static bool isDtls(char *buf) {
 }
 
 static string getPeerAddress(RTC::TransportTuple *tuple) {
-    return SockUtil::inet_ntoa(tuple);
+    return tuple->get_peer_ip();
 }
 
 void WebRtcTransport::inputSockData(char *buf, int len, RTC::TransportTuple *tuple) {
@@ -409,24 +423,27 @@ void WebRtcTransportImp::onDestory() {
 }
 
 void WebRtcTransportImp::onSendSockData(Buffer::Ptr buf, bool flush, RTC::TransportTuple *tuple) {
-    if (!_selected_session) {
+    if (tuple == nullptr) {
+        tuple = _ice_server->GetSelectedTuple();
+        if (!tuple) {
             WarnL << "send data failed:" << buf->size();
             return;
         }
+    }
 
     // 一次性发送一帧的rtp数据，提高网络io性能
-    if (_selected_session->getSock()->sockType() == SockNum::Sock_TCP) {
+    if (tuple->getSock()->sockType() == SockNum::Sock_TCP) {
         // 增加tcp两字节头
         auto len = buf->size();
         char tcp_len[2] = { 0 };
         tcp_len[0] = (len >> 8) & 0xff;
         tcp_len[1] = len & 0xff;
-        _selected_session->SockSender::send(tcp_len, 2);
+        tuple->SockSender::send(tcp_len, 2);
     }
-    _selected_session->send(std::move(buf));
+    tuple->send(std::move(buf));
 
     if (flush) {
-        _selected_session->flushAll();
+        tuple->flushAll();
     }
 }
 
@@ -1040,28 +1057,14 @@ void WebRtcTransportImp::onBeforeEncryptRtp(const char *buf, int &len, void *ctx
 void WebRtcTransportImp::onShutdown(const SockException &ex) {
     WarnL << ex.what();
     unrefSelf();
-    for (auto &pr : _history_sessions) {
+    for (auto &tuple : _ice_server->GetTuples()) {
-        auto session = pr.second.lock();
+        tuple->shutdown(ex);
-        if (session) {
-            session->shutdown(ex);
-        }
     }
 }
 
-void WebRtcTransportImp::setSession(Session::Ptr session) {
+void WebRtcTransportImp::removeTuple(RTC::TransportTuple *tuple) {
-    _history_sessions.emplace(session.get(), session);
+    InfoL << getIdentifier() << " remove tuple " << tuple->get_peer_ip() << ":" << tuple->get_peer_port();
-    if (_selected_session) {
+    this->_ice_server->RemoveTuple(tuple);
-        InfoL << "rtc network changed: " << _selected_session->get_peer_ip() << ":"
-              << _selected_session->get_peer_port() << " -> " << session->get_peer_ip() << ":"
-              << session->get_peer_port() << ", id:" << getIdentifier();
-    }
-    _selected_session = std::move(session);
-    _selected_session->setSendFlushFlag(false);
-    unrefSelf();
-}
-
-const Session::Ptr &WebRtcTransportImp::getSession() const {
-    return _selected_session;
 }
 
 uint64_t WebRtcTransportImp::getBytesUsage() const {

webrtc/WebRtcTransport.h

@@ -110,6 +110,7 @@ public:
     void sendRtcpPacket(const char *buf, int len, bool flush, void *ctx = nullptr);
 
     const EventPoller::Ptr& getPoller() const;
+    Session::Ptr getSession() const;
 
 protected:
     ////  dtls相关的回调 ////
@@ -130,7 +131,6 @@ protected:
 protected:
     //// ice相关的回调 ///
     void OnIceServerSendStunPacket(const RTC::IceServer *iceServer, const RTC::StunPacket *packet, RTC::TransportTuple *tuple) override;
-    void OnIceServerSelectedTuple(const RTC::IceServer *iceServer, RTC::TransportTuple *tuple) override;
     void OnIceServerConnected(const RTC::IceServer *iceServer) override;
     void OnIceServerCompleted(const RTC::IceServer *iceServer) override;
     void OnIceServerDisconnected(const RTC::IceServer *iceServer) override;
@@ -159,7 +159,6 @@ protected:
     virtual void onRtcpBye() = 0;
 
 protected:
-    RTC::TransportTuple* getSelectedTuple() const;
     void sendRtcpRemb(uint32_t ssrc, size_t bit_rate);
     void sendRtcpPli(uint32_t ssrc);
 
@@ -170,11 +169,11 @@ private:
 protected:
     RtcSession::Ptr _offer_sdp;
     RtcSession::Ptr _answer_sdp;
+    std::shared_ptr<RTC::IceServer> _ice_server;
 
 private:
     std::string _identifier;
     EventPoller::Ptr _poller;
-    std::shared_ptr<RTC::IceServer> _ice_server;
     std::shared_ptr<RTC::DtlsTransport> _dtls_transport;
     std::shared_ptr<RTC::SrtpSession> _srtp_session_send;
     std::shared_ptr<RTC::SrtpSession> _srtp_session_recv;
@@ -239,8 +238,6 @@ public:
     using Ptr = std::shared_ptr<WebRtcTransportImp>;
     ~WebRtcTransportImp() override;
 
-    void setSession(Session::Ptr session);
-    const Session::Ptr& getSession() const;
     uint64_t getBytesUsage() const;
     uint64_t getDuration() const;
     bool canSendRtp() const;
@@ -248,8 +245,10 @@ public:
     void onSendRtp(const RtpPacket::Ptr &rtp, bool flush, bool rtx = false);
 
     void createRtpChannel(const std::string &rid, uint32_t ssrc, MediaTrack &track);
+    void removeTuple(RTC::TransportTuple* tuple);
 
 protected:
+    void OnIceServerSelectedTuple(const RTC::IceServer *iceServer, RTC::TransportTuple *tuple) override;
     WebRtcTransportImp(const EventPoller::Ptr &poller,bool preferred_tcp = false);
     void OnDtlsTransportApplicationDataReceived(const RTC::DtlsTransport *dtlsTransport, const uint8_t *data, size_t len) override;
     void onStartWebRTC() override;
@@ -292,10 +291,6 @@ private:
     Ticker _alive_ticker;
     //pli rtcp计时器
     Ticker _pli_ticker;
-    //当前选中的udp链接
-    Session::Ptr _selected_session;
-    //链接迁移前后使用过的udp链接
-    std::unordered_map<Session *, std::weak_ptr<Session> > _history_sessions;
     //twcc rtcp发送上下文对象
     TwccContext _twcc_ctx;
     //根据发送rtp的track类型获取相关信息