Do network message encryption concurrently. (#5328)

In profiles of RMC-14, encrypting network messages accounted for ~8% of main thread time. That's a lot.

Each NetChannel has an "encryption channel" which gets processed on the thread pool.

RELEASE-NOTES.md

@@ -49,6 +49,7 @@ END TEMPLATE-->
 
 * Added obsoletion warning for `Control.Dispose()`. New code should not rely on it.
 * Reduced the default tickrate to 30 ticks.
+* Encryption of network messages is now done concurrently to avoid spending main thread time. In profiles, this added up to ~8% of main thread time on RMC-14.
 
 ### Internal
 

Robust.Shared/CVars.cs

@@ -382,6 +382,18 @@ protected CVars()
         public static readonly CVarDef<bool> NetLidgrenLogError =
             CVarDef.Create("net.lidgren_log_error", true);
 
+        /// <summary>
+        /// If true, run network message encryption on another thread.
+        /// </summary>
+        public static readonly CVarDef<bool> NetEncryptionThread =
+            CVarDef.Create("net.encryption_thread", true);
+
+        /// <summary>
+        /// Outstanding buffer size used by <see cref="NetEncryptionThread"/>.
+        /// </summary>
+        public static readonly CVarDef<int> NetEncryptionThreadChannelSize =
+            CVarDef.Create("net.encryption_thread_channel_size", 16);
+
         /**
          * SUS
          */

Robust.Shared/Network/NetManager.ClientConnect.cs

@@ -221,7 +221,8 @@ private async Task CCDoHandshake(NetPeerData peer, NetConnection connection, str
             _channels.Add(connection, channel);
             peer.AddChannel(channel);
 
-            _clientEncryption = encryption;
+            channel.Encryption = encryption;
+            SetupEncryptionChannel(channel);
         }
 
         private byte[] MakeAuthHash(byte[] sharedSecret, byte[] pkBytes)

Robust.Shared/Network/NetManager.NetChannel.cs

@@ -1,6 +1,8 @@
 using System;
 using System.Collections.Immutable;
 using System.Net;
+using System.Threading.Channels;
+using System.Threading.Tasks;
 using Lidgren.Network;
 using Robust.Shared.ViewVariables;
 
@@ -52,6 +54,9 @@ private sealed class NetChannel : INetChannel
 
             [ViewVariables] public int CurrentMtu => _connection.CurrentMTU;
 
+            public ChannelWriter<EncryptChannelItem>? EncryptionChannel;
+            public Task? EncryptionChannelTask;
+
             /// <summary>
             ///     Creates a new instance of a NetChannel.
             /// </summary>

Robust.Shared/Network/NetManager.Send.cs

@@ -0,0 +1,112 @@
+using System;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using Lidgren.Network;
+
+namespace Robust.Shared.Network;
+
+public sealed partial class NetManager
+{
+    // Encryption is relatively expensive, so we want to not do it on the main thread.
+    // We can't *just* thread pool everything though, because most messages still require strict ordering guarantees.
+    // For this reason, we create an "encryption channel" per player and use that to do encryption of ordered messages.
+
+    private void SetupEncryptionChannel(NetChannel netChannel)
+    {
+        if (!_config.GetCVar(CVars.NetEncryptionThread))
+            return;
+
+        // We create the encryption channel even if the channel isn't encrypted.
+        // This is to ensure consistency of behavior between local testing and production scenarios.
+
+        var channel = Channel.CreateBounded<EncryptChannelItem>(
+            new BoundedChannelOptions(_config.GetCVar(CVars.NetEncryptionThreadChannelSize))
+            {
+                FullMode = BoundedChannelFullMode.Wait,
+                SingleReader = true,
+                SingleWriter = false,
+                AllowSynchronousContinuations = false
+            });
+
+        netChannel.EncryptionChannel = channel.Writer;
+        netChannel.EncryptionChannelTask = Task.Run(async () =>
+        {
+            await EncryptionThread(channel.Reader, netChannel);
+        });
+    }
+
+    private async Task EncryptionThread(ChannelReader<EncryptChannelItem> itemChannel, NetChannel netChannel)
+    {
+        await foreach (var item in itemChannel.ReadAllAsync())
+        {
+            try
+            {
+                CoreEncryptSendMessage(netChannel, item);
+            }
+            catch (Exception e)
+            {
+                _logger.Error($"Error while encrypting message for send on channel {netChannel}: {e}");
+            }
+        }
+    }
+
+    private void CoreSendMessage(
+        NetChannel channel,
+        NetMessage message)
+    {
+        var packet = BuildMessage(message, channel.Connection.Peer);
+        var method = message.DeliveryMethod;
+
+        LogSend(message, method, packet);
+
+        var item = new EncryptChannelItem { Message = packet, Method = method };
+
+        // If the message is ordered, we have to send it to the encryption channel.
+        if (method is NetDeliveryMethod.ReliableOrdered
+            or NetDeliveryMethod.ReliableSequenced
+            or NetDeliveryMethod.UnreliableSequenced)
+        {
+            if (channel.EncryptionChannel is { } encryptionChannel)
+            {
+                var task = encryptionChannel.WriteAsync(item);
+                if (!task.IsCompleted)
+                    task.AsTask().Wait();
+            }
+            else
+            {
+                CoreEncryptSendMessage(channel, item);
+            }
+        }
+        else
+        {
+            if (Environment.CurrentManagedThreadId == _mainThreadId)
+            {
+                ThreadPool.UnsafeQueueUserWorkItem(
+                    static state => CoreEncryptSendMessage(state.channel, state.item),
+                    new
+                    {
+                        channel, item
+                    },
+                    preferLocal: true);
+            }
+            else
+            {
+                CoreEncryptSendMessage(channel, item);
+            }
+        }
+    }
+
+    private static void CoreEncryptSendMessage(NetChannel channel, EncryptChannelItem item)
+    {
+        channel.Encryption?.Encrypt(item.Message);
+
+        channel.Connection.Peer.SendMessage(item.Message, channel.Connection, item.Method);
+    }
+
+    private struct EncryptChannelItem
+    {
+        public required NetOutgoingMessage Message { get; init; }
+        public required NetDeliveryMethod Method { get; init; }
+    }
+}

Robust.Shared/Network/NetManager.cs

@@ -145,8 +145,6 @@ public sealed partial class NetManager : IClientNetManager, IServerNetManager, I
 
         public IReadOnlyDictionary<Type, long> MessageBandwidthUsage => _bandwidthUsage;
 
-        private NetEncryption? _clientEncryption;
-
         /// <inheritdoc />
         public bool IsServer { get; private set; }
 
@@ -226,6 +224,8 @@ private NetChannel? ServerChannelImpl
 
         private bool _initialized;
 
+        private int _mainThreadId;
+
         public NetManager()
         {
             _strings = new StringTable(this);
@@ -244,6 +244,8 @@ public void Initialize(bool isServer)
                 throw new InvalidOperationException("NetManager has already been initialized.");
             }
 
+            _mainThreadId = Environment.CurrentManagedThreadId;
+
             _strings.Sawmill = _logger;
 
             SynchronizeNetTime();
@@ -787,6 +789,7 @@ private async void HandleInitialHandshakeComplete(NetPeerData peer,
             _channels.Add(sender, channel);
             peer.AddChannel(channel);
             channel.Encryption = encryption;
+            SetupEncryptionChannel(channel);
 
             _strings.SendFullTable(channel);
 
@@ -831,6 +834,7 @@ private void HandleDisconnect(NetPeerData peer, NetConnection connection, string
 #endif
             _channels.Remove(connection);
             peer.RemoveChannel(channel);
+            channel.EncryptionChannel?.Complete();
 
             if (IsClient)
             {
@@ -895,9 +899,7 @@ private bool DispatchNetMessage(NetIncomingMessage msg)
                 return true;
             }
 
-            var encryption = IsServer ? channel.Encryption : _clientEncryption;
-
-            encryption?.Decrypt(msg);
+            channel.Encryption?.Decrypt(msg);
 
             var id = msg.ReadByte();
 
@@ -1062,14 +1064,7 @@ public void ServerSendMessage(NetMessage message, INetChannel recipient)
             if (!(recipient is NetChannel channel))
                 throw new ArgumentException($"Not of type {typeof(NetChannel).FullName}", nameof(recipient));
 
-            var peer = channel.Connection.Peer;
-            var packet = BuildMessage(message, peer);
-
-            channel.Encryption?.Encrypt(packet);
-
-            var method = message.DeliveryMethod;
-            peer.SendMessage(packet, channel.Connection, method);
-            LogSend(message, method, packet);
+            CoreSendMessage(channel, message);
         }
 
         private static void LogSend(NetMessage message, NetDeliveryMethod method, NetOutgoingMessage packet)
@@ -1103,13 +1098,10 @@ public void ClientSendMessage(NetMessage message)
             DebugTools.Assert(_netPeers[0].Channels.Count == 1);
 
             var peer = _netPeers[0];
-            var packet = BuildMessage(message, peer.Peer);
-            var method = message.DeliveryMethod;
 
-            _clientEncryption?.Encrypt(packet);
+            var channel = peer.Channels[0];
 
-            peer.Peer.SendMessage(packet, peer.ConnectionsWithChannels[0], method);
-            LogSend(message, method, packet);
+            CoreSendMessage(channel, message);
         }
 
         #endregion NetMessages