feat(Peer): adding send buffer limit (#939)

* feat(Peer): adding send buffer limit

stops send buffer getting too large (holding too many packets in memory at once)

* tests

Assets/Mirage/Runtime/SocketLayer/AckSystem.cs

@@ -42,7 +42,6 @@ internal class AckSystem
         readonly Pool<ReliablePacket> reliablePool;
         readonly Metrics metrics;
 
-        //todo implement this
         readonly int maxPacketsInSendBufferPerConnection;
         readonly int maxPacketSize;
         readonly float ackTimeout;
@@ -284,10 +283,9 @@ public void SendReliable(byte[] message, int offset, int length)
         {
             if (sentAckablePackets.IsFull)
             {
-                throw new InvalidOperationException("Sent queue is full");
+                throw new InvalidOperationException($"Sent queue is full for {connection}");
             }
 
-
             if (length + MIN_RELIABLE_HEADER_SIZE > maxPacketSize)
             {
                 if (allowFragmented)
@@ -388,6 +386,8 @@ static void AddToBatch(ReliablePacket packet, byte[] message, int offset, int le
 
         void SendReliablePacket(ReliablePacket reliable)
         {
+            ThrowIfBufferLimitReached();
+
             ushort sequence = (ushort)sentAckablePackets.Enqueue(new AckablePacket(reliable));
 
             byte[] final = reliable.buffer.array;
@@ -402,6 +402,15 @@ void SendReliablePacket(ReliablePacket reliable)
             Send(final, reliable.length);
         }
 
+        private void ThrowIfBufferLimitReached()
+        {
+            // greater or equal, because we are adding 1 adder this check
+            if (sentAckablePackets.Count >= maxPacketsInSendBufferPerConnection)
+            {
+                throw new InvalidOperationException($"Max packets in send buffer reached for {connection}");
+            }
+        }
+
 
         /// <summary>
         /// Receives incoming Notify packet

Assets/Tests/SocketLayer/AckSystem/AckSystemTest_BufferSize.cs

@@ -0,0 +1,71 @@
+using System;
+using NUnit.Framework;
+
+namespace Mirage.SocketLayer.Tests.AckSystemTests
+{
+    [Category("SocketLayer")]
+    public class AckSystemTest_BufferSize : AckSystemTestBase
+    {
+        [Test]
+        public void ThrowsIfTooManyMessageAreSent()
+        {
+            var time = new Time();
+            var config = new Config
+            {
+                MaxReliablePacketsInSendBufferPerConnection = 50,
+                SequenceSize = 8
+            };
+            var instance = new AckTestInstance
+            {
+                connection = new SubIRawConnection()
+            };
+            instance.ackSystem = new AckSystem(instance.connection, config, time, bufferPool);
+
+            for (int i = 0; i < 50; i++)
+            {
+                instance.ackSystem.SendReliable(createRandomData(i));
+                // update to send batch
+                instance.ackSystem.Update();
+            }
+
+            InvalidOperationException exception = Assert.Throws<InvalidOperationException>(() =>
+            {
+                instance.ackSystem.SendReliable(createRandomData(51));
+                instance.ackSystem.Update();
+            });
+            var expected = new InvalidOperationException($"Max packets in send buffer reached for {instance.connection}");
+            Assert.That(exception, Has.Message.EqualTo(expected.Message));
+        }
+
+        [Test]
+        public void ThrowIfRingBufferIsfull()
+        {
+            var time = new Time();
+            var config = new Config
+            {
+                MaxReliablePacketsInSendBufferPerConnection = 500,
+                SequenceSize = 8
+            };
+            var instance = new AckTestInstance
+            {
+                connection = new SubIRawConnection()
+            };
+            instance.ackSystem = new AckSystem(instance.connection, config, time, bufferPool);
+
+            for (int i = 0; i < 255; i++)
+            {
+                instance.ackSystem.SendReliable(createRandomData(i));
+                // update to send batch
+                instance.ackSystem.Update();
+            }
+
+            InvalidOperationException exception = Assert.Throws<InvalidOperationException>(() =>
+            {
+                instance.ackSystem.SendReliable(createRandomData(0));
+                instance.ackSystem.Update();
+            });
+            var expected = new InvalidOperationException($"Sent queue is full for {instance.connection}");
+            Assert.That(exception, Has.Message.EqualTo(expected.Message));
+        }
+    }
+}