MEDIA-4876: add support for bigger data channel message size

.gitignore

@@ -37,3 +37,9 @@ git_version.*
 versioninfo.txt
 rtprcv-*
 _bwelogs*
+
+# Executables
+smb
+UnitTest2
+UnitTest
+LoadTest

CMakeLists.txt

@@ -577,6 +577,7 @@ set(TEST_FILES
     test/api/ParserTest.cpp
     test/memory/MapTest.cpp
     test/memory/PoolAllocatorTest.cpp
+    test/memory/PoolBufferTest.cpp
     test/memory/RingAllocatorTest.cpp
     test/utils/StringTokenizerTest.cpp
     test/utils/TrackerTest.cpp
@@ -651,6 +652,7 @@ set(TEST_FILES
     test/memory/StackMapTest.cpp
     test/bridge/ActiveMediaListTestLevels.h
     test/bridge/MixerTest.cpp
+    test/bridge/DataChannelMessageSizeTest.cpp
     test/bridge/VideoNackReceiveJobTest.cpp
     test/utils/LogSpamTest.cpp
     test/utils/FunctionTest.cpp

api/DataChannelMessage.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include "memory/PoolBuffer.h"
 #include "utils/StringBuilder.h"
 
 #if ENABLE_LEGACY_API
@@ -12,7 +13,8 @@ namespace api
 namespace DataChannelMessage
 {
 
-inline void makeEndpointMessage(utils::StringBuilder<2048>& outMessage,
+template <size_t T>
+inline void makeEndpointMessage(utils::StringBuilder<T>& outMessage,
     const std::string& toEndpointId,
     const std::string& fromEndpointId,
     const char* message)
@@ -33,6 +35,69 @@ inline void makeEndpointMessage(utils::StringBuilder<2048>& outMessage,
 #endif
 }
 
+inline memory::UniquePoolBuffer<memory::PacketPoolAllocator> makeUniqueEndpointMessageBuffer(
+    const std::string& toEndpointId,
+    const std::string& fromEndpointId,
+    const memory::UniquePoolBuffer<memory::PacketPoolAllocator>& payload)
+{
+#if ENABLE_LEGACY_API
+    return legacyapi::DataChannelMessage::makeUniqueEndpointMessageBuffer(toEndpointId, fromEndpointId, payload);
+#else
+    constexpr const char* TO_STRING = "{\"type\":\"EndpointMessage\",\"to\":\"";
+    constexpr const char* FROM_STRING = "\",\"from\":\"";
+    constexpr const char* MSG_STRING = "\",\"payload\":";
+    constexpr const char* TAIL_STRING = "}";
+
+    constexpr std::size_t overhead_len = std::char_traits<char>::length(TO_STRING) + 
+        std::char_traits<char>::length(FROM_STRING) + 
+        std::char_traits<char>::length(MSG_STRING) + 
+        std::char_traits<char>::length(TAIL_STRING);
+
+    const std::size_t extraLen = toEndpointId.length() + fromEndpointId.length() + payload->getLength();
+    auto buffer = memory::makeUniquePoolBuffer<memory::PacketPoolAllocator>(payload->getAllocator(), overhead_len + extraLen);
+    if (!buffer)
+    {
+        return buffer;
+    }
+
+    auto written = buffer->write(TO_STRING, std::char_traits<char>::length(TO_STRING), 0);
+    written += buffer->write(toEndpointId.c_str(), toEndpointId.length(), written);
+    written += buffer->write(FROM_STRING, std::char_traits<char>::length(FROM_STRING), written);
+    written += buffer->write(fromEndpointId.c_str(), fromEndpointId.length(), written);
+    written += buffer->write(MSG_STRING, std::char_traits<char>::length(MSG_STRING), written);
+    written += buffer->write(*payload.get(), written);
+    written += buffer->write(TAIL_STRING, std::char_traits<char>::length(TAIL_STRING), written);
+
+    assert(written == buffer->getLength());
+    return buffer;
+#endif
+}
+
+template <size_t T>
+inline void makeLoggableStringFromBuffer(memory::Array<char,T>& outArray, memory::UniquePoolBuffer<memory::PacketPoolAllocator>& payload)
+{
+    if (!payload)
+    {
+        return;
+    }
+    outArray.clear();
+    bool ellipsisNeeded = payload->getLength() > T - 1;
+
+    const size_t maxCStrLength = std::min(payload->getLength(), T - 1);
+    outArray.resize(maxCStrLength + 1);
+    const auto read = payload->copyTo(const_cast<void*>(reinterpret_cast<const void*>(outArray.data())), 0, maxCStrLength);
+    assert(read == maxCStrLength);
+    outArray[maxCStrLength] = '\0';
+
+    // Indicate that message was incompletely logged
+    if (ellipsisNeeded && T >= 4)
+    {
+        outArray[T - 2] = '.';
+        outArray[T - 3] = '.';
+        outArray[T - 4] = '.';
+    }
+}
+
 inline void makeDominantSpeaker(utils::StringBuilder<256>& outMessage, const char* endpointId)
 {
 #if ENABLE_LEGACY_API

api/Generator.cpp

@@ -210,6 +210,7 @@ nlohmann::json generateAllocateEndpointResponse(const EndpointDescription& chann
         const auto& data = channelsDescription.data.get();
         nlohmann::json dataJson;
         dataJson["port"] = data.port;
+        dataJson["max-message-size"] = data.maxMessageSize;
         responseJson["data"] = dataJson;
     }
 

api/RtcDescriptors.h

@@ -140,5 +140,6 @@ struct Video
 struct Data
 {
     uint32_t port;
+    uint32_t maxMessageSize;
 };
 } // namespace api

bridge/Mixer.cpp

@@ -1835,14 +1835,17 @@ void Mixer::sendEndpointMessage(const std::string& toEndpointId,
     const utils::SimpleJson& message)
 {
     assert(fromEndpointIdHash);
-    if (message.size() >= memory::AudioPacket::maxLength())
+    if (message.size() >= _config.sctp.maxMessageSize)
     {
+        logger::warn("Endpoint message too big, len %zu",
+            "MixerManager",
+            message.size()
+        );
         return;
     }
 
-    auto& audioAllocator = _engineMixer->getAudioAllocator();
-    auto packet = memory::makeUniquePacket(audioAllocator, message.jsonBegin(), message.size());
-    reinterpret_cast<char*>(packet->get())[message.size()] = 0; // null terminated in packet
+    auto& packetAllocator = _engineMixer->getMainAllocator();
+    auto buffer = memory::makeUniquePoolBuffer(packetAllocator, message.jsonBegin(), message.size());
 
     std::lock_guard<std::mutex> locker(_configurationLock);
 
@@ -1857,7 +1860,7 @@ void Mixer::sendEndpointMessage(const std::string& toEndpointId,
         toEndpointIdHash = dataStreamItr->second->endpointIdHash;
     }
 
-    _engineMixer->asyncSendEndpointMessage(toEndpointIdHash, fromEndpointIdHash, packet);
+    _engineMixer->asyncSendEndpointMessage(toEndpointIdHash, fromEndpointIdHash, buffer);
 }
 
 RecordingStream* Mixer::findRecordingStream(const std::string& recordingId)

bridge/MixerManager.cpp

@@ -5,7 +5,7 @@
 #include "bridge/Mixer.h"
 #include "bridge/MixerJobs.h"
 #include "bridge/VideoStream.h"
-#include "bridge/engine/Engine.h"
+
 #include "bridge/engine/EngineAudioStream.h"
 #include "bridge/engine/EngineBarbell.h"
 #include "bridge/engine/EngineDataStream.h"
@@ -492,24 +492,34 @@ void MixerManager::freeVideoPacketCache(EngineMixer& mixer, uint32_t ssrc, size_
     mixerItr->second->freeVideoPacketCache(ssrc, endpointIdHash);
 }
 
-void MixerManager::sctpReceived(EngineMixer& mixer, memory::UniquePacket msgPacket, size_t endpointIdHash)
+void MixerManager::sctpReceived(EngineMixer& mixer, memory::UniquePoolBuffer<memory::PacketPoolAllocator> message, size_t endpointIdHash)
 {
-    auto& sctpHeader = webrtc::streamMessageHeader(*msgPacket);
+    // HEADER: SctpStreamMessageHeader prepended to payload
+    // Need to get full message instead of first chunk only to form JSON from it.
+    constexpr size_t MAX_BUFFER_SIZE = 8192;
+    if (message->size() > MAX_BUFFER_SIZE) {
+        logger::warn("Received large SCTP message(size = %zu, max allowed = %zu). Dropping.", "MixerManager", message->getLength(), MAX_BUFFER_SIZE);
+        return;
+    }
+
+    char continousBuffer[message->size()];
+    message->copyTo(continousBuffer, 0, message->size());
+
+    auto& sctpHeader = *reinterpret_cast<const webrtc::SctpStreamMessageHeader*>(continousBuffer);
 
     if (sctpHeader.payloadProtocol == webrtc::DataChannelPpid::WEBRTC_ESTABLISH)
     {
         // create command with this packet to send the binary data -> engine -> WebRtcDataStream belonging to this
         // transport
-        mixer.asyncHandleSctpControl(endpointIdHash, msgPacket);
+        mixer.asyncHandleSctpControl(endpointIdHash, message);
         return; // do not free packet as we passed it on
     }
     else if (sctpHeader.payloadProtocol == webrtc::DataChannelPpid::WEBRTC_STRING)
     {
-        std::string body(reinterpret_cast<const char*>(sctpHeader.data()), msgPacket->getLength() - sizeof(sctpHeader));
+        std::string body(sctpHeader.getMessage(), message->getLength() - sizeof(sctpHeader));
         try
         {
-            auto json = utils::SimpleJson::create(reinterpret_cast<const char*>(sctpHeader.data()),
-                msgPacket->getLength() - sizeof(sctpHeader));
+            auto json = utils::SimpleJson::create(sctpHeader.getMessage(), message->getLength() - sizeof(sctpHeader));
 
             if (api::DataChannelMessageParser::isPinnedEndpointsChanged(json))
             {
@@ -573,7 +583,7 @@ void MixerManager::sctpReceived(EngineMixer& mixer, memory::UniquePacket msgPack
         logger::debug("received unexpected DataChannel payload protocol, %u, len %zu",
             "MixerManager",
             sctpHeader.payloadProtocol,
-            msgPacket->getLength());
+            message->getLength());
     }
 }
 

bridge/MixerManager.h

@@ -5,6 +5,7 @@
 #include "bridge/Stats.h"
 #include "bridge/engine/EngineMixer.h"
 #include "bridge/engine/EngineStats.h"
+#include "bridge/engine/Engine.h"
 #include "concurrency/MpmcQueue.h"
 #include "memory/PacketPoolAllocator.h"
 #include "utils/Pacer.h"
@@ -14,11 +15,6 @@
 #include <unordered_map>
 #include <vector>
 
-namespace bridge
-{
-class Engine;
-}
-
 namespace utils
 {
 class IdGenerator;
@@ -122,7 +118,9 @@ class MixerManager : public MixerManagerAsync
     void allocateVideoPacketCache(EngineMixer& mixer, uint32_t ssrc, size_t endpointIdHash) override;
     void allocateRecordingRtpPacketCache(EngineMixer& mixer, uint32_t ssrc, size_t endpointIdHash) override;
     void videoStreamRemoved(EngineMixer& engineMixer, const EngineVideoStream& videoStream) override;
-    void sctpReceived(EngineMixer& mixer, memory::UniquePacket msgPacket, size_t endpointIdHash) override;
+    void sctpReceived(EngineMixer& mixer,
+        memory::UniquePoolBuffer<memory::PacketPoolAllocator> message,
+        size_t endpointIdHash) override;
     void dataStreamRemoved(EngineMixer& mixer, const EngineDataStream& dataStream) override;
     void freeRecordingRtpPacketCache(EngineMixer& mixer, uint32_t ssrc, size_t endpointIdHash) override;
     void barbellRemoved(EngineMixer& mixer, const EngineBarbell& barbell) override;

bridge/MixerManagerAsync.cpp

@@ -52,12 +52,14 @@ bool MixerManagerAsync::asyncVideoStreamRemoved(EngineMixer& engineMixer, const
         utils::bind(&MixerManagerAsync::videoStreamRemoved, this, std::ref(engineMixer), std::cref(videoStream)));
 }
 
-bool MixerManagerAsync::asyncSctpReceived(EngineMixer& mixer, memory::UniquePacket& msgPacket, size_t endpointIdHash)
+bool MixerManagerAsync::asyncSctpReceived(EngineMixer& mixer,
+    memory::UniquePoolBuffer<memory::PacketPoolAllocator>& message,
+    size_t endpointIdHash)
 {
     return post(utils::bind(&MixerManagerAsync::sctpReceived,
         this,
         std::ref(mixer),
-        utils::moveParam(msgPacket),
+        utils::moveParam(message),
         endpointIdHash));
 }
 

bridge/MixerManagerAsync.h

@@ -1,6 +1,7 @@
 #pragma once
 #include "bridge/engine/EndpointId.h"
 #include "memory/PacketPoolAllocator.h"
+#include "memory/PoolBuffer.h"
 #include "utils/Function.h"
 #include <string>
 
@@ -44,7 +45,9 @@ class MixerManagerAsync
     virtual void allocateVideoPacketCache(EngineMixer& mixer, uint32_t ssrc, size_t endpointIdHash) = 0;
     virtual void allocateRecordingRtpPacketCache(EngineMixer& mixer, uint32_t ssrc, size_t endpointIdHash) = 0;
     virtual void videoStreamRemoved(EngineMixer& engineMixer, const EngineVideoStream& videoStream) = 0;
-    virtual void sctpReceived(EngineMixer& mixer, memory::UniquePacket msgPacket, size_t endpointIdHash) = 0;
+    virtual void sctpReceived(EngineMixer& mixer,
+        memory::UniquePoolBuffer<memory::PacketPoolAllocator> message,
+        size_t endpointIdHash) = 0;
     virtual void dataStreamRemoved(EngineMixer& mixer, const EngineDataStream& dataStream) = 0;
     virtual void freeRecordingRtpPacketCache(EngineMixer& mixer, uint32_t ssrc, size_t endpointIdHash) = 0;
     virtual void barbellRemoved(EngineMixer& mixer, const EngineBarbell& barbell) = 0;
@@ -61,7 +64,9 @@ class MixerManagerAsync
     bool asyncAllocateVideoPacketCache(EngineMixer& mixer, uint32_t ssrc, size_t endpointIdHash);
     bool asyncAllocateRecordingRtpPacketCache(EngineMixer& mixer, uint32_t ssrc, size_t endpointIdHash);
     bool asyncVideoStreamRemoved(EngineMixer& engineMixer, const EngineVideoStream& videoStream);
-    bool asyncSctpReceived(EngineMixer& mixer, memory::UniquePacket& msgPacket, size_t endpointIdHash);
+    bool asyncSctpReceived(EngineMixer& mixer,
+        memory::UniquePoolBuffer<memory::PacketPoolAllocator>& msgBuffer,
+        size_t endpointIdHash);
     bool asyncDataStreamRemoved(EngineMixer& mixer, const EngineDataStream& dataStream);
     bool asyncFreeRecordingRtpPacketCache(EngineMixer& mixer, uint32_t ssrc, size_t endpointIdHash);
     bool asyncBarbellRemoved(EngineMixer& mixer, const EngineBarbell& barbell);

bridge/endpointActions/BarbellActions.cpp

@@ -110,6 +110,7 @@ httpd::Response generateBarbellResponse(ActionContext* context,
 
     api::Data responseData;
     responseData.port = 5000;
+    responseData.maxMessageSize = 2048;
     channelsDescription.data = responseData;
 
     const auto responseBody = api::Generator::generateAllocateBarbellResponse(channelsDescription);

bridge/endpointActions/ConferenceActions.cpp

@@ -272,6 +272,7 @@ httpd::Response generateAllocateEndpointResponse(ActionContext* context,
         }
 
         responseData.port = streamDescription.sctpPort.isSet() ? streamDescription.sctpPort.get() : 5000;
+        responseData.maxMessageSize = mixer.getConfig().sctp.maxMessageSize;
         channelsDescription.data.set(responseData);
     }
 

bridge/engine/Engine.h

@@ -28,14 +28,15 @@ class Engine
 public:
     Engine(jobmanager::JobManager& backgroundJobQueue);
     Engine(jobmanager::JobManager& backgroundJobQueue, std::thread&& externalThread);
+    virtual ~Engine() = default;
 
-    void setMessageListener(MixerManagerAsync* messageListener);
+    virtual void setMessageListener(MixerManagerAsync* messageListener);
     void stop();
     void run();
 
-    bool post(utils::Function&& task) { return _tasks.push(std::move(task)); }
+    virtual bool post(utils::Function&& task) { return _tasks.push(std::move(task)); }
 
-    concurrency::SynchronizationContext getSynchronizationContext()
+    virtual concurrency::SynchronizationContext getSynchronizationContext()
     {
         return concurrency::SynchronizationContext(_tasks);
     }
@@ -62,8 +63,8 @@ class Engine
     void updateStats(uint64_t& statsPollTime, EngineStats::EngineStats& currentStatSample, uint64_t timestamp);
 
 public:
-    bool asyncAddMixer(EngineMixer* engineMixer);
-    bool asyncRemoveMixer(EngineMixer* engineMixer);
+    virtual bool asyncAddMixer(EngineMixer* engineMixer);
+    virtual bool asyncRemoveMixer(EngineMixer* engineMixer);
 
 private:
     void addMixer(EngineMixer* engineMixer);