Fix #2763: ConnectionMultiplexer.Subscription is not Thread-safe (#2769)

## Issue #2763 ## Solution Simply added a lock around `_handlers` in `ConnectionMultiplexer.Subscription`, like I was suggesting in the issue. ## Unit Test I added one that does exactly what the example code in #2763 was doing & testing for. I used the other tests as template/guide, let me know if something isn't up to spec. --------- Co-authored-by: Nick Craver <[email protected]>
StackExchange · Aug 18, 2024 · fe40d17 · fe40d17
1 parent c0bb4eb
commit fe40d17
Show file tree

Hide file tree

Showing 3 changed files with 60 additions and 3 deletions.
diff --git a/docs/ReleaseNotes.md b/docs/ReleaseNotes.md
@@ -10,6 +10,7 @@ Current package versions:
 
 - Add support for hash field expiration (see [#2715](https://github.com/StackExchange/StackExchange.Redis/issues/2715)) ([#2716 by atakavci](https://github.com/StackExchange/StackExchange.Redis/pull/2716]))
 - Add support for `HSCAN NOVALUES` (see [#2721](https://github.com/StackExchange/StackExchange.Redis/issues/2721)) ([#2722 by atakavci](https://github.com/StackExchange/StackExchange.Redis/pull/2722))
+- Fix [#2763](https://github.com/StackExchange/StackExchange.Redis/issues/2763): Make ConnectionMultiplexer.Subscription thread-safe ([#2769 by Chuck-EP](https://github.com/StackExchange/StackExchange.Redis/pull/2769))
 
 ## 2.8.0
 

diff --git a/src/StackExchange.Redis/RedisSubscriber.cs b/src/StackExchange.Redis/RedisSubscriber.cs
@@ -159,6 +159,7 @@ internal enum SubscriptionAction
         internal sealed class Subscription
         {
             private Action<RedisChannel, RedisValue>? _handlers;
+            private readonly object _handlersLock = new object();
             private ChannelMessageQueue? _queues;
             private ServerEndPoint? CurrentServer;
             public CommandFlags Flags { get; }
@@ -206,7 +207,10 @@ public void Add(Action<RedisChannel, RedisValue>? handler, ChannelMessageQueue?
             {
                 if (handler != null)
                 {
-                    _handlers += handler;
+                    lock (_handlersLock)
+                    {
+                        _handlers += handler;
+                    }
                 }
                 if (queue != null)
                 {
@@ -218,7 +222,10 @@ public bool Remove(Action<RedisChannel, RedisValue>? handler, ChannelMessageQueu
             {
                 if (handler != null)
                 {
-                    _handlers -= handler;
+                    lock (_handlersLock)
+                    {
+                        _handlers -= handler;
+                    }
                 }
                 if (queue != null)
                 {
@@ -236,7 +243,10 @@ public bool Remove(Action<RedisChannel, RedisValue>? handler, ChannelMessageQueu
 
             internal void MarkCompleted()
             {
-                _handlers = null;
+                lock (_handlersLock)
+                {
+                    _handlers = null;
+                }
                 ChannelMessageQueue.MarkAllCompleted(ref _queues);
             }
 

diff --git a/tests/StackExchange.Redis.Tests/Issues/Issue2763Tests.cs b/tests/StackExchange.Redis.Tests/Issues/Issue2763Tests.cs
@@ -0,0 +1,46 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace StackExchange.Redis.Tests.Issues
+{
+    public class Issue2763Tests : TestBase
+    {
+        public Issue2763Tests(ITestOutputHelper output) : base(output) { }
+
+        [Fact]
+        public void Execute()
+        {
+            using var conn = Create();
+            var subscriber = conn.GetSubscriber();
+
+            static void Handler(RedisChannel c, RedisValue v) { }
+
+            const int COUNT = 1000;
+            RedisChannel channel = RedisChannel.Literal("CHANNEL:TEST");
+
+            List<Action> subscribes = new List<Action>(COUNT);
+            for (int i = 0; i < COUNT; i++)
+                subscribes.Add(() => subscriber.Subscribe(channel, Handler));
+            Parallel.ForEach(subscribes, action => action());
+
+            Assert.Equal(COUNT, CountSubscriptionsForChannel(subscriber, channel));
+
+            List<Action> unsubscribes = new List<Action>(COUNT);
+            for (int i = 0; i < COUNT; i++)
+                unsubscribes.Add(() => subscriber.Unsubscribe(channel, Handler));
+            Parallel.ForEach(unsubscribes, action => action());
+
+            Assert.Equal(0, CountSubscriptionsForChannel(subscriber, channel));
+        }
+
+        private static int CountSubscriptionsForChannel(ISubscriber subscriber, RedisChannel channel)
+        {
+            ConnectionMultiplexer connMultiplexer = (ConnectionMultiplexer)subscriber.Multiplexer;
+            connMultiplexer.GetSubscriberCounts(channel, out int handlers, out int _);
+            return handlers;
+        }
+    }
+}