Fixed transport synchronization on destruction

include/rtc/datachannel.hpp

@@ -71,7 +71,7 @@ private:
 	void incoming(message_ptr message);
 	void processOpenMessage(message_ptr message);
 
-	const std::shared_ptr<PeerConnection> mPeerConnection;
+	std::shared_ptr<PeerConnection> mPeerConnection;
 	std::shared_ptr<SctpTransport> mSctpTransport;
 
 	unsigned int mStream;

include/rtc/peerconnection.hpp

@@ -87,17 +87,16 @@ private:
 	void initDtlsTransport();
 	void initSctpTransport();
 
-	bool checkFingerprint(std::weak_ptr<PeerConnection> weak_this, const std::string &fingerprint) const;
-	void forwardMessage(std::weak_ptr<PeerConnection> weak_this, message_ptr message);
-	void forwardBufferedAmount(std::weak_ptr<PeerConnection> weak_this, uint16_t stream,
-	                           size_t amount);
+	bool checkFingerprint(const std::string &fingerprint) const;
+	void forwardMessage(message_ptr message);
+	void forwardBufferedAmount(uint16_t stream, size_t amount);
 	void iterateDataChannels(std::function<void(std::shared_ptr<DataChannel> channel)> func);
 	void openDataChannels();
 	void closeDataChannels();
 
 	void processLocalDescription(Description description);
-	void processLocalCandidate(std::weak_ptr<PeerConnection> weak_this, Candidate candidate);
-	void triggerDataChannel(std::weak_ptr<PeerConnection> weak_this, std::weak_ptr<DataChannel> weakDataChannel);
+	void processLocalCandidate(Candidate candidate);
+	void triggerDataChannel(std::weak_ptr<DataChannel> weakDataChannel);
 	void changeState(State state);
 	void changeGatheringState(GatheringState state);
 

src/datachannel.cpp

@@ -81,6 +81,10 @@ void DataChannel::close() {
 		if (mSctpTransport)
 			mSctpTransport->reset(mStream);
 	}
