UTP: Add proper support for channels

com.unity.multiplayer.transport.utp/Runtime/UTPTransport.cs

@@ -36,13 +36,14 @@ private enum State
             Connected,
         }
 
+        public const int MaximumMessageLength = 6 * 1024;
+
         [SerializeField] private ProtocolType m_ProtocolType;
-        [SerializeField] private int m_MessageBufferSize = 1024;
+        [SerializeField] private int m_MessageBufferSize = MaximumMessageLength;
         [SerializeField] private string m_ServerAddress = "127.0.0.1";
         [SerializeField] private ushort m_ServerPort = 7777;
         [SerializeField] private int m_RelayMaxPlayers = 10;
         [SerializeField] private string m_RelayServer = "https://relay-allocations.cloud.unity3d.com";
-        [SerializeField] private int m_MaxFragmentationCapacity = 6 * 1024;
 
         private State m_State = State.Disconnected;
         private NetworkDriver m_Driver;
@@ -51,23 +52,28 @@ private enum State
         private string m_RelayJoinCode;
         private ulong m_ServerClientId;
 
+        private NetworkPipeline m_UnreliableSequencedPipeline;
+        private NetworkPipeline m_ReliableSequencedPipeline;
+        private NetworkPipeline m_ReliableSequencedFragmentedPipeline;
+
         public override ulong ServerClientId => m_ServerClientId;
 
         public string RelayJoinCode => m_RelayJoinCode;
 
         public ProtocolType Protocol => m_ProtocolType;
 
-        NetworkPipeline unreliableFragmentedPipeline;
-
         private void InitDriver()
         {
             if (m_NetworkParameters.Count > 0)
                 m_Driver = NetworkDriver.Create(m_NetworkParameters.ToArray());
             else
                 m_Driver = NetworkDriver.Create();
 
-            unreliableFragmentedPipeline = m_Driver.CreatePipeline(
-                    typeof(FragmentationPipelineStage));
+            m_UnreliableSequencedPipeline = m_Driver.CreatePipeline(typeof(UnreliableSequencedPipelineStage));
+            m_ReliableSequencedPipeline = m_Driver.CreatePipeline(typeof(ReliableSequencedPipelineStage));
+            m_ReliableSequencedFragmentedPipeline = m_Driver.CreatePipeline(
+                typeof(FragmentationPipelineStage), typeof(ReliableSequencedPipelineStage)
+            );
         }
 
         private void DisposeDriver()
@@ -109,6 +115,35 @@ private static RelayConnectionData ConvertConnectionData(byte[] connectionData)
             }
         }
 
