perf improvements

Author: endel

integrations/Unity/colyseus.nativewebsocket.asmdef

@@ -0,0 +1,3 @@
+{
+	"name": "colyseus.nativewebsocket"
+}

integrations/Unity/endel.nativewebsocket.asmdef

@@ -28,6 +28,10 @@ public class WebSocket : IWebSocket
         private List<Action> m_DispatchQueue = new List<Action>();
         private readonly object EventQueueLock = new object();
 
+        private List<byte[]> m_MessageQueue = new List<byte[]>();
+        private List<byte[]> m_MessageDispatchQueue = new List<byte[]>();
+        private readonly object MessageQueueLock = new object();
+
         private readonly object OutgoingMessageLock = new object();
         private bool isSending = false;
         private Queue<ArraySegment<byte>> sendBytesQueue = new Queue<ArraySegment<byte>>();
@@ -84,12 +88,46 @@ private void EnqueueEvent(Action action)
             }
         }
 
+        private void EnqueueMessage(byte[] data)
+        {
+            if (_syncContext != null)
+            {
+                _syncContext.Post(_ => OnMessage?.Invoke(data), null);
+            }
+            else
+            {
+                lock (MessageQueueLock)
+                {
+                    m_MessageQueue.Add(data);
+                }
+            }
+        }
+
         /// <summary>
         /// Dispatches queued events when no SynchronizationContext is available.
         /// Not needed when a SynchronizationContext is present (Unity, Godot, MonoGame with WebSocketGameComponent).
         /// </summary>
         public void DispatchMessageQueue()
         {
+            // Hot path: dispatch messages without closure overhead
+            if (m_MessageQueue.Count > 0)
+            {
+                lock (MessageQueueLock)
+                {
+                    var tmp = m_MessageDispatchQueue;
+                    m_MessageDispatchQueue = m_MessageQueue;
+                    m_MessageQueue = tmp;
+                }
+
+                for (int i = 0; i < m_MessageDispatchQueue.Count; i++)
+                {
+                    OnMessage?.Invoke(m_MessageDispatchQueue[i]);
+                }
+
+                m_MessageDispatchQueue.Clear();
+            }
+
+            // Rare events: OnOpen, OnError, OnClose
             if (m_EventQueue.Count == 0) return;
 
             lock (EventQueueLock)
@@ -179,75 +217,116 @@ public WebSocketState State
 
         public Task Send(byte[] bytes)
         {
-            return SendMessage(sendBytesQueue, WebSocketMessageType.Binary, new ArraySegment<byte>(bytes));
+            if (bytes.Length == 0 || State != WebSocketState.Open)
+                return Task.CompletedTask;
+
+            var segment = new ArraySegment<byte>(bytes);
+
+            lock (OutgoingMessageLock)
+            {
+                if (isSending)
+                {
+                    sendBytesQueue.Enqueue(segment);
+                    return Task.CompletedTask;
+                }
+                isSending = true;
+            }
+
+            return SendAndDrainAsync(sendBytesQueue, WebSocketMessageType.Binary, segment);
         }
 
         public Task SendText(string message)
         {
+            if (State != WebSocketState.Open)
+                return Task.CompletedTask;
+
             var encoded = Encoding.UTF8.GetBytes(message);
-            return SendMessage(sendTextQueue, WebSocketMessageType.Text, new ArraySegment<byte>(encoded, 0, encoded.Length));
+            if (encoded.Length == 0)
+                return Task.CompletedTask;
+
+            var segment = new ArraySegment<byte>(encoded, 0, encoded.Length);
+
+            lock (OutgoingMessageLock)
+            {
+                if (isSending)
+                {
+                    sendTextQueue.Enqueue(segment);
+                    return Task.CompletedTask;
+                }
+                isSending = true;
+            }
+
+            return SendAndDrainAsync(sendTextQueue, WebSocketMessageType.Text, segment);
         }
 
-        private async Task SendMessage(Queue<ArraySegment<byte>> queue, WebSocketMessageType messageType, ArraySegment<byte> buffer)
+        private Task SendAndDrainAsync(Queue<ArraySegment<byte>> queue, WebSocketMessageType messageType, ArraySegment<byte> buffer)
         {
-            if (buffer.Count == 0 || State != WebSocketState.Open)
+            // Try synchronous fast path: avoid async state machine when SendAsync completes inline
+            var sendTask = m_Socket.SendAsync(buffer, messageType, true, m_CancellationToken);
+            if (sendTask.Status == TaskStatus.RanToCompletion)
             {
-                return;
+                return DrainQueueSync(queue, messageType);
             }
 
-            bool sending;
+            return AwaitAndDrainAsync(sendTask, queue, messageType);
+        }
 
-            lock (OutgoingMessageLock)
+        private Task DrainQueueSync(Queue<ArraySegment<byte>> queue, WebSocketMessageType messageType)
+        {
+            while (true)
             {
-                sending = isSending;
+                ArraySegment<byte> next;
+                lock (OutgoingMessageLock)
+                {
+                    if (queue.Count == 0)
+                    {
+                        isSending = false;
+                        return Task.CompletedTask;
+                    }
+                    next = queue.Dequeue();
+                }
 
-                if (!isSending)
+                var sendTask = m_Socket.SendAsync(next, messageType, true, m_CancellationToken);
+                if (sendTask.Status != TaskStatus.RanToCompletion)
                 {
-                    isSending = true;
+                    // This send went async — fall through to async drain
+                    return AwaitAndDrainAsync(sendTask, queue, messageType);
                 }
             }
+        }
 
-            if (!sending)
+        private async Task AwaitAndDrainAsync(Task pendingSend, Queue<ArraySegment<byte>> queue, WebSocketMessageType messageType)
+        {
+            try
             {
-                try
-                {
-                    await m_Socket.SendAsync(buffer, messageType, true, m_CancellationToken).ConfigureAwait(false);
+                await pendingSend.ConfigureAwait(false);
 
-                    // Drain the queue iteratively instead of recursively
-                    while (true)
-                    {
-                        ArraySegment<byte> next;
-                        lock (OutgoingMessageLock)
-                        {
-                            if (queue.Count == 0)
-                                break;
-                            next = queue.Dequeue();
-                        }
-
-                        await m_Socket.SendAsync(next, messageType, true, m_CancellationToken).ConfigureAwait(false);
-                    }
-                }
-                finally
+                while (true)
                 {
+                    ArraySegment<byte> next;
                     lock (OutgoingMessageLock)
                     {
-                        isSending = false;
+                        if (queue.Count == 0)
+                            break;
+                        next = queue.Dequeue();
                     }
+
+                    await m_Socket.SendAsync(next, messageType, true, m_CancellationToken).ConfigureAwait(false);
                 }
             }
-            else
+            finally
             {
                 lock (OutgoingMessageLock)
                 {
-                    queue.Enqueue(buffer);
+                    isSending = false;
                 }
             }
         }
 
         public async Task Receive()
         {
             WebSocketCloseCode closeCode = WebSocketCloseCode.Abnormal;
-            ArraySegment<byte> buffer = new ArraySegment<byte>(new byte[8192]);
+            ArraySegment<byte> buffer = new ArraySegment<byte>(new byte[65536]);
             try
             {
                 while (m_Socket.State == System.Net.WebSockets.WebSocketState.Open)
@@ -266,7 +345,7 @@ public async Task Receive()
                         // Fast path: single-frame message, avoid MemoryStream
                         var data = new byte[result.Count];
                         Buffer.BlockCopy(buffer.Array, buffer.Offset, data, 0, result.Count);
-                        EnqueueEvent(() => OnMessage?.Invoke(data));
+                        EnqueueMessage(data);
                     }
                     else
                     {
@@ -283,7 +362,7 @@ public async Task Receive()
                             while (!result.EndOfMessage);
 
                             var data = ms.ToArray();
-                            EnqueueEvent(() => OnMessage?.Invoke(data));
+                            EnqueueMessage(data);
                         }
                     }
                 }