Merge pull request #386 from paullouisageneau/message-size

Add setting to change local maximum message size

DOC.md

@@ -92,6 +92,7 @@ Arguments:
   - `portRangeBegin` (optional): first port (included) of the allowed local port range (0 if unused)
   - `portRangeEnd` (optional): last port (included) of the allowed local port (0 if unused)
   - `mtu` (optional): manually set the Maximum Transfer Unit (MTU) for the connection (0 if automatic)
+  - `maxMessageSize` (optional): manually set the local maximum message size for Data Channels (0 if default)
 
 Return value: the identifier of the new Peer Connection or a negative error code.
 

include/rtc/configuration.hpp

@@ -86,6 +86,9 @@ struct RTC_CPP_EXPORT Configuration {
 
 	// MTU
 	optional<size_t> mtu;
+
+	// Local max message size at reception
+	optional<size_t> maxMessageSize;
 };
 
 } // namespace rtc

include/rtc/description.hpp

@@ -228,11 +228,13 @@ public:
 	int addApplication(string mid = "data");
 	int addVideo(string mid = "video", Direction dir = Direction::SendOnly);
 	int addAudio(string mid = "audio", Direction dir = Direction::SendOnly);
+	void clearMedia();
 
 	variant<Media *, Application *> media(unsigned int index);
 	variant<const Media *, const Application *> media(unsigned int index) const;
 	unsigned int mediaCount() const;
 
+	const Application *application() const;
 	Application *application();
 
 	static Type stringToType(const string &typeString);

include/rtc/rtc.h

@@ -131,6 +131,7 @@ typedef struct {
 	uint16_t portRangeBegin;
 	uint16_t portRangeEnd;
 	int mtu; // <= 0 means automatic
+	int maxMessageSize; // <= 0 means default
 } rtcConfiguration;
 
 typedef struct {

src/capi.cpp

@@ -360,9 +360,12 @@ int rtcCreatePeerConnection(const rtcConfiguration *config) {
 			c.portRangeEnd = config->portRangeEnd;
 		}
 
-		if(config->mtu > 0)
+		if (config->mtu > 0)
 			c.mtu = size_t(config->mtu);
 
+		if (config->maxMessageSize)
+			c.maxMessageSize = size_t(config->maxMessageSize);
+
 		return emplacePeerConnection(std::make_shared<PeerConnection>(c));
 	});
 }

src/description.cpp

