feat: Queue extra packets in the adapter (1.0.0) (#1514)

* Add batched send/receive queues

The send queue is meant to be used to store individual messages and read
them back in batched chunks (to be sent on the network). It's also meant
to act as a sort of overflow storage when UTP internal queues are full.

The receive queue is meant to store batched messages (batched by the
send queue) and read them out one at a time.

* Allow tests to access internals

* Remove dummy tests

* Update UnityTransport to use new batched queues

* Update transport tests

* Add latest changes to CHANGELOG

* Minor stylistic improvements for consistency

* Fix whitespace issue

* Ignore test until fix for UTP is released

* Correctly handle ArraySegment not at beginning of array

When handling incoming messages from the transport

* Don't change how existing settings behave

m_SendQueueBatchSize was effectively the maximum payload size and this
is what it will stay as. The new maximum payload size setting has been
removed.

The size of the send queue is now controlled by m_MaxSendQueueSize.

* Improve "too many in-flight" error message

* Had left merge markers in CHANGELOG...

* Stylistic improvements

* Fix standards issues

* Enable test that was blocked on UTP 1.0.0-pre.10

Author: simon-lemay-unity

com.unity.netcode.adapter.utp/CHANGELOG.md

@@ -3,21 +3,27 @@ All notable changes to this package will be documented in this file. The format
 
 ## [Unreleased]
 
+### Added
+
+- Added new 'Max Send Queue Size' configuration field in the inspector. This controls the size of the send queue that is used to accumulate small sends together and also acts as an overflow queue when there are too many in-flight packets or when other internal queues are full.
+
 ### Changed
 
 - Removed 'Maximum Packet Size' configuration field in the inspector. This would cause confusion since the maximum packet size is in effect always the MTU (1400 bytes on most platforms).
 - Updated com.unity.transport to 1.0.0-pre.10.
+- The 'Send Queue Batch Size' configuration field now controls the size of the send queue, rather than the size of a single batch of messages. Consequently, it should be set much higher than it was previously.
 - All delivery methods now support fragmentation, meaning the 'Send Queue Batch Size' setting (which controls the maximum payload size) now applies to all delivery methods, not just reliable ones.
 
 
 ### Fixed
 
 - Fixed packet overflow errors when sending payloads too close to the MTU (was mostly visible when using Relay).
 - Don't throw an exception when the host disconnects (issue 1439 on GitHub).
+- Avoid "too many inflight packets" errors by queueing packets in a queue when the limit of inflight packets is reached in UTP. The size of this queue can be controlled with the 'Max Send Queue Size' configuration field.
 
 ## [1.0.0-pre.3] - 2021-10-22
 
-#### Added 
+### Added 
 
 - Exposed `m_HeartbeatTimeoutMS`, `m_ConnectTimeoutMS`, `m_MaxConnectAttempts`, and `m_DisconnectTimeoutMS` parameters. (#1314)
 

com.unity.netcode.adapter.utp/Runtime/AssemblyInfo.cs

@@ -0,0 +1,4 @@
+using System.Runtime.CompilerServices;
+
+[assembly: InternalsVisibleTo("Unity.Netcode.Adapter.UTP.EditorTests")]
+[assembly: InternalsVisibleTo("Unity.Netcode.Adapter.UTP.RuntimeTests")]

com.unity.netcode.adapter.utp/Tests/Runtime/DummyTestScript.cs.meta renamed to com.unity.netcode.adapter.utp/Runtime/AssemblyInfo.cs.meta

@@ -0,0 +1,56 @@
+using System;
+using Unity.Networking.Transport;
+
+namespace Unity.Netcode.UTP.Utilities
+{
+    /// <summary>Queue for batched messages received through UTP.</summary>
+    /// <remarks>This is meant as a companion to <see cref="BatchedSendQueue"/>.</remarks>
+    internal class BatchedReceiveQueue
+    {
+        private byte[] m_Data;
+        private DataStreamReader m_Reader;
+        private int m_ReaderOffset;
+
+        public bool IsEmpty => m_ReaderOffset >= m_Reader.Length;
+
+        /// <summary>
+        /// Construct a new receive queue from a <see cref="DataStreamReader"/> returned by
+        /// <see cref="NetworkDriver"/> when popping a data event.
+        /// </summary>
+        /// <param name="reader">The <see cref="DataStreamReader"/> to construct from.</param>
+        public BatchedReceiveQueue(DataStreamReader reader)
+        {
+            m_Data = new byte[reader.Length];
+            unsafe
+            {
+                fixed (byte* dataPtr = m_Data)
+                {
+                    reader.ReadBytes(dataPtr, reader.Length);
+                }
+            }
+
+            m_Reader = reader;
+            m_ReaderOffset = 0;
+        }
+
+        /// <summary>Pop the next message in the queue.</summary>
+        /// <returns>The message, or the default value if no more messages.</returns>
+        public ArraySegment<byte> PopMessage()
+        {
+            if (IsEmpty)
+            {
+                return default;
+            }
+
+            m_Reader.SeekSet(m_ReaderOffset);
+
+            var messageLength = m_Reader.ReadInt();
+            m_ReaderOffset += sizeof(int);
+
+            var data = new ArraySegment<byte>(m_Data, m_ReaderOffset, messageLength);
+            m_ReaderOffset += messageLength;
+
+            return data;
+        }
+    }
+}

com.unity.netcode.adapter.utp/Runtime/BatchedReceiveQueue.cs

@@ -0,0 +1,198 @@
+using System;
+using Unity.Collections;
+using Unity.Collections.LowLevel.Unsafe;
+using Unity.Networking.Transport;
+
+namespace Unity.Netcode.UTP.Utilities
+{
+    /// <summary>Queue for batched messages meant to be sent through UTP.</summary>
+    /// <remarks>
+    /// Messages should be pushed on the queue with <see cref="PushMessage"/>. To send batched
+    /// messages, call <see cref="FillWriter"> with the <see cref="DataStreamWriter"/> obtained from
+    /// <see cref="NetworkDriver.BeginSend"/>. This will fill the writer with as many messages as
+    /// possible. If the send is successful, call <see cref="Consume"/> to remove the data from the
+    /// queue.
+    ///
+    /// This is meant as a companion to <see cref="BatchedReceiveQueue"/>, which should be used to
+    /// read messages sent with this queue.
+    /// </remarks>
+    internal struct BatchedSendQueue : IDisposable
+    {
+        private NativeArray<byte> m_Data;
+        private NativeArray<int> m_HeadTailIndices;
+
+        /// <summary>Overhead that is added to each message in the queue.</summary>
+        public const int PerMessageOverhead = sizeof(int);
+
+        // Indices into m_HeadTailIndicies.
+        private const int k_HeadInternalIndex = 0;
+        private const int k_TailInternalIndex = 1;
+
+        /// <summary>Index of the first byte of the oldest data in the queue.</summary>
+        private int HeadIndex
+        {
+            get { return m_HeadTailIndices[k_HeadInternalIndex]; }
+            set { m_HeadTailIndices[k_HeadInternalIndex] = value; }
+        }
+
+        /// <summary>Index one past the last byte of the most recent data in the queue.</summary>
+        private int TailIndex
+        {
+            get { return m_HeadTailIndices[k_TailInternalIndex]; }
+            set { m_HeadTailIndices[k_TailInternalIndex] = value; }
+        }
+
+        public int Length => TailIndex - HeadIndex;
+
+        public bool IsEmpty => HeadIndex == TailIndex;
+
+        public bool IsCreated => m_Data.IsCreated;
+
+        /// <summary>Construct a new empty send queue.</summary>
+        /// <param name="capacity">Maximum capacity of the send queue.</param>
+        public BatchedSendQueue(int capacity)
+        {
+            m_Data = new NativeArray<byte>(capacity, Allocator.Persistent);
+            m_HeadTailIndices = new NativeArray<int>(2, Allocator.Persistent);
+
+            HeadIndex = 0;
+            TailIndex = 0;
+        }
+
+        public void Dispose()
+        {
+            if (IsCreated)
+            {
+                m_Data.Dispose();
+                m_HeadTailIndices.Dispose();
+            }
+        }
+
+        /// <summary>Append data at the tail of the queue. No safety checks.</summary>
+        private void AppendDataAtTail(ArraySegment<byte> data)
+        {
+            unsafe
+            {
+                var writer = new DataStreamWriter((byte*)m_Data.GetUnsafePtr() + TailIndex, m_Data.Length - TailIndex);
+
+                writer.WriteInt(data.Count);
+
+                fixed (byte* dataPtr = data.Array)
+                {
+                    writer.WriteBytes(dataPtr + data.Offset, data.Count);
+                }
+            }
+
+            TailIndex += sizeof(int) + data.Count;
+        }
+
+        /// <summary>Append a new message to the queue.</summary>
+        /// <param name="message">Message to append to the queue.</param>
+        /// <returns>
+        /// Whether the message was appended successfully. The only way it can fail is if there's
+        /// no more room in the queue. On failure, nothing is written to the queue.
+        /// </returns>
+        public bool PushMessage(ArraySegment<byte> message)
+        {
+            if (!IsCreated)
+            {
+                return false;
+            }
+
+            // Check if there's enough room after the current tail index.
+            if (m_Data.Length - TailIndex >= sizeof(int) + message.Count)
+            {
+                AppendDataAtTail(message);
+                return true;
+            }
+
+            // Check if there would be enough room if we moved data at the beginning of m_Data.
+            if (m_Data.Length - TailIndex + HeadIndex >= sizeof(int) + message.Count)
+            {
+                // Move the data back at the beginning of m_Data.
+                unsafe
+                {
+                    UnsafeUtility.MemMove(m_Data.GetUnsafePtr(), (byte*)m_Data.GetUnsafePtr() + HeadIndex, Length);
+                }
+
+                TailIndex = Length;
+                HeadIndex = 0;
+
+                AppendDataAtTail(message);
+                return true;
+            }
+
+            return false;
+        }
+
+        /// <summary>
+        /// Fill as much of a <see cref="DataStreamWriter"/> as possible with data from the head of
+        /// the queue. Only full messages (and their length) are written to the writer.
+        /// </summary>
+        /// <remarks>
+        /// This does NOT actually consume anything from the queue. That is, calling this method
+        /// does not reduce the length of the queue. Callers are expected to call
+        /// <see cref="Consume"/> with the value returned by this method afterwards if the data can
+        /// be safely removed from the queue (e.g. if it was sent successfully).
+        /// </remarks>
+        /// <param name="writer">The <see cref="DataStreamWriter"/> to write to.</param>
+        /// <returns>How many bytes were written to the writer.</returns>
+        public int FillWriter(ref DataStreamWriter writer)
+        {
+            if (!IsCreated || Length == 0)
+            {
+                return 0;
+            }
+
+            unsafe
+            {
+                var reader = new DataStreamReader((byte*)m_Data.GetUnsafePtr() + HeadIndex, Length);
+
+                var writerAvailable = writer.Capacity;
+                var readerOffset = 0;
+
+                while (readerOffset < Length)
+                {
+                    reader.SeekSet(readerOffset);
+                    var messageLength = reader.ReadInt();
+
+                    if (writerAvailable < sizeof(int) + messageLength)
+                    {
+                        break;
+                    }
+                    else
+                    {
+                        writer.WriteInt(messageLength);
+
+                        var messageOffset = HeadIndex + reader.GetBytesRead();
+                        writer.WriteBytes((byte*)m_Data.GetUnsafePtr() + messageOffset, messageLength);
+
+                        writerAvailable -= sizeof(int) + messageLength;
+                        readerOffset += sizeof(int) + messageLength;
+                    }
+                }
+
+                return writer.Capacity - writerAvailable;
+            }
+        }
+
+        /// <summary>Consume a number of bytes from the head of the queue.</summary>
+        /// <remarks>
+        /// This should only be called with a size that matches the last value returned by
+        /// <see cref="FillWriter"/>. Anything else will result in a corrupted queue.
+        /// </remarks>
+        /// <param name="size">Number of bytes to consume from the queue.</param>
+        public void Consume(int size)
+        {
+            if (size >= Length)
+            {
+                HeadIndex = 0;
+                TailIndex = 0;
+            }
+            else
+            {
+                HeadIndex += size;
+            }
+        }
+    }
+}