Merge pull request #485 from paullouisageneau/fix-transport-race-condition

Fix race condition between close and transport start

src/impl/dtlstransport.cpp

@@ -320,9 +320,9 @@ ssize_t DtlsTransport::ReadCallback(gnutls_transport_ptr_t ptr, void *data, size
 int DtlsTransport::TimeoutCallback(gnutls_transport_ptr_t ptr, unsigned int ms) {
 	DtlsTransport *t = static_cast<DtlsTransport *>(ptr);
 	try {
-		bool notEmpty = t->mIncomingQueue.wait(
+		bool isReadable = t->mIncomingQueue.wait(
 		    ms != GNUTLS_INDEFINITE_TIMEOUT ? std::make_optional(milliseconds(ms)) : nullopt);
-		return notEmpty ? 1 : 0;
+		return isReadable ? 1 : 0;
 
 	} catch (const std::exception &e) {
 		PLOG_WARNING << e.what();

src/impl/peerconnection.cpp

@@ -115,6 +115,19 @@ size_t PeerConnection::remoteMaxMessageSize() const {
 	return std::min(remoteMax, localMax);
 }
 
+// Helper for PeerConnection::initXTransport methods: start and emplace the transport
+template <typename T>
+shared_ptr<T> emplaceTransport(PeerConnection *pc, shared_ptr<T> *member, shared_ptr<T> transport) {
+	transport->start();
+	std::atomic_store(member, transport);
+	if (pc->state.load() == PeerConnection::State::Closed) {
+		std::atomic_store(member, decltype(transport)(nullptr));
+		transport->stop();
+		return nullptr;
+	}
+	return transport;
+}
+
 shared_ptr<IceTransport> PeerConnection::initIceTransport() {
 	try {
 		if (auto transport = std::atomic_load(&mIceTransport))
@@ -136,7 +149,7 @@ shared_ptr<IceTransport> PeerConnection::initIceTransport() {
 				    changeState(State::Failed);
 				    break;
 			    case IceTransport::State::Connected:
-				    initDtlsTransport();
+				    mProcessor->enqueue(&PeerConnection::initDtlsTransport, this);
 				    break;
 			    case IceTransport::State::Disconnected:
 				    changeState(State::Disconnected);
@@ -164,13 +177,7 @@ shared_ptr<IceTransport> PeerConnection::initIceTransport() {
 			    }
 		    });
 
-		std::atomic_store(&mIceTransport, transport);
-		if (state.load() == State::Closed) {
-			std::atomic_store(&mIceTransport, decltype(mIceTransport)(nullptr));
-			throw std::runtime_error("Connection is closed");
-		}
-		transport->start();
-		return transport;
+		return emplaceTransport(this, &mIceTransport, std::move(transport));
 
 	} catch (const std::exception &e) {
 		PLOG_ERROR << e.what();
@@ -201,7 +208,7 @@ shared_ptr<DtlsTransport> PeerConnection::initDtlsTransport() {
 			    switch (transportState) {
 			    case DtlsTransport::State::Connected:
 				    if (auto remote = remoteDescription(); remote && remote->hasApplication())
-					    initSctpTransport();
+					    mProcessor->enqueue(&PeerConnection::initSctpTransport, this);
 				    else
 					    changeState(State::Connected);
 
@@ -239,13 +246,7 @@ shared_ptr<DtlsTransport> PeerConnection::initDtlsTransport() {
 			                                            verifierCallback, dtlsStateChangeCallback);
 		}
 
-		std::atomic_store(&mDtlsTransport, transport);
-		if (state.load() == State::Closed) {
-			std::atomic_store(&mDtlsTransport, decltype(mDtlsTransport)(nullptr));
-			throw std::runtime_error("Connection is closed");
-		}
-		transport->start();
-		return transport;
+		return emplaceTransport(this, &mDtlsTransport, std::move(transport));
 
 	} catch (const std::exception &e) {
 		PLOG_ERROR << e.what();
@@ -301,13 +302,7 @@ shared_ptr<SctpTransport> PeerConnection::initSctpTransport() {
 			    }
 		    });
 
-		std::atomic_store(&mSctpTransport, transport);
-		if (state.load() == State::Closed) {
-			std::atomic_store(&mSctpTransport, decltype(mSctpTransport)(nullptr));
-			throw std::runtime_error("Connection is closed");
-		}
-		transport->start();
-		return transport;
+		return emplaceTransport(this, &mSctpTransport, std::move(transport));
 
 	} catch (const std::exception &e) {
 		PLOG_ERROR << e.what();
@@ -344,18 +339,19 @@ void PeerConnection::closeTransports() {
 		auto sctp = std::atomic_exchange(&mSctpTransport, decltype(mSctpTransport)(nullptr));
 		auto dtls = std::atomic_exchange(&mDtlsTransport, decltype(mDtlsTransport)(nullptr));
 		auto ice = std::atomic_exchange(&mIceTransport, decltype(mIceTransport)(nullptr));
-		ThreadPool::Instance().enqueue([sctp, dtls, ice]() mutable {
-			if (sctp)
-				sctp->stop();
-			if (dtls)
-				dtls->stop();
-			if (ice)
-				ice->stop();
-
-			sctp.reset();
-			dtls.reset();
-			ice.reset();
-		});
+		ThreadPool::Instance().enqueue(
+		    [sctp = std::move(sctp), dtls = std::move(dtls), ice = std::move(ice)]() mutable {
+			    if (sctp)
+				    sctp->stop();
+			    if (dtls)
+				    dtls->stop();
+			    if (ice)
+				    ice->stop();
+
+			    sctp.reset();
+			    dtls.reset();
+			    ice.reset();
+		    });
 	});
 }
 
@@ -937,6 +933,9 @@ void PeerConnection::processRemoteDescription(Description description) {
 	}
 
 	auto iceTransport = initIceTransport();
+	if (!iceTransport)
+		return; // closed
+
 	iceTransport->setRemoteDescription(std::move(description));
 
 	// Since we assumed passive role during DataChannel creation, we might need to shift the stream
@@ -1037,11 +1036,11 @@ void PeerConnection::triggerPendingTracks() {
 }
 
 void PeerConnection::flushPendingDataChannels() {
-	mProcessor->enqueue(std::bind(&PeerConnection::triggerPendingDataChannels, this));
+	mProcessor->enqueue(&PeerConnection::triggerPendingDataChannels, this);
 }
 
 void PeerConnection::flushPendingTracks() {
-	mProcessor->enqueue(std::bind(&PeerConnection::triggerPendingTracks, this));
+	mProcessor->enqueue(&PeerConnection::triggerPendingTracks, this);
 }
 
 bool PeerConnection::changeState(State newState) {

src/impl/queue.hpp

@@ -138,12 +138,13 @@ template <typename T> optional<T> Queue<T>::exchange(T element) {
 
 template <typename T> bool Queue<T>::wait(const optional<std::chrono::milliseconds> &duration) {
 	std::unique_lock lock(mMutex);
-	if (duration)
-		mPopCondition.wait_for(lock, *duration, [this]() { return !mQueue.empty() || mStopping; });
-	else
+	if (duration) {
+		return mPopCondition.wait_for(lock, *duration,
+		                              [this]() { return !mQueue.empty() || mStopping; });
+	} else {
 		mPopCondition.wait(lock, [this]() { return !mQueue.empty() || mStopping; });
-
-	return !mQueue.empty();
+		return true;
+	}
 }
 
 template <typename T> void Queue<T>::pushImpl(T element) {

src/impl/tlstransport.cpp

@@ -267,9 +267,9 @@ ssize_t TlsTransport::ReadCallback(gnutls_transport_ptr_t ptr, void *data, size_
 int TlsTransport::TimeoutCallback(gnutls_transport_ptr_t ptr, unsigned int ms) {
 	TlsTransport *t = static_cast<TlsTransport *>(ptr);
 	try {
-		bool notEmpty = t->mIncomingQueue.wait(
+		bool isReadable = t->mIncomingQueue.wait(
 		    ms != GNUTLS_INDEFINITE_TIMEOUT ? std::make_optional(milliseconds(ms)) : nullopt);
-		return notEmpty ? 1 : 0;
+		return isReadable ? 1 : 0;
 
 	} catch (const std::exception &e) {
 		PLOG_WARNING << e.what();

src/impl/websocket.cpp

@@ -175,6 +175,19 @@ void WebSocket::incoming(message_ptr message) {
 	}
 }
 
+// Helper for WebSocket::initXTransport methods: start and emplace the transport
+template <typename T>
+shared_ptr<T> emplaceTransport(WebSocket *ws, shared_ptr<T> *member, shared_ptr<T> transport) {
+	transport->start();
+	std::atomic_store(member, transport);
+	if (ws->state.load() == WebSocket::State::Closed) {
+		std::atomic_store(member, decltype(transport)(nullptr));
+		transport->stop();
+		return nullptr;
+	}
+	return transport;
+}
+
 shared_ptr<TcpTransport> WebSocket::setTcpTransport(shared_ptr<TcpTransport> transport) {
 	PLOG_VERBOSE << "Starting TCP transport";
 
@@ -210,13 +223,7 @@ shared_ptr<TcpTransport> WebSocket::setTcpTransport(shared_ptr<TcpTransport> tra
 			}
 		});
 
-		std::atomic_store(&mTcpTransport, transport);
-		if (state == WebSocket::State::Closed) {
-			std::atomic_store(&mTcpTransport, decltype(mTcpTransport)(nullptr));
-			throw std::runtime_error("Connection is closed");
-		}
-		transport->start();
-		return transport;
+		return emplaceTransport(this, &mTcpTransport, std::move(transport));
 
 	} catch (const std::exception &e) {
 		PLOG_ERROR << e.what();
@@ -273,13 +280,7 @@ shared_ptr<TlsTransport> WebSocket::initTlsTransport() {
 			transport =
 			    std::make_shared<TlsTransport>(lower, mHostname, mCertificate, stateChangeCallback);
 
-		std::atomic_store(&mTlsTransport, transport);
-		if (state == WebSocket::State::Closed) {
-			std::atomic_store(&mTlsTransport, decltype(mTlsTransport)(nullptr));
-			throw std::runtime_error("Connection is closed");
-		}
-		transport->start();
-		return transport;
+		return emplaceTransport(this, &mTlsTransport, std::move(transport));
 
 	} catch (const std::exception &e) {
 		PLOG_ERROR << e.what();
@@ -341,13 +342,7 @@ shared_ptr<WsTransport> WebSocket::initWsTransport() {
 		auto transport = std::make_shared<WsTransport>(
 		    lower, mWsHandshake, weak_bind(&WebSocket::incoming, this, _1), stateChangeCallback);
 
-		std::atomic_store(&mWsTransport, transport);
-		if (state == WebSocket::State::Closed) {
-			std::atomic_store(&mWsTransport, decltype(mWsTransport)(nullptr));
-			throw std::runtime_error("Connection is closed");
-		}
-		transport->start();
-		return transport;
+		return emplaceTransport(this, &mWsTransport, std::move(transport));
 
 	} catch (const std::exception &e) {
 		PLOG_ERROR << e.what();
@@ -387,18 +382,19 @@ void WebSocket::closeTransports() {
 	auto ws = std::atomic_exchange(&mWsTransport, decltype(mWsTransport)(nullptr));
 	auto tls = std::atomic_exchange(&mTlsTransport, decltype(mTlsTransport)(nullptr));
 	auto tcp = std::atomic_exchange(&mTcpTransport, decltype(mTcpTransport)(nullptr));
-	ThreadPool::Instance().enqueue([ws, tls, tcp]() mutable {
-		if (ws)
-			ws->stop();
-		if (tls)
-			tls->stop();
-		if (tcp)
-			tcp->stop();
-
-		ws.reset();
-		tls.reset();
-		tcp.reset();
-	});
+	ThreadPool::Instance().enqueue(
+	    [ws = std::move(ws), tls = std::move(tls), tcp = std::move(tcp)]() mutable {
+		    if (ws)
+			    ws->stop();
+		    if (tls)
+			    tls->stop();
+		    if (tcp)
+			    tcp->stop();
+
+		    ws.reset();
+		    tls.reset();
+		    tcp.reset();
+	    });
 }
 
 } // namespace rtc::impl

src/peerconnection.cpp

@@ -137,6 +137,8 @@ void PeerConnection::setLocalDescription(Description::Type type) {
 	}
 
 	auto iceTransport = impl()->initIceTransport();
+	if (!iceTransport)
+		return; // closed
 
 	Description local = iceTransport->getLocalDescription(type);
 	impl()->processLocalDescription(std::move(local));