semaphore slim, concurrent broadcast

Author: gencebay

src/NetCoreStack.WebSockets/ConnectionManager.cs

@@ -19,12 +19,14 @@ namespace NetCoreStack.WebSockets
 {
     public class ConnectionManager : IConnectionManager
     {
+        private readonly SemaphoreSlim _sendFrameAsyncLock = new SemaphoreSlim(1, 1);
         private readonly InvocatorRegistry _invocatorRegistry;
         private readonly ServerSocketsOptions _options;
         private readonly IHandshakeStateTransport _initState;
         private readonly ILoggerFactory _loggerFactory;
         private readonly IStreamCompressor _compressor;
         private readonly TransportLifetimeManager _lifetimeManager;
+
         public ConcurrentDictionary<string, WebSocketTransport> Connections { get; }
 
         public ConnectionManager(IStreamCompressor compressor,
@@ -43,7 +45,7 @@ public ConnectionManager(IStreamCompressor compressor,
             Connections = new ConcurrentDictionary<string, WebSocketTransport>(StringComparer.OrdinalIgnoreCase);
         }
 
-        private async Task<byte[]> PrepareBytesAsync(byte[] body, IDictionary<string, object> properties)
+        private async Task<byte[]> PrepareFramesBytesAsync(byte[] body, IDictionary<string, object> properties)
         {
             if (body == null)
             {
@@ -80,7 +82,7 @@ private async Task<byte[]> PrepareBytesAsync(byte[] body, IDictionary<string, ob
             return body;
         }
 
-        private async Task SendAsync(WebSocketTransport transport, WebSocketMessageDescriptor descriptor)
+        private Task SendAsync(WebSocketTransport transport, WebSocketMessageDescriptor descriptor)
         {
             if (descriptor == null)
             {
@@ -94,7 +96,7 @@ private async Task SendAsync(WebSocketTransport transport, WebSocketMessageDescr
 
             if (!transport.WebSocket.CloseStatus.HasValue)
             {
-                await transport.WebSocket.SendAsync(descriptor.Segments,
+                return transport.WebSocket.SendAsync(descriptor.Segments,
                    descriptor.MessageType,
                    descriptor.EndOfMessage,
                    CancellationToken.None);
@@ -107,10 +109,12 @@ await transport.WebSocket.SendAsync(descriptor.Segments,
                     Segments = descriptor.Segments,
                     KeepTime = DateTime.Now.AddMinutes(3)
                 });
+
+                return TaskCache.CompletedTask;
             }
         }
 
-        private async Task SendBinaryAsync(WebSocketTransport transport, byte[] chunkedBytes, bool endOfMessage, CancellationToken token)
+        private Task SendBinaryAsync(WebSocketTransport transport, byte[] chunkedBytes, bool endOfMessage)
         {
             if (transport == null)
             {
@@ -121,10 +125,40 @@ private async Task SendBinaryAsync(WebSocketTransport transport, byte[] chunkedB
 
             if (!transport.WebSocket.CloseStatus.HasValue)
             {
-                await transport.WebSocket.SendAsync(segments,
+                return transport.WebSocket.SendAsync(segments,
                    WebSocketMessageType.Binary,
                    endOfMessage,
-                   token);
+                   CancellationToken.None);
+            }
+
+            return TaskCache.CompletedTask;
+        }
+
+        private async Task SendConcurrentBinaryAsync(byte[] bytes)
+        {
+            using (var ms = new MemoryStream(bytes))
+            {
+                using (var br = new BinaryReader(ms))
+                {
+                    byte[] chunkedBytes = null;
+                    do
+                    {
+                        chunkedBytes = br.ReadBytes(ChunkSize);
+                        var endOfMessage = false;
+
+                        if (chunkedBytes.Length < ChunkSize)
+                            endOfMessage = true;
+
+                        foreach (var connection in Connections)
+                        {
+                            await SendBinaryAsync(connection.Value, chunkedBytes, endOfMessage);
+                        }
+
+                        if (endOfMessage)
+                            break;
+
+                    } while (chunkedBytes.Length <= ChunkSize);
+                }
             }
         }
 
@@ -197,10 +231,12 @@ public async Task BroadcastAsync(WebSocketMessageContext context)
                 MessageType = WebSocketMessageType.Text
             };
 
+            _sendFrameAsyncLock.Wait();
             foreach (var connection in Connections)
             {
                 await SendAsync(connection.Value, descriptor);
             }
+            _sendFrameAsyncLock.Release();
         }
 
         public async Task BroadcastBinaryAsync(byte[] inputs, IDictionary<string, object> properties)
@@ -210,39 +246,17 @@ public async Task BroadcastBinaryAsync(byte[] inputs, IDictionary<string, object
                 return;
             }
 
-            var bytes = await PrepareBytesAsync(inputs, properties);
-
-            using (var ms = new MemoryStream(bytes))
-            {
-                using (var br = new BinaryReader(ms))
-                {
-                    byte[] chunkedBytes = null;
-                    do
-                    {
-                        chunkedBytes = br.ReadBytes(ChunkSize);
-                        var endOfMessage = false;
-
-                        if (chunkedBytes.Length < ChunkSize)
-                            endOfMessage = true;
-
-                        foreach (var connection in Connections)
-                        {
-                            await SendBinaryAsync(connection.Value, chunkedBytes, endOfMessage, CancellationToken.None);
-                        }
-
-                        if (endOfMessage)
-                            break;
-
-                    } while (chunkedBytes.Length <= ChunkSize);
-                }
-            }
+            _sendFrameAsyncLock.Wait();
+            var bytes = await PrepareFramesBytesAsync(inputs, properties);
+            await SendConcurrentBinaryAsync(bytes);
+            _sendFrameAsyncLock.Release();
         }
 
-        public async Task SendAsync(string connectionId, WebSocketMessageContext context)
+        public Task SendAsync(string connectionId, WebSocketMessageContext context)
         {
             if (!Connections.Any())
             {
-                return;
+                return TaskCache.CompletedTask;
             }
 
             WebSocketTransport transport = null;
@@ -259,7 +273,7 @@ public async Task SendAsync(string connectionId, WebSocketMessageContext context
                 MessageType = WebSocketMessageType.Text
             };
 
-            await SendAsync(transport, descriptor);
+            return SendAsync(transport, descriptor);
         }
 
         public async Task SendBinaryAsync(string connectionId, byte[] input, IDictionary<string, object> properties)
@@ -275,7 +289,7 @@ public async Task SendBinaryAsync(string connectionId, byte[] input, IDictionary
                 throw new ArgumentOutOfRangeException(nameof(transport));
             }
 
-            byte[] bytes = await PrepareBytesAsync(input, properties);
+            byte[] bytes = await PrepareFramesBytesAsync(input, properties);
 
             using (var ms = new MemoryStream(bytes))
             {

src/NetCoreStack.WebSockets/Internal/TaskCache.cs

@@ -0,0 +1,24 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
+
+using System.Threading.Tasks;
+
+namespace NetCoreStack.WebSockets.Internal
+{
+    public static class TaskCache
+    {
+
+        /// <summary>
+        /// A <see cref="Task"/> that's already completed successfully.
+        /// </summary>
+        /// <remarks>
+        /// We're caching this in a static readonly field to make it more inlinable and avoid the volatile lookup done
+        /// by <c>Task.CompletedTask</c>.
+        /// </remarks>
+#if NET451
+        public static readonly Task CompletedTask = Task.FromResult(0);
+#else
+        public static readonly Task CompletedTask = Task.CompletedTask;
+#endif
+    }
+}

src/NetCoreStack.WebSockets/NetCoreStack.WebSockets.csproj

@@ -15,7 +15,7 @@
   <ItemGroup>
     <PackageReference Include="Microsoft.AspNetCore.WebSockets" Version="1.0.2" />
     <PackageReference Include="Microsoft.Extensions.Logging" Version="1.1.2" />
-    <PackageReference Include="Newtonsoft.Json" Version="10.0.2" />
+    <PackageReference Include="Newtonsoft.Json" Version="10.0.3" />
   </ItemGroup>
 
   <ItemGroup Condition=" '$(TargetFramework)' == 'net451' ">

test/Common.Libs/CacheHelper.cs

@@ -0,0 +1,35 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+
+namespace Common.Libs
+{
+    public static class CacheHelper
+    {
+        public static IDictionary<string, CacheItemDescriptor> CacheKeys
+        {
+            get
+            {
+                var dict = new Dictionary<string, CacheItemDescriptor>()
+                {
+                    [nameof(HizmetEnvanteri)] = new CacheItemDescriptor { Type = typeof(HizmetEnvanteri), Weight = CacheItemWeights.MiddleWeight },
+                    [nameof(KurumBirimDto)] = new CacheItemDescriptor { Type = typeof(KurumBirimDto), Weight = CacheItemWeights.HeavyWeight },
+                    [nameof(KurumBirimPasifDto)] = new CacheItemDescriptor { Type = typeof(KurumBirimPasifDto), Weight = CacheItemWeights.MiddleWeight },
+                };
+
+                var keys = dict.OrderBy(x => x.Key).ToDictionary(t => t.Key, t => t.Value);
+                return keys;
+            }
+        }
+
+        public static CacheItemDescriptor GetDescriptor(string key)
+        {
+            if (CacheKeys.TryGetValue(key, out CacheItemDescriptor descriptor))
+            {
+                return descriptor;
+            }
+
+            throw new ArgumentOutOfRangeException($"Key could not be found: \"{key}\"");
+        }
+    }
+}

test/Common.Libs/CacheItemDescriptor.cs

@@ -0,0 +1,15 @@
+using System;
+using System.Collections.Generic;
+namespace Common.Libs
+{
+    public class CacheItemDescriptor
+    {
+        public Type Type { get; set; }
+        public CacheItemWeights Weight { get; set; }
+
+        public CacheItemDescriptor()
+        {
+            Weight = CacheItemWeights.LightWeight;
+        }
+    }
+}

test/Common.Libs/CacheItemWeights.cs

@@ -0,0 +1,9 @@
+namespace Common.Libs
+{
+    public enum CacheItemWeights
+    {
+        LightWeight = 0,
+        MiddleWeight = 1,
+        HeavyWeight = 2
+    }
+}

test/Common.Libs/HizmetEnvanteri.cs

@@ -0,0 +1,10 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace Common.Libs
+{
+    public class HizmetEnvanteri
+    {
+    }
+}

test/Common.Libs/InMemoryCacheExtensions.cs

@@ -0,0 +1,10 @@
+using Microsoft.Extensions.Caching.Memory;
+using System;
+using static Common.Libs.CacheHelper;
+
+namespace Common.Libs
+{
+    public static class InMemoryCacheExtensions
+    {
+    }
+}

test/Common.Libs/KurumBirimDto.cs

@@ -0,0 +1,10 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace Common.Libs
+{
+    public class KurumBirimDto
+    {
+    }
+}

test/Common.Libs/KurumBirimPasifDto.cs

@@ -0,0 +1,10 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace Common.Libs
+{
+    public class KurumBirimPasifDto
+    {
+    }
+}