+
+	// Reset mSctpTransport first so SctpTransport is never alive without PeerConnection
+	mSctpTransport.reset();
+	mPeerConnection.reset();
 }
 
 bool DataChannel::send(const std::variant<binary, string> &data) {

src/dtlstransport.cpp

@@ -85,10 +85,10 @@ DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certific
 }
 
 DtlsTransport::~DtlsTransport() {
-	mIncomingQueue.stop();
+	resetLower();
 
-	if (mRecvThread.joinable())
-		mRecvThread.join();
+	mIncomingQueue.stop();
+	mRecvThread.join();
 
 	gnutls_bye(mSession, GNUTLS_SHUT_RDWR);
 	gnutls_deinit(mSession);
@@ -356,10 +356,10 @@ DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certific
 }
 
 DtlsTransport::~DtlsTransport() {
-	mIncomingQueue.stop();
+	resetLower();
 
-	if (mRecvThread.joinable())
-		mRecvThread.join();
+	mIncomingQueue.stop();
+	mRecvThread.join();
 
 	SSL_shutdown(mSsl);
 	SSL_free(mSsl);

src/icetransport.cpp

@@ -132,8 +132,7 @@ IceTransport::IceTransport(const Configuration &config, Description::Role role,
 
 IceTransport::~IceTransport() {
 	g_main_loop_quit(mMainLoop.get());
-	if (mMainLoopThread.joinable())
-		mMainLoopThread.join();
+	mMainLoopThread.join();
 }
 
 Description::Role IceTransport::role() const { return mRole; }

src/peerconnection.cpp

@@ -36,7 +36,11 @@ PeerConnection::PeerConnection() : PeerConnection(Configuration()) {}
 PeerConnection::PeerConnection(const Configuration &config)
     : mConfig(config), mCertificate(make_certificate("libdatachannel")), mState(State::New) {}
 
-PeerConnection::~PeerConnection() {}
+PeerConnection::~PeerConnection() {
+	mSctpTransport.reset();
+	mDtlsTransport.reset();
+	mIceTransport.reset();
+}
 
 const Configuration *PeerConnection::config() const { return &mConfig; }
 
@@ -168,11 +172,8 @@ void PeerConnection::onGatheringStateChange(std::function<void(GatheringState st
 
 void PeerConnection::initIceTransport(Description::Role role) {
 	mIceTransport = std::make_shared<IceTransport>(
-	    mConfig, role, std::bind(&PeerConnection::processLocalCandidate, this, weak_ptr<PeerConnection>{shared_from_this()}, _1),
-	    [this, weak_this = weak_ptr<PeerConnection>{shared_from_this()}](IceTransport::State state) {
-        auto strong_this = weak_this.lock();
-        if (!strong_this) return;
-
+	    mConfig, role, std::bind(&PeerConnection::processLocalCandidate, this, _1),
+	    [this](IceTransport::State state) {
 		    switch (state) {
 		    case IceTransport::State::Connecting:
 			    changeState(State::Connecting);
@@ -188,10 +189,7 @@ void PeerConnection::initIceTransport(Description::Role role) {
 			    break;
 		    }
 	    },
-	    [this, weak_this = weak_ptr<PeerConnection>{shared_from_this()}](IceTransport::GatheringState state) {
-        auto strong_this = weak_this.lock();
-        if (!strong_this) return;
-
+	    [this](IceTransport::GatheringState state) {
 		    switch (state) {
 		    case IceTransport::GatheringState::InProgress:
 			    changeGatheringState(GatheringState::InProgress);
@@ -210,11 +208,8 @@ void PeerConnection::initIceTransport(Description::Role role) {
 
 void PeerConnection::initDtlsTransport() {
 	mDtlsTransport = std::make_shared<DtlsTransport>(
-	    mIceTransport, mCertificate, std::bind(&PeerConnection::checkFingerprint, this, weak_ptr<PeerConnection>{shared_from_this()}, _1),
-	    [this, weak_this = weak_ptr<PeerConnection>{shared_from_this()}](DtlsTransport::State state) {
-        auto strong_this = weak_this.lock();
-        if (!strong_this) return;
-
+	    mIceTransport, mCertificate, std::bind(&PeerConnection::checkFingerprint, this, _1),
+	    [this](DtlsTransport::State state) {
 		    switch (state) {
 		    case DtlsTransport::State::Connected:
 			    initSctpTransport();
@@ -232,17 +227,9 @@ void PeerConnection::initDtlsTransport() {
 void PeerConnection::initSctpTransport() {
 	uint16_t sctpPort = mRemoteDescription->sctpPort().value_or(DEFAULT_SCTP_PORT);
 	mSctpTransport = std::make_shared<SctpTransport>(
-	    mDtlsTransport, sctpPort,
-	    std::bind(&PeerConnection::forwardMessage, this,
-	              weak_ptr<PeerConnection>{shared_from_this()}, _1),
-	    std::bind(&PeerConnection::forwardBufferedAmount, this,
-	              weak_ptr<PeerConnection>{shared_from_this()}, _1, _2),
-	    [this,
-	     weak_this = weak_ptr<PeerConnection>{shared_from_this()}](SctpTransport::State state) {
-		    auto strong_this = weak_this.lock();
-		    if (!strong_this)
-			    return;
-
+	    mDtlsTransport, sctpPort, std::bind(&PeerConnection::forwardMessage, this, _1),
+	    std::bind(&PeerConnection::forwardBufferedAmount, this, _1, _2),
+	    [this](SctpTransport::State state) {
 		    switch (state) {
 		    case SctpTransport::State::Connected:
 			    changeState(State::Connected);
@@ -261,10 +248,7 @@ void PeerConnection::initSctpTransport() {
 	    });
 }
 
-bool PeerConnection::checkFingerprint(weak_ptr<PeerConnection> weak_this, const std::string &fingerprint) const {
-  auto strong_this = weak_this.lock();
-  if (!strong_this) return false;
-
+bool PeerConnection::checkFingerprint(const std::string &fingerprint) const {
 	if (auto expectedFingerprint =
 	        mRemoteDescription ? mRemoteDescription->fingerprint() : nullopt) {
 		return *expectedFingerprint == fingerprint;
@@ -272,10 +256,7 @@ bool PeerConnection::checkFingerprint(weak_ptr<PeerConnection> weak_this, const
 	return false;
 }
 
-void PeerConnection::forwardMessage(weak_ptr<PeerConnection> weak_this, message_ptr message) {
-  auto strong_this = weak_this.lock();
-  if (!strong_this) return;
-
+void PeerConnection::forwardMessage(message_ptr message) {
 	if (!mIceTransport || !mSctpTransport)
 		throw std::logic_error("Got a DataChannel message without transport");
 
@@ -300,7 +281,8 @@ void PeerConnection::forwardMessage(weak_ptr<PeerConnection> weak_this, message_
 		    message->stream % 2 == remoteParity) {
 			channel =
 			    std::make_shared<DataChannel>(shared_from_this(), mSctpTransport, message->stream);
-			channel->onOpen(std::bind(&PeerConnection::triggerDataChannel, this, weak_this, weak_ptr<DataChannel>{channel}));
+			channel->onOpen(std::bind(&PeerConnection::triggerDataChannel, this,
+			                          weak_ptr<DataChannel>{channel}));
 			mDataChannels.insert(std::make_pair(message->stream, channel));
 		} else {
 			// Invalid, close the DataChannel by resetting the stream
@@ -312,12 +294,7 @@ void PeerConnection::forwardMessage(weak_ptr<PeerConnection> weak_this, message_
 	channel->incoming(message);
 }
 
-void PeerConnection::forwardBufferedAmount(weak_ptr<PeerConnection> weak_this, uint16_t stream,
-                                           size_t amount) {
-	auto strong_this = weak_this.lock();
-	if (!strong_this)
-		return;
-
+void PeerConnection::forwardBufferedAmount(uint16_t stream, size_t amount) {
 	shared_ptr<DataChannel> channel;
 	if (auto it = mDataChannels.find(stream); it != mDataChannels.end()) {
 		channel = it->second.lock();
@@ -364,11 +341,7 @@ void PeerConnection::processLocalDescription(Description description) {
 	mLocalDescriptionCallback(*mLocalDescription);
 }
 
-void PeerConnection::processLocalCandidate(weak_ptr<PeerConnection> weak_this, Candidate candidate) {
-	auto strong_this = weak_this.lock();
-	if (!strong_this)
-		return;
-
+void PeerConnection::processLocalCandidate(Candidate candidate) {
 	if (!mLocalDescription)
 		throw std::logic_error("Got a local candidate without local description");
 
@@ -377,11 +350,7 @@ void PeerConnection::processLocalCandidate(weak_ptr<PeerConnection> weak_this, C
 	mLocalCandidateCallback(candidate);
 }
 
-void PeerConnection::triggerDataChannel(weak_ptr<PeerConnection> weak_this, weak_ptr<DataChannel> weakDataChannel) {
-	auto strong_this = weak_this.lock();
-	if (!strong_this)
-		return;
-
+void PeerConnection::triggerDataChannel(weak_ptr<DataChannel> weakDataChannel) {
 	auto dataChannel = weakDataChannel.lock();
 	if (!dataChannel)
 		return;

src/sctptransport.cpp

@@ -143,6 +143,7 @@ SctpTransport::SctpTransport(std::shared_ptr<Transport> lower, uint16_t port,
 }
 
 SctpTransport::~SctpTransport() {
+	resetLower();
 	onRecv(nullptr); // unset recv callback
 
 	mSendQueue.stop();

src/transport.hpp

@@ -22,6 +22,7 @@
 #include "include.hpp"
 #include "message.hpp"
 
+#include <atomic>
 #include <functional>
 #include <memory>
 
@@ -32,31 +33,28 @@ using namespace std::placeholders;
 class Transport {
 public:
 	Transport(std::shared_ptr<Transport> lower = nullptr) : mLower(std::move(lower)) {
-		if (mLower)
-			mLower->onRecv(std::bind(&Transport::incoming, this, _1));
-	}
-	virtual ~Transport() {
-		if (mLower)
-			mLower->onRecv(nullptr);
+		if (auto lower = std::atomic_load(&mLower))
+			lower->onRecv(std::bind(&Transport::incoming, this, _1));
 	}
+	virtual ~Transport() { resetLower(); }
 
 	virtual bool send(message_ptr message) = 0;
 	void onRecv(message_callback callback) { mRecvCallback = std::move(callback); }
 
 protected:
 	void recv(message_ptr message) { mRecvCallback(message); }
+	void resetLower() {
+		if (auto lower = std::atomic_exchange(&mLower, std::shared_ptr<Transport>(nullptr)))
+			lower->onRecv(nullptr);
+	}
 
 	virtual void incoming(message_ptr message) = 0;
-	virtual void outgoing(message_ptr message) { getLower()->send(message); }
-
-private:
-	std::shared_ptr<Transport> getLower() {
-		if (mLower)
-			return mLower;
-		else
-			throw std::logic_error("No lower transport to call");
+	virtual void outgoing(message_ptr message) {
+		if (auto lower = std::atomic_load(&mLower))
+			lower->send(message);
 	}
 
+private:
 	std::shared_ptr<Transport> mLower;
 	synchronized_callback<message_ptr> mRecvCallback;
 };

test/main.cpp

@@ -100,5 +100,16 @@ int main(int argc, char **argv) {
 	});
 
 	this_thread::sleep_for(3s);
+
+	if (dc1->isOpen() && dc2->isOpen()) {
+		dc1->close();
+		dc2->close();
+
+		cout << "Success" << endl;
+		return 0;
+	} else {
+		cout << "Failure" << endl;
+		return 1;
+	}
 }