Bumped version to 0.11.10

Do not enable SCTP NR-SACKs

CMakeLists.txt

@@ -1,6 +1,6 @@
 cmake_minimum_required(VERSION 3.7)
 project(libdatachannel
-	VERSION 0.11.5
+	VERSION 0.11.10
 	LANGUAGES CXX)
 set(PROJECT_DESCRIPTION "WebRTC Data Channels Library")
 

src/peerconnection.cpp

@@ -37,17 +37,6 @@
 #include <set>
 #include <thread>
 
-#if __clang__ && defined(__APPLE__)
-namespace {
-template <typename To, typename From>
-inline std::shared_ptr<To> reinterpret_pointer_cast(std::shared_ptr<From> const &ptr) noexcept {
-	return std::shared_ptr<To>(ptr, reinterpret_cast<To *>(ptr.get()));
-}
-} // namespace
-#else
-using std::reinterpret_pointer_cast;
-#endif
-
 static rtc::LogCounter COUNTER_MEDIA_TRUNCATED(plog::warning,
                                                "Number of RTP packets truncated over past second");
 static rtc::LogCounter
@@ -701,7 +690,7 @@ void PeerConnection::forwardMedia(message_ptr message) {
 		std::set<uint32_t> ssrcs;
 		size_t offset = 0;
 		while ((sizeof(rtc::RTCP_HEADER) + offset) <= message->size()) {
-			auto header = reinterpret_cast<rtc::RTCP_HEADER *>(message->data() + offset);
+			auto header = reinterpret_cast<RTCP_HEADER *>(message->data() + offset);
 			if (header->lengthInBytes() > message->size() - offset) {
 				COUNTER_MEDIA_TRUNCATED++;
 				break;
@@ -923,12 +912,13 @@ void PeerConnection::incomingTrack(Description::Media description) {
 void PeerConnection::openTracks() {
 #if RTC_ENABLE_MEDIA
 	if (auto transport = std::atomic_load(&mDtlsTransport)) {
-		auto srtpTransport = reinterpret_pointer_cast<DtlsSrtpTransport>(transport);
+		if (auto srtpTransport = std::dynamic_pointer_cast<DtlsSrtpTransport>(transport)) {
-		std::shared_lock lock(mTracksMutex); // read-only
+			std::shared_lock lock(mTracksMutex); // read-only
-		for (auto it = mTracks.begin(); it != mTracks.end(); ++it)
+			for (auto it = mTracks.begin(); it != mTracks.end(); ++it)
-			if (auto track = it->second.lock())
+				if (auto track = it->second.lock())
-				if (!track->isOpen())
+					if (!track->isOpen())
-					track->open(srtpTransport);
+						track->open(srtpTransport);
+		}
 	}
 #endif
 }

src/sctptransport.cpp

@@ -81,11 +81,12 @@ void SctpTransport::Init() {
 
 	usrsctp_sysctl_set_sctp_max_chunks_on_queue(10 * 1024);
 
-	// Change congestion control from the default TCP Reno (RFC 2581) to H-TCP
+	// Use default congestion control (RFC 4960)
-	usrsctp_sysctl_set_sctp_default_cc_module(SCTP_CC_HTCP);
+	// See https://github.com/paullouisageneau/libdatachannel/issues/354
+	usrsctp_sysctl_set_sctp_default_cc_module(0);
 
-	// Enable Non-Renegable Selective Acknowledgments (NR-SACKs)
+	// Enable Partial Reliability Extension (RFC 3758)
-	usrsctp_sysctl_set_sctp_nrsack_enable(1);
+	usrsctp_sysctl_set_sctp_pr_enable(1);
 
 	// Increase the initial window size to 10 MTUs (RFC 6928)
 	usrsctp_sysctl_set_sctp_initial_cwnd(10);
@@ -103,7 +104,7 @@ SctpTransport::SctpTransport(std::shared_ptr<Transport> lower, uint16_t port,
                              std::optional<size_t> mtu, message_callback recvCallback,
                              amount_callback bufferedAmountCallback,
                              state_callback stateChangeCallback)
-    : Transport(lower, std::move(stateChangeCallback)), mPort(port), mPendingRecvCount(0),
+    : Transport(lower, std::move(stateChangeCallback)), mPort(port),
       mSendQueue(0, message_size_func), mBufferedAmountCallback(std::move(bufferedAmountCallback)) {
 	onRecv(recvCallback);
 
@@ -259,7 +260,7 @@ bool SctpTransport::stop() {
 		return false;
 
 	mSendQueue.stop();
-	safeFlush();
+	flush();
 	shutdown();
 	onRecv(nullptr);
 	return true;
@@ -333,13 +334,20 @@ bool SctpTransport::send(message_ptr message) {
 	return false;
 }
 
-void SctpTransport::closeStream(unsigned int stream) {
+bool SctpTransport::flush() {
-	send(make_message(0, Message::Reset, uint16_t(stream)));
+	try {
+		std::lock_guard lock(mSendMutex);
+		trySendQueue();
+		return true;
+
+	} catch (const std::exception &e) {
+		PLOG_WARNING << "SCTP flush: " << e.what();
+		return false;
+	}
 }
 
-void SctpTransport::flush() {
+void SctpTransport::closeStream(unsigned int stream) {
-	std::lock_guard lock(mSendMutex);
+	send(make_message(0, Message::Reset, uint16_t(stream)));
-	trySendQueue();
 }
 
 void SctpTransport::incoming(message_ptr message) {
@@ -427,6 +435,16 @@ void SctpTransport::doRecv() {
 	}
 }
 
+void SctpTransport::doFlush() {
+	std::lock_guard lock(mSendMutex);
+	--mPendingFlushCount;
+	try {
+		trySendQueue();
+	} catch (const std::exception &e) {
+		PLOG_WARNING << e.what();
+	}
+}
+
 bool SctpTransport::trySendQueue() {
 	// Requires mSendMutex to be locked
 	while (auto next = mSendQueue.peek()) {
@@ -533,13 +551,16 @@ void SctpTransport::updateBufferedAmount(uint16_t streamId, long delta) {
 	else
 		it->second = amount;
 
-	mSendMutex.unlock();
+	// Synchronously call the buffered amount callback
+	triggerBufferedAmount(streamId, amount);
+}
+
+void SctpTransport::triggerBufferedAmount(uint16_t streamId, size_t amount) {
 	try {
 		mBufferedAmountCallback(streamId, amount);
 	} catch (const std::exception &e) {
-		PLOG_DEBUG << "SCTP buffered amount callback: " << e.what();
+		PLOG_WARNING << "SCTP buffered amount callback: " << e.what();
 	}
-	mSendMutex.lock();
 }
 
 void SctpTransport::sendReset(uint16_t streamId) {
@@ -569,17 +590,6 @@ void SctpTransport::sendReset(uint16_t streamId) {
 	}
 }
 
-bool SctpTransport::safeFlush() {
-	try {
-		flush();
-		return true;
-
-	} catch (const std::exception &e) {
-		PLOG_WARNING << "SCTP flush: " << e.what();
-		return false;
-	}
-}
-
 void SctpTransport::handleUpcall() {
 	if (!mSock)
 		return;
@@ -593,8 +603,10 @@ void SctpTransport::handleUpcall() {
 		mProcessor.enqueue(&SctpTransport::doRecv, this);
 	}
 
-	if (events & SCTP_EVENT_WRITE)
+	if (events & SCTP_EVENT_WRITE && mPendingFlushCount == 0) {
-		mProcessor.enqueue(&SctpTransport::safeFlush, this);
+		++mPendingFlushCount;
+		mProcessor.enqueue(&SctpTransport::doFlush, this);
+	}
 }
 
 int SctpTransport::handleWrite(byte *data, size_t len, uint8_t /*tos*/, uint8_t /*set_df*/) {
@@ -709,7 +721,7 @@ void SctpTransport::processNotification(const union sctp_notification *notify, s
 		PLOG_VERBOSE << "SCTP dry event";
 		// It should not be necessary since the send callback should have been called already,
 		// but to be sure, let's try to send now.
-		safeFlush();
+		flush();
 		break;
 	}
 

src/sctptransport.hpp

@@ -51,8 +51,8 @@ public:
 	void start() override;
 	bool stop() override;
 	bool send(message_ptr message) override; // false if buffered
+	bool flush();
 	void closeStream(unsigned int stream);
-	void flush();
 
 	// Stats
 	void clearStats();
@@ -80,11 +80,12 @@ private:
 	bool outgoing(message_ptr message) override;
 
 	void doRecv();
+	void doFlush();
 	bool trySendQueue();
 	bool trySendMessage(message_ptr message);
 	void updateBufferedAmount(uint16_t streamId, long delta);
+	void triggerBufferedAmount(uint16_t streamId, size_t amount);
 	void sendReset(uint16_t streamId);
-	bool safeFlush();
 
 	void handleUpcall();
 	int handleWrite(byte *data, size_t len, uint8_t tos, uint8_t set_df);
@@ -96,8 +97,10 @@ private:
 	struct socket *mSock;
 
 	Processor mProcessor;
-	std::atomic<int> mPendingRecvCount;
+	std::atomic<int> mPendingRecvCount = 0;
-	std::mutex mRecvMutex, mSendMutex;
+	std::atomic<int> mPendingFlushCount = 0;
+	std::mutex mRecvMutex;
+	std::recursive_mutex mSendMutex; // buffered amount callback is synchronous
 	Queue<message_ptr> mSendQueue;
 	std::map<uint16_t, size_t> mBufferedAmount;
 	amount_callback mBufferedAmountCallback;

src/threadpool.cpp

@@ -51,7 +51,7 @@ void ThreadPool::spawn(int count) {
 void ThreadPool::join() {
 	{
 		std::unique_lock lock(mMutex);
-		mWaitingCondition.wait(lock, [&]() { return mWaitingWorkers == int(mWorkers.size()); });
+		mWaitingCondition.wait(lock, [&]() { return mBusyWorkers == 0; });
 		mJoining = true;
 		mTasksCondition.notify_all();
 	}
@@ -66,6 +66,8 @@ void ThreadPool::join() {
 }
 
 void ThreadPool::run() {
+	++mBusyWorkers;
+	scope_guard guard([&]() { --mBusyWorkers; });
 	while (runOne()) {
 	}
 }
@@ -81,24 +83,23 @@ bool ThreadPool::runOne() {
 std::function<void()> ThreadPool::dequeue() {
 	std::unique_lock lock(mMutex);
 	while (!mJoining) {
+		std::optional<clock::time_point> time;
 		if (!mTasks.empty()) {
-			if (mTasks.top().time <= clock::now()) {
+			time = mTasks.top().time;
+			if (*time <= clock::now()) {
 				auto func = std::move(mTasks.top().func);
 				mTasks.pop();
 				return func;
 			}
-
-			++mWaitingWorkers;
-			mWaitingCondition.notify_all();
-			mTasksCondition.wait_until(lock, mTasks.top().time);
-
-		} else {
-			++mWaitingWorkers;
-			mWaitingCondition.notify_all();
-			mTasksCondition.wait(lock);
 		}
 
-		--mWaitingWorkers;
+		--mBusyWorkers;
+		scope_guard guard([&]() { ++mBusyWorkers; });
+		mWaitingCondition.notify_all();
+		if(time)
+			mTasksCondition.wait_until(lock, *time);
+		else
+			mTasksCondition.wait(lock);
 	}
 	return nullptr;
 }

src/threadpool.hpp

@@ -72,7 +72,7 @@ protected:
 	std::function<void()> dequeue(); // returns null function if joining
 
 	std::vector<std::thread> mWorkers;
-	int mWaitingWorkers = 0;
+	std::atomic<int> mBusyWorkers = 0;
 	std::atomic<bool> mJoining = false;
 
 	struct Task {

src/track.cpp

@@ -114,7 +114,7 @@ bool Track::isOpen(void) const {
 bool Track::isClosed(void) const { return mIsClosed; }
 
 size_t Track::maxMessageSize() const {
-	return 65535 - 12 - 4; // SRTP/UDP
+	return DEFAULT_IPV4_MTU - 12 - 8 - 20; // SRTP/UDP/IPv4
 }
 
 size_t Track::availableAmount() const { return mRecvQueue.amount(); }
@@ -158,7 +158,7 @@ void Track::incoming(message_ptr message) {
 }
 
 bool Track::outgoing([[maybe_unused]] message_ptr message) {
-#if RTC_ENABLfiE_MEDIA
+#if RTC_ENABLE_MEDIA
 	std::shared_ptr<DtlsSrtpTransport> transport;
 	{
 		std::shared_lock lock(mMutex);

test/benchmark.cpp

@@ -127,8 +127,12 @@ size_t benchmark(milliseconds duration) {
 		openTime = steady_clock::now();
 
 		cout << "DataChannel open, sending data..." << endl;
-		while (dc1->bufferedAmount() == 0) {
+		try {
-			dc1->send(messageData);
+			while (dc1->bufferedAmount() == 0) {
+				dc1->send(messageData);
+			}
+		} catch (const std::exception &e) {
+			std::cout << "Send failed: " << e.what() << std::endl;
 		}
 
 		// When sent data is buffered in the DataChannel,
@@ -141,8 +145,12 @@ size_t benchmark(milliseconds duration) {
 			return;
 
 		// Continue sending
-		while (dc1->bufferedAmount() == 0) {
+		try {
-			dc1->send(messageData);
+			while (dc1->isOpen() && dc1->bufferedAmount() == 0) {
+				dc1->send(messageData);
+			}
+		} catch (const std::exception &e) {
+			std::cout << "Send failed: " << e.what() << std::endl;
 		}
 	});
 

test/capi_connectivity.cpp

@@ -326,10 +326,6 @@ int test_capi_connectivity_main() {
 	deletePeer(peer2);
 	sleep(1);
 
-	// You may call rtcCleanup() when finished to free static resources
-	rtcCleanup();
-	sleep(1);
-
 	printf("Success\n");
 	return 0;
 

test/capi_track.cpp

@@ -177,10 +177,6 @@ int test_capi_track_main() {
 	deletePeer(peer2);
 	sleep(1);
 
-	// You may call rtcCleanup() when finished to free static resources
-	rtcCleanup();
-	sleep(1);
-
 	printf("Success\n");
 	return 0;
 

test/connectivity.cpp

@@ -251,9 +251,5 @@ void test_connectivity() {
 	pc2->close();
 	this_thread::sleep_for(1s);
 
-	// You may call rtc::Cleanup() when finished to free static resources
-	rtc::Cleanup();
-	this_thread::sleep_for(1s);
-
 	cout << "Success" << endl;
 }

test/track.cpp

@@ -143,9 +143,5 @@ void test_track() {
 	pc2->close();
 	this_thread::sleep_for(1s);
 
-	// You may call rtc::Cleanup() when finished to free static resources
-	rtc::Cleanup();
-	this_thread::sleep_for(1s);
-
 	cout << "Success" << endl;
 }

test/turn_connectivity.cpp

@@ -258,9 +258,5 @@ void test_turn_connectivity() {
 	pc2->close();
 	this_thread::sleep_for(1s);
 
-	// You may call rtc::Cleanup() when finished to free static resources
-	rtc::Cleanup();
-	this_thread::sleep_for(1s);
-
 	cout << "Success" << endl;
 }

test/websocket.cpp

@@ -78,10 +78,6 @@ void test_websocket() {
 	ws->close();
 	this_thread::sleep_for(1s);
 
-	// You may call rtc::Cleanup() when finished to free static resources
-	rtc::Cleanup();
-	this_thread::sleep_for(1s);
-
 	cout << "Success" << endl;
 }