Bumped version to 0.7.2

Extend mitigation for usrsctp send after unregistering

CMakeLists.txt

@@ -1,7 +1,7 @@
 cmake_minimum_required(VERSION 3.7)
 project(libdatachannel
 	DESCRIPTION "WebRTC Data Channels Library"
-	VERSION 0.7.0
+	VERSION 0.7.2
 	LANGUAGES CXX)
 
 # Options

README.md

@@ -197,5 +197,6 @@ ws->open("wss://my.websocket/service");
 ```
 
 ## External resources
-- Rust wrappers for libdatachannel: [lerouxrgd/datachannel-rs](https://github.com/lerouxrgd/datachannel-rs)
+- Rust wrappers for libdatachannel: [datachannel-rs](https://github.com/lerouxrgd/datachannel-rs)
+- WebAssembly wrappers compatible with libdatachannel: [datachannel-wasm](https://github.com/paullouisageneau/datachannel-wasm)
 

include/rtc/include.hpp

@@ -84,9 +84,24 @@ template <typename F, typename T, typename... Args> auto weak_bind(F &&f, T *t,
 template <typename... P> class synchronized_callback {
 public:
 	synchronized_callback() = default;
+	synchronized_callback(synchronized_callback &&cb) { *this = std::move(cb); }
+	synchronized_callback(const synchronized_callback &cb) { *this = cb; }
 	synchronized_callback(std::function<void(P...)> func) { *this = std::move(func); }
 	~synchronized_callback() { *this = nullptr; }
 
+	synchronized_callback &operator=(synchronized_callback &&cb) {
+		std::scoped_lock lock(mutex, cb.mutex);
+		callback = std::move(cb.callback);
+		cb.callback = nullptr;
+		return *this;
+	}
+
+	synchronized_callback &operator=(const synchronized_callback &cb) {
+		std::scoped_lock lock(mutex, cb.mutex);
+		callback = cb.callback;
+		return *this;
+	}
+
 	synchronized_callback &operator=(std::function<void(P...)> func) {
 		std::lock_guard lock(mutex);
 		callback = std::move(func);

include/rtc/message.hpp

@@ -37,6 +37,8 @@ struct Message : binary {
 	Message(Iterator begin_, Iterator end_, Type type_ = Binary)
 	    : binary(begin_, end_), type(type_) {}
 
+	Message(binary &&data, Type type_ = Binary) : binary(std::move(data)), type(type_) {}
+
 	Type type;
 	unsigned int stream = 0;
 	std::shared_ptr<Reliability> reliability;
@@ -68,6 +70,15 @@ inline message_ptr make_message(size_t size, Message::Type type = Message::Binar
 	return message;
 }
 
+inline message_ptr make_message(binary &&data, Message::Type type = Message::Binary,
+                                unsigned int stream = 0,
+                                std::shared_ptr<Reliability> reliability = nullptr) {
+	auto message = std::make_shared<Message>(std::move(data), type);
+	message->stream = stream;
+	message->reliability = reliability;
+	return message;
+}
+
 } // namespace rtc
 
 #endif

src/peerconnection.cpp

@@ -51,13 +51,17 @@ PeerConnection::PeerConnection(const Configuration &config)
 }
 
 PeerConnection::~PeerConnection() {
-	close();
 	PLOG_VERBOSE << "Destroying PeerConnection";
+	close();
+	mProcessor->join();
 }
 
 void PeerConnection::close() {
 	PLOG_VERBOSE << "Closing PeerConnection";
-	closeDataChannels();
+
+	// Close data channels asynchronously
+	mProcessor->enqueue(std::bind(&PeerConnection::closeDataChannels, this));
+
 	closeTransports();
 }
 
@@ -78,6 +82,11 @@ std::optional<Description> PeerConnection::remoteDescription() const {
 }
 
 void PeerConnection::setLocalDescription(std::optional<Description> description) {
+	if (description)
+		PLOG_VERBOSE << "Setting local description: " << string(*description);
+	else
+		PLOG_VERBOSE << "Setting default local description";
+
 	if (auto iceTransport = std::atomic_load(&mIceTransport)) {
 		throw std::logic_error("Local description is already set");
 	} else {
@@ -94,6 +103,8 @@ void PeerConnection::setLocalDescription(std::optional<Description> description)
 }
 
 void PeerConnection::setRemoteDescription(Description description) {
+	PLOG_VERBOSE << "Setting remote description: " << string(description);
+
 	description.hintType(localDescription() ? Description::Type::Answer : Description::Type::Offer);
 	auto type = description.type();
 	auto remoteCandidates = description.extractCandidates(); // Candidates will be added at the end
@@ -139,6 +150,7 @@ void PeerConnection::setRemoteDescription(Description description) {
 }
 
 void PeerConnection::addRemoteCandidate(Candidate candidate) {
+	PLOG_VERBOSE << "Adding remote candidate: " << string(candidate);
 
 	auto iceTransport = std::atomic_load(&mIceTransport);
 	if (!mRemoteDescription || !iceTransport)
@@ -439,27 +451,30 @@ shared_ptr<SctpTransport> PeerConnection::initSctpTransport() {
 void PeerConnection::closeTransports() {
 	PLOG_VERBOSE << "Closing transports";
 
-	// Change state to sink state Closed to block init methods
+	// Change state to sink state Closed
 	changeState(State::Closed);
 
 	// Reset callbacks now that state is changed
 	resetCallbacks();
 
-	// Pass the references to a thread, allowing to terminate a transport from its own thread
-	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();
+	// Initiate transport stop on the processor after closing the data channels
+	mProcessor->enqueue([this]() {
+		// Pass the pointers to a thread
+		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();
+			sctp.reset();
+			dtls.reset();
+			ice.reset();
+		});
 	});
 }
 
@@ -648,7 +663,12 @@ bool PeerConnection::changeState(State state) {
 
 	} while (!mState.compare_exchange_weak(current, state));
 
-	mProcessor->enqueue([this, state]() { mStateChangeCallback(state); });
+	if (state == State::Closed)
+		// This is the last state change, so we may steal the callback
+		mProcessor->enqueue([this, cb = std::move(mStateChangeCallback)]() { cb(State::Closed); });
+	else
+		mProcessor->enqueue([this, state]() { mStateChangeCallback(state); });
+
 	return true;
 }
 

src/sctptransport.cpp

@@ -451,7 +451,7 @@ void SctpTransport::sendReset(uint16_t streamId) {
 		mWrittenCondition.wait_for(lock, 1000ms,
 		                           [&]() { return mWritten || state() != State::Connected; });
 	} else if (errno == EINVAL) {
-		PLOG_VERBOSE << "SCTP stream " << streamId << " already reset";
+		PLOG_DEBUG << "SCTP stream " << streamId << " already reset";
 	} else {
 		PLOG_WARNING << "SCTP reset stream " << streamId << " failed, errno=" << errno;
 	}
@@ -479,32 +479,22 @@ int SctpTransport::handleRecv(struct socket * /*sock*/, union sctp_sockstore /*a
 		// therefore partial notifications and messages need to be handled separately.
 		if (flags & MSG_NOTIFICATION) {
 			// SCTP event notification
+			mPartialNotification.insert(mPartialNotification.end(), data, data + len);
 			if (flags & MSG_EOR) {
-				if (!mPartialNotification.empty()) {
-					mPartialNotification.insert(mPartialNotification.end(), data, data + len);
-					data = mPartialNotification.data();
-					len = mPartialNotification.size();
-				}
 				// Notification is complete, process it
-				processNotification(reinterpret_cast<const union sctp_notification *>(data), len);
+				processNotification(
+				    reinterpret_cast<const union sctp_notification *>(mPartialNotification.data()),
+				    mPartialNotification.size());
 				mPartialNotification.clear();
-			} else {
-				mPartialNotification.insert(mPartialNotification.end(), data, data + len);
 			}
-
 		} else {
 			// SCTP message
+			mPartialMessage.insert(mPartialMessage.end(), data, data + len);
 			if (flags & MSG_EOR) {
-				if (!mPartialMessage.empty()) {
-					mPartialMessage.insert(mPartialMessage.end(), data, data + len);
-					data = mPartialMessage.data();
-					len = mPartialMessage.size();
-				}
 				// Message is complete, process it
-				processData(data, len, info.rcv_sid, PayloadId(htonl(info.rcv_ppid)));
+				processData(std::move(mPartialMessage), info.rcv_sid,
+				            PayloadId(htonl(info.rcv_ppid)));
 				mPartialMessage.clear();
-			} else {
-				mPartialMessage.insert(mPartialMessage.end(), data, data + len);
 			}
 		}
 
@@ -539,62 +529,56 @@ int SctpTransport::handleWrite(byte *data, size_t len, uint8_t /*tos*/, uint8_t
 	return 0; // success
 }
 
-void SctpTransport::processData(const byte *data, size_t len, uint16_t sid, PayloadId ppid) {
-	PLOG_VERBOSE << "Process data, len=" << len;
+void SctpTransport::processData(binary &&data, uint16_t sid, PayloadId ppid) {
+	PLOG_VERBOSE << "Process data, size=" << data.size();
 
 	// The usage of the PPIDs "WebRTC String Partial" and "WebRTC Binary Partial" is deprecated.
 	// See https://tools.ietf.org/html/draft-ietf-rtcweb-data-channel-13#section-6.6
 	// We handle them at reception for compatibility reasons but should never send them.
 	switch (ppid) {
 	case PPID_CONTROL:
-		recv(make_message(data, data + len, Message::Control, sid));
+		recv(make_message(std::move(data), Message::Control, sid));
 		break;
 
 	case PPID_STRING_PARTIAL: // deprecated
-		mPartialStringData.insert(mPartialStringData.end(), data, data + len);
+		mPartialStringData.insert(mPartialStringData.end(), data.begin(), data.end());
 		break;
 
 	case PPID_STRING:
 		if (mPartialStringData.empty()) {
-			mBytesReceived += len;
-			recv(make_message(data, data + len, Message::String, sid));
+			mBytesReceived += data.size();
+			recv(make_message(std::move(data), Message::String, sid));
 		} else {
-			mPartialStringData.insert(mPartialStringData.end(), data, data + len);
+			mPartialStringData.insert(mPartialStringData.end(), data.begin(), data.end());
 			mBytesReceived += mPartialStringData.size();
-			recv(make_message(mPartialStringData.begin(), mPartialStringData.end(), Message::String,
-			                  sid));
+			recv(make_message(std::move(mPartialStringData), Message::String, sid));
 			mPartialStringData.clear();
 		}
 		break;
 
 	case PPID_STRING_EMPTY:
-		// This only accounts for when the partial data is empty
-		recv(make_message(mPartialStringData.begin(), mPartialStringData.end(), Message::String,
-		                  sid));
+		recv(make_message(std::move(mPartialStringData), Message::String, sid));
 		mPartialStringData.clear();
 		break;
 
 	case PPID_BINARY_PARTIAL: // deprecated
-		mPartialBinaryData.insert(mPartialBinaryData.end(), data, data + len);
+		mPartialBinaryData.insert(mPartialBinaryData.end(), data.begin(), data.end());
 		break;
 
 	case PPID_BINARY:
 		if (mPartialBinaryData.empty()) {
-			mBytesReceived += len;
-			recv(make_message(data, data + len, Message::Binary, sid));
+			mBytesReceived += data.size();
+			recv(make_message(std::move(data), Message::Binary, sid));
 		} else {
-			mPartialBinaryData.insert(mPartialBinaryData.end(), data, data + len);
-			mBytesReceived += mPartialStringData.size();
-			recv(make_message(mPartialBinaryData.begin(), mPartialBinaryData.end(), Message::Binary,
-			                  sid));
+			mPartialBinaryData.insert(mPartialBinaryData.end(), data.begin(), data.end());
+			mBytesReceived += mPartialBinaryData.size();
+			recv(make_message(std::move(mPartialBinaryData), Message::Binary, sid));
 			mPartialBinaryData.clear();
 		}
 		break;
 
 	case PPID_BINARY_EMPTY:
-		// This only accounts for when the partial data is empty
-		recv(make_message(mPartialBinaryData.begin(), mPartialBinaryData.end(), Message::Binary,
-		                  sid));
+		recv(make_message(std::move(mPartialBinaryData), Message::Binary, sid));
 		mPartialBinaryData.clear();
 		break;
 
@@ -692,8 +676,16 @@ std::optional<milliseconds> SctpTransport::rtt() {
 
 int SctpTransport::RecvCallback(struct socket *sock, union sctp_sockstore addr, void *data,
                                 size_t len, struct sctp_rcvinfo recv_info, int flags, void *ptr) {
-	int ret = static_cast<SctpTransport *>(ptr)->handleRecv(
-	    sock, addr, static_cast<const byte *>(data), len, recv_info, flags);
+	auto *transport = static_cast<SctpTransport *>(ptr);
+
+	std::shared_lock lock(InstancesMutex);
+	if (Instances.find(transport) == Instances.end()) {
+		free(data);
+		return -1;
+	}
+
+	int ret =
+	    transport->handleRecv(sock, addr, static_cast<const byte *>(data), len, recv_info, flags);
 	free(data);
 	return ret;
 }
@@ -708,8 +700,6 @@ int SctpTransport::SendCallback(struct socket *sock, uint32_t sb_free) {
 	void *ptr = sconn->sconn_addr;
 	auto *transport = static_cast<SctpTransport *>(ptr);
 
-	// Workaround for sctplab/usrsctp#405: Send callback is invoked on already closed socket
-	// https://github.com/sctplab/usrsctp/issues/405
 	std::shared_lock lock(InstancesMutex);
 	if (Instances.find(transport) == Instances.end())
 		return -1;
@@ -718,8 +708,15 @@ int SctpTransport::SendCallback(struct socket *sock, uint32_t sb_free) {
 }
 
 int SctpTransport::WriteCallback(void *ptr, void *data, size_t len, uint8_t tos, uint8_t set_df) {
-	return static_cast<SctpTransport *>(ptr)->handleWrite(static_cast<byte *>(data), len, tos,
-	                                                      set_df);
+	auto *transport = static_cast<SctpTransport *>(ptr);
+
+	// Workaround for sctplab/usrsctp#405: Send callback is invoked on already closed socket
+	// https://github.com/sctplab/usrsctp/issues/405
+	std::shared_lock lock(InstancesMutex);
+	if (Instances.find(transport) == Instances.end())
+		return -1;
+
+	return transport->handleWrite(static_cast<byte *>(data), len, tos, set_df);
 }
 
 } // namespace rtc

src/sctptransport.hpp

@@ -86,7 +86,7 @@ private:
 	int handleSend(size_t free);
 	int handleWrite(byte *data, size_t len, uint8_t tos, uint8_t set_df);
 
-	void processData(const byte *data, size_t len, uint16_t streamId, PayloadId ppid);
+	void processData(binary &&data, uint16_t streamId, PayloadId ppid);
 	void processNotification(const union sctp_notification *notify, size_t len);
 
 	const uint16_t mPort;

src/transport.hpp

@@ -65,10 +65,20 @@ public:
 	virtual bool send(message_ptr message) { return outgoing(message); }
 
 protected:
-	void recv(message_ptr message) { mRecvCallback(message); }
+	void recv(message_ptr message) {
+		try {
+			mRecvCallback(message);
+		} catch (const std::exception &e) {
+			PLOG_WARNING << e.what();
+		}
+	}
 	void changeState(State state) {
-		if (mState.exchange(state) != state)
-			mStateChangeCallback(state);
+		try {
+			if (mState.exchange(state) != state)
+				mStateChangeCallback(state);
+		} catch (const std::exception &e) {
+			PLOG_WARNING << e.what();
+		}
 	}
 
 	virtual void incoming(message_ptr message) { recv(message); }

src/websocket.cpp

@@ -317,7 +317,7 @@ void WebSocket::closeTransports() {
 	// Reset callbacks now that state is changed
 	resetCallbacks();
 
-	// Pass the references to a thread, allowing to terminate a transport from its own thread
+	// Pass the pointers to a thread, allowing to terminate a transport from its own thread
 	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));