Add PacingHandler

MediaHandler that can be used to pace packet delivery

Resolves #1017

Co-authored-by: Paul-Louis Ageneau <[email protected]>

CMakeLists.txt

@@ -87,6 +87,7 @@ set(LIBDATACHANNEL_SOURCES
 	${CMAKE_CURRENT_SOURCE_DIR}/src/rtp.cpp
 	${CMAKE_CURRENT_SOURCE_DIR}/src/capi.cpp
 	${CMAKE_CURRENT_SOURCE_DIR}/src/plihandler.cpp
+	${CMAKE_CURRENT_SOURCE_DIR}/src/pacinghandler.cpp
 )
 
 set(LIBDATACHANNEL_HEADERS
@@ -123,6 +124,7 @@ set(LIBDATACHANNEL_HEADERS
 	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/rtcpnackresponder.hpp
 	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/utils.hpp
 	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/plihandler.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/pacinghandler.hpp
 	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/version.h
 )
 

include/rtc/pacinghandler.hpp

@@ -0,0 +1,51 @@
+/**
+ * Copyright (c) 2020 Staz Modrzynski
+ * Copyright (c) 2020 Paul-Louis Ageneau
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/.
+ */
+
+#ifndef RTC_PACING_HANDLER_H
+#define RTC_PACING_HANDLER_H
+
+#if RTC_ENABLE_MEDIA
+
+#include "mediahandler.hpp"
+#include "utils.hpp"
+
+#include <atomic>
+#include <queue>
+
+namespace rtc {
+
+// Paced sending of RTP packets. Takes a stream of RTP packets that can an
+// uneven bitrate. It then delivers these packets in a smoother manner by
+// sending a fixed size of them on an interval
+class RTC_CPP_EXPORT PacingHandler : public MediaHandler {
+public:
+	PacingHandler(double bitsPerSecond, std::chrono::milliseconds sendInterval);
+
+	void outgoing(message_vector &messages, const message_callback &send) override;
+
+private:
+	std::atomic<bool> mHaveScheduled = false;
+
+	double mBytesPerSecond;
+	double mBudget;
+
+	std::chrono::milliseconds mSendInterval;
+	std::chrono::time_point<std::chrono::high_resolution_clock> mLastRun;
+
+	std::mutex mMutex;
+	std::queue<message_ptr> mRtpBuffer;
+
+	void schedule(const message_callback &send);
+};
+
+} // namespace rtc
+
+#endif // RTC_ENABLE_MEDIA
+
+#endif // RTC_PACING_HANDLER_H

include/rtc/rtc.hpp

@@ -34,6 +34,7 @@
 #include "h265rtppacketizer.hpp"
 #include "mediahandler.hpp"
 #include "plihandler.hpp"
+#include "pacinghandler.hpp"
 #include "rtcpnackresponder.hpp"
 #include "rtcpreceivingsession.hpp"
 #include "rtcpsrreporter.hpp"

src/impl/track.cpp

@@ -142,7 +142,11 @@ void Track::incoming(message_ptr message) {
 
 	message_vector messages{std::move(message)};
 	if (auto handler = getMediaHandler())
-		handler->incomingChain(messages, [this](message_ptr m) { transportSend(m); });
+		handler->incomingChain(messages, [this, weak_this = weak_from_this()](message_ptr m) {
+			if (auto locked = weak_this.lock()) {
+				transportSend(m);
+			}
+		});
 
 	for (auto &m : messages) {
 		// Tail drop if queue is full
@@ -175,7 +179,11 @@ bool Track::outgoing(message_ptr message) {
 
 	if (handler) {
 		message_vector messages{std::move(message)};
-		handler->outgoingChain(messages, [this](message_ptr m) { transportSend(m); });
+		handler->outgoingChain(messages, [this, weak_this = weak_from_this()](message_ptr m) {
+			if (auto locked = weak_this.lock()) {
+				transportSend(m);
+			}
+		});
 		bool ret = false;
 		for (auto &m : messages)
 			ret = transportSend(std::move(m));

src/pacinghandler.cpp

@@ -0,0 +1,75 @@
+/**
+ * Copyright (c) 2020 Filip Klembara (in2core)
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/.
+ */
+
+#if RTC_ENABLE_MEDIA
+
+#include <memory>
+
+#include "pacinghandler.hpp"
+
+#include "impl/internals.hpp"
+#include "impl/threadpool.hpp"
+
+namespace rtc {
+
+PacingHandler::PacingHandler(double bitsPerSecond, std::chrono::milliseconds sendInterval)
+    : mBytesPerSecond(bitsPerSecond / 8), mBudget(0), mSendInterval(sendInterval){};
+
+void PacingHandler::schedule(const message_callback &send) {
+	if (!mHaveScheduled.exchange(true)) {
+		return;
+	}
+
+	impl::ThreadPool::Instance().schedule(mSendInterval, [this, weak_this = weak_from_this(),
+	                                                      send]() {
+		if (auto locked = weak_this.lock()) {
+			const std::lock_guard<std::mutex> lock(mMutex);
+			mHaveScheduled.store(false);
+
+			// Update the budget and cap it
+			auto newBudget =
+			    std::chrono::duration<double>(std::chrono::high_resolution_clock::now() - mLastRun)
+			        .count() *
+			    mBytesPerSecond;
+			auto maxBudget = std::chrono::duration<double>(mSendInterval).count() * mBytesPerSecond;
+			mBudget = std::min(mBudget + newBudget, maxBudget);
+			mLastRun = std::chrono::high_resolution_clock::now();
+
+			// Send packets while there is budget, allow a single partial packet over budget
+			while (!mRtpBuffer.empty() && mBudget > 0) {
+				auto size = int(mRtpBuffer.front()->size());
+				send(std::move(mRtpBuffer.front()));
+				mRtpBuffer.pop();
+				mBudget -= size;
+			}
+
+			if (!mRtpBuffer.empty()) {
+				printf("\n more \n");
+				schedule(send);
+			} else {
+				printf("\n empty \n");
+			}
+		}
+	});
+}
+
+void PacingHandler::outgoing(message_vector &messages, const message_callback &send) {
+
+	std::lock_guard<std::mutex> lock(mMutex);
+
+	for (auto &m : messages) {
+		mRtpBuffer.push(std::move(m));
+	}
+	messages.clear();
+
+	schedule(send);
+}
+
+} // namespace rtc
+
+#endif /* RTC_ENABLE_MEDIA */