+        private NetworkPipeline SelectSendPipeline(NetworkChannel channel, int size)
+        {
+            TransportChannel transportChannel = Array.Find(MLAPI_CHANNELS, tc => tc.Channel == channel);
+
+            switch (transportChannel.Delivery)
+            {
+                case NetworkDelivery.Unreliable:
+                    return NetworkPipeline.Null;
+
+                case NetworkDelivery.UnreliableSequenced:
+                    return m_UnreliableSequencedPipeline;
+
+                case NetworkDelivery.Reliable:
+                case NetworkDelivery.ReliableSequenced:
+                    return m_ReliableSequencedPipeline;
+
+                case NetworkDelivery.ReliableFragmentedSequenced:
+                    // No need to send on the fragmented pipeline if data is smaller than MTU.
+                    if (size < NetworkParameterConstants.MTU)
+                        return m_ReliableSequencedPipeline;
+
+                    return m_ReliableSequencedFragmentedPipeline;
+
+                default:
+                    Debug.LogError($"Unknown NetworkDelivery value: {transportChannel.Delivery}");
+                    return NetworkPipeline.Null;
+            }
+        }
+
         private IEnumerator ClientBindAndConnect(SocketTask task)
         {
             var serverEndpoint = default(NetworkEndPoint);
@@ -428,7 +463,11 @@ public override void Init()
             m_NetworkParameters = new List<INetworkParameter>();
 
 
-            m_NetworkParameters.Add(new FragmentationUtility.Parameters(){PayloadCapacity = m_MaxFragmentationCapacity});
+            // If we want to be able to actually handle messages MaximumMessageLength bytes in
+            // size, we need to allow a bit more than that in FragmentationUtility since this needs
+            // to account for headers and such. 128 bytes is plenty enough for such overhead.
+            var maxFragmentationCapacity = MaximumMessageLength + 128;
+            m_NetworkParameters.Add(new FragmentationUtility.Parameters(){PayloadCapacity = maxFragmentationCapacity});
 
             m_MessageBuffer = new byte[m_MessageBufferSize];
 #if ENABLE_RELAY_SERVICE
@@ -452,21 +491,21 @@ public override void Send(ulong clientId, ArraySegment<byte> data, NetworkChanne
         {
             var size = data.Count + 5;
 
-            // Debug.Log($"Sending: {String.Join(", ", data.Skip(data.Offset).Take(data.Count).Select(x => string.Format("{0:x}", x)))}");
-            var defaultPipeline = NetworkPipeline.Null;
-            if (data.Count >= NetworkParameterConstants.MTU)
-                defaultPipeline = unreliableFragmentedPipeline;
+            var pipeline = SelectSendPipeline(networkChannel, size);
 
-            if (m_Driver.BeginSend(defaultPipeline, ParseClientId(clientId), out var writer, size) == 0)
+            if (m_Driver.BeginSend(pipeline, ParseClientId(clientId), out var writer, size) == 0)
             {
                 writer.WriteByte((byte)networkChannel);
                 writer.WriteInt(data.Count);
 
-                unsafe
+                if (data.Array != null)
                 {
-                    fixed(byte* dataPtr = &data.Array[data.Offset])
+                    unsafe
                     {
-                        writer.WriteBytes(dataPtr, data.Count);
+                        fixed(byte* dataPtr = &data.Array[data.Offset])
+                        {
+                            writer.WriteBytes(dataPtr, data.Count);
+                        }
                     }
                 }
 

com.unity.multiplayer.transport.utp/Tests/Runtime/ChannelsTests.cs

@@ -0,0 +1,222 @@
+using NUnit.Framework;
+using System;
+using System.Collections;
+using System.Collections.Generic;
+using System.Linq;
+
+using MLAPI.Transports;
+using Unity.Networking.Transport;
+using UnityEngine;
+using UnityEngine.TestTools;
+
+using NetworkEvent = MLAPI.Transports.NetworkEvent;
+using UTPNetworkEvent = Unity.Networking.Transport.NetworkEvent;
+
+namespace MLAPI.UTP.RuntimeTests
+{
+    using static RuntimeTestsHelpers;
+
+    public class ChannelsTests
+    {
+        // Check that we receive the correct channel (one message after the other).
+        [UnityTest]
+        public IEnumerator ReceiveCorrectChannelSequenced()
+        {
+            UTPTransport server, client;
+            List<TransportEvent> serverEvents, clientEvents;
+
+            InitializeTransport(out server, out serverEvents);
+            InitializeTransport(out client, out clientEvents);
+
+            server.StartServer();
+            client.StartClient();
+
+            yield return WaitForNetworkEvent(NetworkEvent.Connect, serverEvents);
+
+            int eventIndex = 1;
+            foreach (var transportChannel in server.MLAPI_CHANNELS)
+            {
+                server.Send(serverEvents[0].ClientID, default(ArraySegment<byte>), transportChannel.Channel);
+
+                yield return WaitForNetworkEvent(NetworkEvent.Data, clientEvents);
+
+                Assert.AreEqual(transportChannel.Channel, clientEvents[eventIndex].Channel);
+
+                eventIndex++;
+            }
+
+            server.Shutdown();
+            client.Shutdown();
+
+            yield return null;
+        }
+
+        // Check that we receive the correct channel (all messages received at once).
+        [UnityTest]
+        public IEnumerator ReceiveCorrectChannelSameFrame()
+        {
+            UTPTransport server, client;
+            List<TransportEvent> serverEvents, clientEvents;
+
+            InitializeTransport(out server, out serverEvents);
+            InitializeTransport(out client, out clientEvents);
+
+            server.StartServer();
+            client.StartClient();
+
+            yield return WaitForNetworkEvent(NetworkEvent.Connect, serverEvents);
+
+            foreach (var transportChannel in server.MLAPI_CHANNELS)
+                client.Send(client.ServerClientId, default(ArraySegment<byte>), transportChannel.Channel);
+
+            yield return WaitForNetworkEvent(NetworkEvent.Data, serverEvents);
+
+            Assert.AreEqual(server.MLAPI_CHANNELS.Length + 1, serverEvents.Count);
+
+            int eventIndex = 1;
+            foreach (var transportChannel in server.MLAPI_CHANNELS)
+            {
+                Assert.AreEqual(transportChannel.Channel, serverEvents[eventIndex].Channel);
+                eventIndex++;
+            }
+
+            server.Shutdown();
+            client.Shutdown();
+
+            yield return null;
+        }
+
+        // Check pipeline mapping for every default channel (except fragmented).
+        [UnityTest]
+        public IEnumerator ChannelPipelineMapping()
+        {
+            UTPTransport server;
+            List<TransportEvent> serverEvents;
+            DriverClient client = new GameObject().AddComponent<DriverClient>();
+
+            InitializeTransport(out server, out serverEvents);
+
+            server.StartServer();
+            client.Connect();
+
+            yield return client.WaitForNetworkEvent(UTPNetworkEvent.Type.Connect);
+
+            foreach (var transportChannel in server.MLAPI_CHANNELS)
+            {
+                // Skip over fragmented channels (covered by different test).
+                if (transportChannel.Delivery == NetworkDelivery.ReliableFragmentedSequenced)
+                    continue;
+
+                server.Send(serverEvents[0].ClientID, default(ArraySegment<byte>), transportChannel.Channel);
+
+                yield return client.WaitForNetworkEvent(UTPNetworkEvent.Type.Data);
+
+                // Check that data's pipeline is what's expected for the delivery.
+                switch (transportChannel.Delivery)
+                {
+                    case NetworkDelivery.Unreliable:
+                        Assert.AreEqual(NetworkPipeline.Null, client.LastEventPipeline);
+                        break;
+
+                    case NetworkDelivery.UnreliableSequenced:
+                        Assert.AreEqual(client.UnreliableSequencedPipeline, client.LastEventPipeline);
+                        break;
+
+                    case NetworkDelivery.Reliable:
+                    case NetworkDelivery.ReliableSequenced:
+                        Assert.AreEqual(client.ReliableSequencedPipeline, client.LastEventPipeline);
+                        break;
+                }
+            }
+
+            server.Shutdown();
+
+            yield return null;
+        }
+
+        // Check pipeline mapping for every default channel that has fragmentation.
+        [UnityTest]
+        public IEnumerator ChannelPipelineMappingFragmented()
+        {
+            UTPTransport server;
+            List<TransportEvent> serverEvents;
+            DriverClient client = new GameObject().AddComponent<DriverClient>();
+
+            InitializeTransport(out server, out serverEvents);
+
+            server.StartServer();
+            client.Connect();
+
+            yield return client.WaitForNetworkEvent(UTPNetworkEvent.Type.Connect);
+
+            foreach (var transportChannel in server.MLAPI_CHANNELS)
+            {
+                // Skip over non-fragmented channels (covered by different test).
+                if (transportChannel.Delivery != NetworkDelivery.ReliableFragmentedSequenced)
+                    continue;
+
+                // Check that data smaller than MTU doesn't trigger fragmented pipeline.
+
+                server.Send(serverEvents[0].ClientID, default(ArraySegment<byte>), transportChannel.Channel);
+
+                yield return client.WaitForNetworkEvent(UTPNetworkEvent.Type.Data);
+
+                Assert.AreEqual(client.ReliableSequencedPipeline, client.LastEventPipeline);
+
+                // Check that data larger than MTU does trigger fragmented pipeline.
+
+                var data = new ArraySegment<byte>(new byte[UTPTransport.MaximumMessageLength]);
+                server.Send(serverEvents[0].ClientID, data, transportChannel.Channel);
+
+                yield return client.WaitForNetworkEvent(UTPNetworkEvent.Type.Data);
+
+                Assert.AreEqual(client.ReliableSequencedFragmentedPipeline, client.LastEventPipeline);
+            }
+
+            server.Shutdown();
+
+            yield return null;
+        }
+
+        // Check fragmentation on channels where it is expected.
+        [UnityTest]
+        public IEnumerator FragmentedDelivery()
+        {
+            UTPTransport server, client;
+            List<TransportEvent> serverEvents, clientEvents;
+
+            InitializeTransport(out server, out serverEvents);
+            InitializeTransport(out client, out clientEvents);
+
+            server.StartServer();
+            client.StartClient();
+
+            yield return WaitForNetworkEvent(NetworkEvent.Connect, serverEvents);
+
+            int eventIndex = 1;
+            foreach (var transportChannel in server.MLAPI_CHANNELS)
+            {
+                // Only want to test fragmentation-enabled channels.
+                if (transportChannel.Delivery != NetworkDelivery.ReliableFragmentedSequenced)
+                    continue;
+
+                var data = new byte[UTPTransport.MaximumMessageLength];
+                for (int i = 0; i < data.Length; i++)
+                    data[i] = (byte) i;
+
+                client.Send(client.ServerClientId, new ArraySegment<byte>(data), transportChannel.Channel);
+
+                yield return WaitForNetworkEvent(NetworkEvent.Data, serverEvents);
+
+                Assert.True(serverEvents[eventIndex].Data.SequenceEqual(data));
+
+                eventIndex++;
+            }
+
+            server.Shutdown();
+            client.Shutdown();
+
+            yield return null;
+        }
+    }
+}

com.unity.multiplayer.transport.utp/Tests/Runtime/ConnectionTests.cs

@@ -133,7 +133,7 @@ public IEnumerator ServerDisconnectMultipleClients()
             for (int i = 1; i < NumClients; i++)
                 server.DisconnectRemoteClient(serverEvents[i].ClientID);
 
-            // Need to manually wait since we don't know which client will got the Disconnect.
+            // Need to manually wait since we don't know which client got the Disconnect.
             yield return new WaitForSeconds(MaxNetworkEventWaitTime);
 
             // Check that all clients got a Disconnect event.