feat: per-peer MTU (#2676)

Author: ShadauxCat

com.unity.netcode.gameobjects/CHANGELOG.md

@@ -7,8 +7,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 Additional documentation and release notes are available at [Multiplayer Documentation](https://docs-multiplayer.unity3d.com).
 
 ## [Unreleased]
+
 ### Added
 
+- Added methods NetworkManager.SetPeerMTU and NetworkManager.GetPeerMTU to be able to set MTU sizes per-peer (#2676)
+
 ### Fixed
 
 - Fixed issue where `SpawnWithObservers` was not being honored when `NetworkConfig.EnableSceneManagement` was disabled. (#2682)

com.unity.netcode.gameobjects/Runtime/Core/NetworkManager.cs

@@ -581,7 +581,12 @@ private void OnEnable()
 
         /// <summary>
         /// Sets the maximum size of a single non-fragmented message (or message batch) passed through the transport.
-        /// This should represent the transport's MTU size, minus any transport-level overhead.
+        /// This should represent the transport's default MTU size, minus any transport-level overhead.
+        /// This value will be used for any remote endpoints that haven't had per-endpoint MTUs set.
+        /// This value is also used as the size of the temporary buffer used when serializing
+        /// a single message (to avoid serializing multiple times when sending to multiple endpoints),
+        /// and thus should be large enough to ensure it can hold each message type.
+        /// This value defaults to 1296.
         /// </summary>
         /// <param name="size"></param>
         public int MaximumTransmissionUnitSize
@@ -590,6 +595,34 @@ public int MaximumTransmissionUnitSize
             get => MessageManager.NonFragmentedMessageMaxSize;
         }
 
+        /// <summary>
+        /// Set the maximum transmission unit for a specific peer.
+        /// This determines the maximum size of a message batch that can be sent to that client.
+        /// If not set for any given client, <see cref="MaximumTransmissionUnitSize"/> will be used instead.
+        /// </summary>
+        /// <param name="clientId"></param>
+        /// <param name="size"></param>
+        public void SetPeerMTU(ulong clientId, int size)
+        {
+            MessageManager.PeerMTUSizes[clientId] = size;
+        }
+
+        /// <summary>
+        /// Queries the current MTU size for a client.
+        /// If no MTU has been set for that client, will return <see cref="MaximumTransmissionUnitSize"/>
+        /// </summary>
+        /// <param name="clientId"></param>
+        /// <returns></returns>
+        public int GetPeerMTU(ulong clientId)
+        {
+            if (MessageManager.PeerMTUSizes.TryGetValue(clientId, out var ret))
+            {
+                return ret;
+            }
+
+            return MessageManager.NonFragmentedMessageMaxSize;
+        }
+
         /// <summary>
         /// Sets the maximum size of a message (or message batch) passed through the transport with the ReliableFragmented delivery.
         /// Warning: setting this value too low may result in the SDK becoming non-functional with projects that have a large number of NetworkBehaviours or NetworkVariables, as the SDK relies on the transport's ability to fragment some messages when they grow beyond the MTU size.

com.unity.netcode.gameobjects/Runtime/Messaging/NetworkMessageManager.cs

@@ -99,6 +99,8 @@ internal uint GetMessageType(Type t)
         public int NonFragmentedMessageMaxSize = DefaultNonFragmentedMessageMaxSize;
         public int FragmentedMessageMaxSize = int.MaxValue;
 
+        public Dictionary<ulong, int> PeerMTUSizes = new Dictionary<ulong, int>();
+
         internal struct MessageWithHandler
         {
             public Type MessageType;
@@ -497,6 +499,7 @@ private void CleanupDisconnectedClient(ulong clientId)
             m_SendQueues.Remove(clientId);
 
             m_PerClientMessageVersions.Remove(clientId);
+            PeerMTUSizes.Remove(clientId);
         }
 
         internal void CleanupDisconnectedClients()
@@ -678,6 +681,21 @@ internal unsafe int SendPreSerializedMessage<TMessageType>(in FastBufferWriter t
                     continue;
                 }
 
+                var startSize = NonFragmentedMessageMaxSize;
+                if (delivery != NetworkDelivery.ReliableFragmentedSequenced)
+                {
+                    if (PeerMTUSizes.TryGetValue(clientId, out var clientMaxSize))
+                    {
+                        maxSize = clientMaxSize;
+                    }
+                    startSize = maxSize;
+                    if (tmpSerializer.Position >= maxSize)
+                    {
+                        Debug.LogError($"MTU size for {clientId} is too small to contain a message of type {typeof(TMessageType).FullName}");
+                        continue;
+                    }
+                }
+
                 for (var hookIdx = 0; hookIdx < m_Hooks.Count; ++hookIdx)
                 {
                     m_Hooks[hookIdx].OnBeforeSendMessage(clientId, ref message, delivery);
@@ -686,15 +704,15 @@ internal unsafe int SendPreSerializedMessage<TMessageType>(in FastBufferWriter t
                 var sendQueueItem = m_SendQueues[clientId];
                 if (sendQueueItem.Length == 0)
                 {
-                    sendQueueItem.Add(new SendQueueItem(delivery, NonFragmentedMessageMaxSize, Allocator.TempJob, maxSize));
+                    sendQueueItem.Add(new SendQueueItem(delivery, startSize, Allocator.TempJob, maxSize));
                     sendQueueItem.ElementAt(0).Writer.Seek(sizeof(NetworkBatchHeader));
                 }
                 else
                 {
                     ref var lastQueueItem = ref sendQueueItem.ElementAt(sendQueueItem.Length - 1);
                     if (lastQueueItem.NetworkDelivery != delivery || lastQueueItem.Writer.MaxCapacity - lastQueueItem.Writer.Position < tmpSerializer.Length + headerSerializer.Length)
                     {
-                        sendQueueItem.Add(new SendQueueItem(delivery, NonFragmentedMessageMaxSize, Allocator.TempJob, maxSize));
+                        sendQueueItem.Add(new SendQueueItem(delivery, startSize, Allocator.TempJob, maxSize));
                         sendQueueItem.ElementAt(sendQueueItem.Length - 1).Writer.Seek(sizeof(NetworkBatchHeader));
                     }
                 }

com.unity.netcode.gameobjects/Tests/Editor/Messaging/MessageSendingTests.cs

@@ -181,6 +181,64 @@ public void WhenExceedingBatchSize_NewBatchesAreCreated([Values(500, 1000, 1300,
             Assert.AreEqual(2, m_MessageSender.MessageQueue.Count);
         }
 
+        [Test]
+        public void WhenExceedingPerClientBatchSizeLessThanDefault_NewBatchesAreCreated([Values(500, 1000, 1300, 2000)] int maxMessageSize)
+        {
+            var message = GetMessage();
+            m_MessageManager.NonFragmentedMessageMaxSize = maxMessageSize * 5;
+            var clients = new ulong[] { 0, 1, 2 };
+            m_MessageManager.ClientConnected(1);
+            m_MessageManager.ClientConnected(2);
+            m_MessageManager.SetVersion(1, XXHash.Hash32(typeof(TestMessage).FullName), 0);
+            m_MessageManager.SetVersion(2, XXHash.Hash32(typeof(TestMessage).FullName), 0);
+
+            for (var i = 0; i < clients.Length; ++i)
+            {
+                m_MessageManager.PeerMTUSizes[clients[i]] = maxMessageSize * (i + 1);
+            }
+
+            var size = UnsafeUtility.SizeOf<TestMessage>() + 2; // MessageHeader packed with this message will be 2 bytes
+            for (var i = 0; i < clients.Length; ++i)
+            {
+                for (var j = 0; j < ((m_MessageManager.PeerMTUSizes[clients[i]] - UnsafeUtility.SizeOf<NetworkBatchHeader>()) / size) + 1; ++j)
+                {
+                    m_MessageManager.SendMessage(ref message, NetworkDelivery.Reliable, clients[i]);
+                }
+            }
+
+            m_MessageManager.ProcessSendQueues();
+            Assert.AreEqual(2 * clients.Length, m_MessageSender.MessageQueue.Count);
+        }
+
+        [Test]
+        public void WhenExceedingPerClientBatchSizeGreaterThanDefault_OnlyOneNewBatcheIsCreated([Values(500, 1000, 1300, 2000)] int maxMessageSize)
+        {
+            var message = GetMessage();
+            m_MessageManager.NonFragmentedMessageMaxSize = 128;
+            var clients = new ulong[] { 0, 1, 2 };
+            m_MessageManager.ClientConnected(1);
+            m_MessageManager.ClientConnected(2);
+            m_MessageManager.SetVersion(1, XXHash.Hash32(typeof(TestMessage).FullName), 0);
+            m_MessageManager.SetVersion(2, XXHash.Hash32(typeof(TestMessage).FullName), 0);
+
+            for (var i = 0; i < clients.Length; ++i)
+            {
+                m_MessageManager.PeerMTUSizes[clients[i]] = maxMessageSize * (i + 1);
+            }
+
+            var size = UnsafeUtility.SizeOf<TestMessage>() + 2; // MessageHeader packed with this message will be 2 bytes
+            for (var i = 0; i < clients.Length; ++i)
+            {
+                for (var j = 0; j < ((m_MessageManager.PeerMTUSizes[clients[i]] - UnsafeUtility.SizeOf<NetworkBatchHeader>()) / size) + 1; ++j)
+                {
+                    m_MessageManager.SendMessage(ref message, NetworkDelivery.Reliable, clients[i]);
+                }
+            }
+
+            m_MessageManager.ProcessSendQueues();
+            Assert.AreEqual(2 * clients.Length, m_MessageSender.MessageQueue.Count);
+        }
+
         [Test]
         public void WhenExceedingMTUSizeWithFragmentedDelivery_NewBatchesAreNotCreated([Values(500, 1000, 1300, 2000)] int maxMessageSize)
         {