@@ -414,6 +414,8 @@ int Description::addMedia(Application application) {
 
 int Description::addApplication(string mid) { return addMedia(Application(std::move(mid))); }
 
+const Description::Application *Description::application() const { return mApplication.get(); }
+
 Description::Application *Description::application() { return mApplication.get(); }
 
 int Description::addVideo(string mid, Direction dir) {
@@ -424,6 +426,11 @@ int Description::addAudio(string mid, Direction dir) {
 	return addMedia(Audio(std::move(mid), dir));
 }
 
+void Description::clearMedia() {
+	mEntries.clear();
+	mApplication.reset();
+}
+
 variant<Description::Media *, Description::Application *> Description::media(unsigned int index) {
 	if (index >= mEntries.size())
 		throw std::out_of_range("Media index out of range");

src/globals.hpp

@@ -26,9 +26,10 @@ namespace rtc {
 const size_t MAX_NUMERICNODE_LEN = 48; // Max IPv6 string representation length
 const size_t MAX_NUMERICSERV_LEN = 6;  // Max port string representation length
 
-const uint16_t DEFAULT_SCTP_PORT = 5000;          // SCTP port to use by default
-const size_t DEFAULT_MAX_MESSAGE_SIZE = 65536;    // Remote max message size if not specified in SDP
-const size_t LOCAL_MAX_MESSAGE_SIZE = 256 * 1024; // Local max message size
+const uint16_t DEFAULT_SCTP_PORT = 5000;                  // SCTP port to use by default
+
+const size_t DEFAULT_LOCAL_MAX_MESSAGE_SIZE = 256 * 1024; // Default local max message size
+const size_t DEFAULT_MAX_MESSAGE_SIZE = 65536; // Remote max message size if not specified in SDP
 
 const size_t RECV_QUEUE_LIMIT = 1024 * 1024; // Max per-channel queue size
 

src/impl/datachannel.cpp

@@ -165,14 +165,8 @@ bool DataChannel::isOpen(void) const { return mIsOpen; }
 bool DataChannel::isClosed(void) const { return mIsClosed; }
 
 size_t DataChannel::maxMessageSize() const {
-	size_t remoteMax = DEFAULT_MAX_MESSAGE_SIZE;
-	if (auto pc = mPeerConnection.lock())
-		if (auto description = pc->remoteDescription())
-			if (auto *application = description->application())
-				if (auto maxMessageSize = application->maxMessageSize())
-					remoteMax = *maxMessageSize > 0 ? *maxMessageSize : LOCAL_MAX_MESSAGE_SIZE;
-
-	return std::min(remoteMax, LOCAL_MAX_MESSAGE_SIZE);
+	auto pc = mPeerConnection.lock();
+	return pc ? pc->remoteMaxMessageSize() : DEFAULT_MAX_MESSAGE_SIZE;
 }
 
 void DataChannel::shiftStream() {
@@ -260,8 +254,7 @@ NegotiatedDataChannel::NegotiatedDataChannel(weak_ptr<PeerConnection> pc, uint16
     : DataChannel(pc, stream, std::move(label), std::move(protocol), std::move(reliability)) {}
 
 NegotiatedDataChannel::NegotiatedDataChannel(weak_ptr<PeerConnection> pc,
-                                             weak_ptr<SctpTransport> transport,
-                                             uint16_t stream)
+                                             weak_ptr<SctpTransport> transport, uint16_t stream)
     : DataChannel(pc, stream, "", "", {}) {
 	mSctpTransport = transport;
 }

src/impl/peerconnection.cpp

@@ -98,6 +98,23 @@ optional<Description> PeerConnection::remoteDescription() const {
 	return mRemoteDescription;
 }
 
+size_t PeerConnection::remoteMaxMessageSize() const {
+	const size_t localMax = config.maxMessageSize.value_or(DEFAULT_LOCAL_MAX_MESSAGE_SIZE);
+
+	size_t remoteMax = DEFAULT_MAX_MESSAGE_SIZE;
+	std::lock_guard lock(mRemoteDescriptionMutex);
+	if (mRemoteDescription)
+		if (auto *application = mRemoteDescription->application())
+			if (auto max = application->maxMessageSize()) {
+				// RFC 8841: If the SDP "max-message-size" attribute contains a maximum message
+				// size value of zero, it indicates that the SCTP endpoint will handle messages
+				// of any size, subject to memory capacity, etc.
+				remoteMax = *max > 0 ? *max : std::numeric_limits<size_t>::max();
+			}
+
+	return std::min(remoteMax, localMax);
+}
+
 shared_ptr<IceTransport> PeerConnection::initIceTransport() {
 	try {
 		if (auto transport = std::atomic_load(&mIceTransport))
@@ -248,7 +265,7 @@ shared_ptr<SctpTransport> PeerConnection::initSctpTransport() {
 		uint16_t sctpPort = remote->application()->sctpPort().value_or(DEFAULT_SCTP_PORT);
 		auto lower = std::atomic_load(&mDtlsTransport);
 		auto transport = std::make_shared<SctpTransport>(
-		    lower, sctpPort, config.mtu, weak_bind(&PeerConnection::forwardMessage, this, _1),
+		    lower, config, sctpPort, weak_bind(&PeerConnection::forwardMessage, this, _1),
 		    weak_bind(&PeerConnection::forwardBufferedAmount, this, _1, _2),
 		    [this, weak_this = weak_from_this()](SctpTransport::State transportState) {
 			    auto shared_this = weak_this.lock();
@@ -712,6 +729,11 @@ void PeerConnection::validateRemoteDescription(const Description &description) {
 }
 
 void PeerConnection::processLocalDescription(Description description) {
+	const size_t localSctpPort = DEFAULT_SCTP_PORT;
+	const size_t localMaxMessageSize = config.maxMessageSize.value_or(DEFAULT_LOCAL_MAX_MESSAGE_SIZE);
+
+	// Clean up the application entry the ICE transport might have added already (libnice)
+	description.clearMedia();
 
 	if (auto remote = remoteDescription()) {
 		// Reciprocate remote description
@@ -723,8 +745,8 @@ void PeerConnection::processLocalDescription(Description description) {
 				        if (!mDataChannels.empty()) {
 					        // Prefer local description
 					        Description::Application app(remoteApp->mid());
-					        app.setSctpPort(DEFAULT_SCTP_PORT);
-					        app.setMaxMessageSize(LOCAL_MAX_MESSAGE_SIZE);
+					        app.setSctpPort(localSctpPort);
+					        app.setMaxMessageSize(localMaxMessageSize);
 
 					        PLOG_DEBUG << "Adding application to local description, mid=\""
 					                   << app.mid() << "\"";
@@ -734,8 +756,8 @@ void PeerConnection::processLocalDescription(Description description) {
 				        }
 
 				        auto reciprocated = remoteApp->reciprocate();
-				        reciprocated.hintSctpPort(DEFAULT_SCTP_PORT);
-				        reciprocated.setMaxMessageSize(LOCAL_MAX_MESSAGE_SIZE);
+				        reciprocated.hintSctpPort(localSctpPort);
+				        reciprocated.setMaxMessageSize(localMaxMessageSize);
 
 				        PLOG_DEBUG << "Reciprocating application in local description, mid=\""
 				                   << reciprocated.mid() << "\"";
@@ -799,8 +821,8 @@ void PeerConnection::processLocalDescription(Description description) {
 				while (description.hasMid(std::to_string(m)))
 					++m;
 				Description::Application app(std::to_string(m));
-				app.setSctpPort(DEFAULT_SCTP_PORT);
-				app.setMaxMessageSize(LOCAL_MAX_MESSAGE_SIZE);
+				app.setSctpPort(localSctpPort);
+				app.setMaxMessageSize(localMaxMessageSize);
 
 				PLOG_DEBUG << "Adding application to local description, mid=\"" << app.mid()
 				           << "\"";

src/impl/peerconnection.hpp

@@ -47,6 +47,7 @@ struct PeerConnection : std::enable_shared_from_this<PeerConnection> {
 
 	optional<Description> localDescription() const;
 	optional<Description> remoteDescription() const;
+	size_t remoteMaxMessageSize() const;
 
 	shared_ptr<IceTransport> initIceTransport();
 	shared_ptr<DtlsTransport> initDtlsTransport();

src/impl/sctptransport.cpp

@@ -100,8 +100,9 @@ void SctpTransport::Cleanup() {
 		std::this_thread::sleep_for(100ms);
 }
 
-SctpTransport::SctpTransport(shared_ptr<Transport> lower, uint16_t port, optional<size_t> mtu,
-                             message_callback recvCallback, amount_callback bufferedAmountCallback,
+SctpTransport::SctpTransport(shared_ptr<Transport> lower, const Configuration &config,
+                             uint16_t port, message_callback recvCallback,
+                             amount_callback bufferedAmountCallback,
                              state_callback stateChangeCallback)
     : Transport(lower, std::move(stateChangeCallback)), mPort(port),
       mSendQueue(0, message_size_func), mBufferedAmountCallback(std::move(bufferedAmountCallback)) {
@@ -180,7 +181,7 @@ SctpTransport::SctpTransport(shared_ptr<Transport> lower, uint16_t port, optiona
 	// 1200 bytes.
 	// See https://tools.ietf.org/html/rfc8261#section-5
 #if USE_PMTUD
-	if (!mtu.has_value()) {
+	if (!config.mtu.has_value()) {
 #else
 	if (false) {
 #endif
@@ -193,7 +194,7 @@ SctpTransport::SctpTransport(shared_ptr<Transport> lower, uint16_t port, optiona
 		spp.spp_flags |= SPP_PMTUD_DISABLE;
 		// The MTU value provided specifies the space available for chunks in the
 		// packet, so we also subtract the SCTP header size.
-		size_t pmtu = mtu.value_or(DEFAULT_MTU) - 12 - 37 - 8 - 40; // SCTP/DTLS/UDP/IPv6
+		size_t pmtu = config.mtu.value_or(DEFAULT_MTU) - 12 - 37 - 8 - 40; // SCTP/DTLS/UDP/IPv6
 		spp.spp_pathmtu = uint32_t(pmtu);
 		PLOG_VERBOSE << "Path MTU discovery disabled, SCTP MTU set to " << pmtu;
 	}
@@ -222,9 +223,13 @@ SctpTransport::SctpTransport(shared_ptr<Transport> lower, uint16_t port, optiona
 		                         std::to_string(errno));
 
 	// The default send and receive window size of usrsctp is 256KiB, which is too small for
-	// realistic RTTs, therefore we increase it to 1MiB for better performance.
+	// realistic RTTs, therefore we increase it to at least 1MiB for better performance.
 	// See https://bugzilla.mozilla.org/show_bug.cgi?id=1051685
-	int bufferSize = 1024 * 1024;
+	const size_t minBufferSize = 1024 * 1024;
+
+	// Ensure the buffer is also large enough to accomodate the largest messages
+	const size_t maxMessageSize = config.maxMessageSize.value_or(DEFAULT_LOCAL_MAX_MESSAGE_SIZE);
+	const int bufferSize = int(std::max(minBufferSize, maxMessageSize * 2)); // usrsctp reads as int
 	if (usrsctp_setsockopt(mSock, SOL_SOCKET, SO_RCVBUF, &bufferSize, sizeof(bufferSize)))
 		throw std::runtime_error("Could not set SCTP recv buffer size, errno=" +
 		                         std::to_string(errno));

src/impl/sctptransport.hpp

@@ -23,6 +23,7 @@
 #include "processor.hpp"
 #include "queue.hpp"
 #include "transport.hpp"
+#include "configuration.hpp"
 
 #include <condition_variable>
 #include <functional>
@@ -42,7 +43,7 @@ public:
 
 	using amount_callback = std::function<void(uint16_t streamId, size_t amount)>;
 
-	SctpTransport(shared_ptr<Transport> lower, uint16_t port, optional<size_t> mtu,
+	SctpTransport(shared_ptr<Transport> lower, const Configuration &config, uint16_t port,
 	              message_callback recvCallback, amount_callback bufferedAmountCallback,
 	              state_callback stateChangeCallback);
 	~SctpTransport();

test/connectivity.cpp

@@ -38,6 +38,8 @@ void test_connectivity() {
 	config1.iceServers.emplace_back("stun:stun.ageneau.net:3478");
 	// Custom MTU example
 	config1.mtu = 1500;
+	// Custom max message size
+	config1.maxMessageSize = 1048576;
 
 	PeerConnection pc1(config1);
 
@@ -47,6 +49,8 @@ void test_connectivity() {
 	config2.iceServers.emplace_back("stun:stun.ageneau.net:3478");
 	// Custom MTU example
 	config2.mtu = 1500;
+	// Custom max message size
+	config2.maxMessageSize = 1048576;
 	// Port range example
 	config2.portRangeBegin = 5000;
 	config2.portRangeEnd = 6000;
@@ -140,6 +144,9 @@ void test_connectivity() {
 	if (!adc2 || !adc2->isOpen() || !dc1->isOpen())
 		throw runtime_error("DataChannel is not open");
 
+	if (dc1->maxMessageSize() != 1048576 || dc2->maxMessageSize() != 1048576)
+		throw runtime_error("DataChannel max message size is incorrect");
+
 	if (auto addr = pc1.localAddress())
 		cout << "Local address 1:  " << *addr << endl;
 	if (auto addr = pc1.remoteAddress())

test/websocket.cpp

@@ -36,10 +36,9 @@ void test_websocket() {
 
 	const string myMessage = "Hello world from libdatachannel";
 
-	WebSocket ws;
-
-	// Certificate verification can be disabled
-	// WebSocket ws(WebSocket::Configuration{.disableTlsVerification = true});
+	WebSocket::Configuration config;
+	config.disableTlsVerification = true;
+	WebSocket ws(std::move(config));
 
 	ws.onOpen([&ws, &myMessage]() {
 		cout << "WebSocket: Open" << endl;