add option to use the thread-pool for scheduling

Author: mgravell

src/StackExchange.Redis/PhysicalBridge.cs

@@ -2,6 +2,7 @@
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.IO;
+using System.IO.Pipelines;
 using System.Runtime.CompilerServices;
 using System.Text;
 using System.Threading;
@@ -768,7 +769,7 @@ private bool PushToBacklog(Message message, bool onlyIfExists)
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         private void StartBacklogProcessor()
         {
-            var sched = Multiplexer.SocketManager?.SchedulerPool ?? DedicatedThreadPoolPipeScheduler.Default;
+            var sched = Multiplexer.SocketManager?.Scheduler ?? PipeScheduler.ThreadPool;
 #if DEBUG
             _backlogProcessorRequestedTime = Environment.TickCount;
 #endif

src/StackExchange.Redis/SocketManager.cs

@@ -24,27 +24,61 @@ public sealed partial class SocketManager : IDisposable
         /// </summary>
         /// <param name="name">The name for this <see cref="SocketManager"/>.</param>
         public SocketManager(string name)
-            : this(name, DEFAULT_WORKERS, false) { }
+            : this(name, DEFAULT_WORKERS, SocketManagerOptions.None) { }
 
         /// <summary>
         /// Creates a new <see cref="SocketManager"/> instance
         /// </summary>
         /// <param name="name">The name for this <see cref="SocketManager"/>.</param>
         /// <param name="useHighPrioritySocketThreads">Whether this <see cref="SocketManager"/> should use high priority sockets.</param>
         public SocketManager(string name, bool useHighPrioritySocketThreads)
-            : this(name, DEFAULT_WORKERS, useHighPrioritySocketThreads) { }
+            : this(name, DEFAULT_WORKERS, UseHighPrioritySocketThreads(useHighPrioritySocketThreads)) { }
 
         /// <summary>
         /// Creates a new (optionally named) <see cref="SocketManager"/> instance
         /// </summary>
         /// <param name="name">The name for this <see cref="SocketManager"/>.</param>
         /// <param name="workerCount">the number of dedicated workers for this <see cref="SocketManager"/>.</param>
         /// <param name="useHighPrioritySocketThreads">Whether this <see cref="SocketManager"/> should use high priority sockets.</param>
-        public SocketManager(string name = null, int workerCount = 0, bool useHighPrioritySocketThreads = false)
+        public SocketManager(string name, int workerCount, bool useHighPrioritySocketThreads)
+            : this(name, workerCount, UseHighPrioritySocketThreads(useHighPrioritySocketThreads)) {}
+
+        private static SocketManagerOptions UseHighPrioritySocketThreads(bool value)
+            => value ? SocketManagerOptions.UseHighPrioritySocketThreads : SocketManagerOptions.None;
+
+        /// <summary>
+        /// Additional options for configuring the socket manager
+        /// </summary>
+        [Flags]
+        public enum SocketManagerOptions
+        {
+            /// <summary>
+            /// No additional options
+            /// </summary>
+            None = 0,
+            /// <summary>
+            /// Whether the <see cref="SocketManager"/> should use high priority sockets.
+            /// </summary>
+            UseHighPrioritySocketThreads = 1 << 0,
+            /// <summary>
+            /// Use the regular thread-pool for all scheduling
+            /// </summary>
+            UseThreadPool = 1 << 1,
+        }
+
+        /// <summary>
+        /// Creates a new (optionally named) <see cref="SocketManager"/> instance
+        /// </summary>
+        /// <param name="name">The name for this <see cref="SocketManager"/>.</param>
+        /// <param name="workerCount">the number of dedicated workers for this <see cref="SocketManager"/>.</param>
+        /// <param name="options"></param>
+        public SocketManager(string name = null, int workerCount = 0, SocketManagerOptions options = SocketManagerOptions.None)
         {
             if (string.IsNullOrWhiteSpace(name)) name = GetType().Name;
             if (workerCount <= 0) workerCount = DEFAULT_WORKERS;
             Name = name;
+            bool useHighPrioritySocketThreads = (options & SocketManagerOptions.UseHighPrioritySocketThreads) != 0,
+                useThreadPool = (options & SocketManagerOptions.UseThreadPool) != 0;
 
             const long Receive_PauseWriterThreshold = 4L * 1024 * 1024 * 1024; // receive: let's give it up to 4GiB of buffer for now
             const long Receive_ResumeWriterThreshold = 3L * 1024 * 1024 * 1024; // (large replies get crazy big)
@@ -58,45 +92,70 @@ public SocketManager(string name = null, int workerCount = 0, bool useHighPriori
                 Send_PauseWriterThreshold / 2,
                 defaultPipeOptions.ResumeWriterThreshold);
 
-            _schedulerPool = new DedicatedThreadPoolPipeScheduler(name + ":IO",
-                workerCount: workerCount,
-                priority: useHighPrioritySocketThreads ? ThreadPriority.AboveNormal : ThreadPriority.Normal);
+            Scheduler = PipeScheduler.ThreadPool;
+            if (!useThreadPool)
+            {
+                Scheduler = new DedicatedThreadPoolPipeScheduler(name + ":IO",
+                    workerCount: workerCount,
+                    priority: useHighPrioritySocketThreads ? ThreadPriority.AboveNormal : ThreadPriority.Normal);
+            }
             SendPipeOptions = new PipeOptions(
                 pool: defaultPipeOptions.Pool,
-                readerScheduler: _schedulerPool,
-                writerScheduler: _schedulerPool,
+                readerScheduler: Scheduler,
+                writerScheduler: Scheduler,
                 pauseWriterThreshold: Send_PauseWriterThreshold,
                 resumeWriterThreshold: Send_ResumeWriterThreshold,
                 minimumSegmentSize: Math.Max(defaultPipeOptions.MinimumSegmentSize, MINIMUM_SEGMENT_SIZE),
                 useSynchronizationContext: false);
             ReceivePipeOptions = new PipeOptions(
                 pool: defaultPipeOptions.Pool,
-                readerScheduler: _schedulerPool,
-                writerScheduler: _schedulerPool,
+                readerScheduler: Scheduler,
+                writerScheduler: Scheduler,
                 pauseWriterThreshold: Receive_PauseWriterThreshold,
                 resumeWriterThreshold: Receive_ResumeWriterThreshold,
                 minimumSegmentSize: Math.Max(defaultPipeOptions.MinimumSegmentSize, MINIMUM_SEGMENT_SIZE),
                 useSynchronizationContext: false);
         }
 
         /// <summary>
-        /// Default / shared socket manager
+        /// Default / shared socket manager using a dedicated thread-pool
         /// </summary>
         public static SocketManager Shared
         {
             get
             {
-                var shared = _shared;
-                if (shared != null) return _shared;
+                var shared = s_shared;
+                if (shared != null) return shared;
                 try
                 {
                     // note: we'll allow a higher max thread count on the shared one
                     shared = new SocketManager("DefaultSocketManager", DEFAULT_WORKERS * 2, false);
-                    if (Interlocked.CompareExchange(ref _shared, shared, null) == null)
+                    if (Interlocked.CompareExchange(ref s_shared, shared, null) == null)
+                        shared = null;
+                }
+                finally { shared?.Dispose(); }
+                return Volatile.Read(ref s_shared);
+            }
+        }
+
+        /// <summary>
+        /// Shared socket manager using the main thread-pool
+        /// </summary>
+        public static SocketManager ThreadPool
+        {
+            get
+            {
+                var shared = s_threadPool;
+                if (shared != null) return shared;
+                try
+                {
+                    // note: we'll allow a higher max thread count on the shared one
+                    shared = new SocketManager("ThreadPoolSocketManager", options: SocketManagerOptions.UseThreadPool);
+                    if (Interlocked.CompareExchange(ref s_threadPool, shared, null) == null)
                         shared = null;
                 }
                 finally { shared?.Dispose(); }
-                return Volatile.Read(ref _shared);
+                return Volatile.Read(ref s_threadPool);
             }
         }
 
@@ -105,18 +164,19 @@ public static SocketManager Shared
         public override string ToString()
         {
             var scheduler = SchedulerPool;
-
-            return $"scheduler - queue: {scheduler?.TotalServicedByQueue}, pool: {scheduler?.TotalServicedByPool}";
+            if (scheduler == null) return Name;
+            return $"{Name} - queue: {scheduler?.TotalServicedByQueue}, pool: {scheduler?.TotalServicedByPool}";
         }
 
-        private static SocketManager _shared;
+        private static SocketManager s_shared, s_threadPool;
 
         private const int DEFAULT_WORKERS = 5, MINIMUM_SEGMENT_SIZE = 8 * 1024;
 
-        private DedicatedThreadPoolPipeScheduler _schedulerPool;
         internal readonly PipeOptions SendPipeOptions, ReceivePipeOptions;
 
-        internal DedicatedThreadPoolPipeScheduler SchedulerPool => _schedulerPool;
+        internal PipeScheduler Scheduler { get; private set; }
+
+        internal DedicatedThreadPoolPipeScheduler SchedulerPool => Scheduler as DedicatedThreadPoolPipeScheduler;
 
         private enum CallbackOperation
         {
@@ -134,8 +194,9 @@ private void Dispose(bool disposing)
             // note: the scheduler *can't* be collected by itself - there will
             // be threads, and those threads will be rooting the DedicatedThreadPool;
             // but: we can lend a hand! We need to do this even in the finalizer
-            try { _schedulerPool?.Dispose(); } catch { }
-            _schedulerPool = null;
+            var tmp = SchedulerPool;
+            Scheduler = PipeScheduler.ThreadPool;
+            try { tmp?.Dispose(); } catch { }
             if (disposing)
             {
                 GC.SuppressFinalize(this);
@@ -167,7 +228,7 @@ internal static Socket CreateSocket(EndPoint endpoint)
 
         internal string GetState()
         {
-            var s = _schedulerPool;
+            var s = SchedulerPool;
             return s == null ? null : $"{s.AvailableCount} of {s.WorkerCount} available";
         }
     }

tests/StackExchange.Redis.Tests/Config.cs

@@ -1,10 +1,12 @@
 using System;
 using System.IO;
+using System.IO.Pipelines;
 using System.Linq;
 using System.Net;
 using System.Net.Sockets;
 using System.Security.Authentication;
 using System.Threading.Tasks;
+using Pipelines.Sockets.Unofficial;
 using Xunit;
 using Xunit.Abstractions;
 
@@ -403,5 +405,28 @@ public void EndpointIteratorIsReliableOverChanges()
             Assert.Equal(8001, ((IPEndPoint)iter.Current).Port);
             Assert.False(iter.MoveNext());
         }
+
+        [Fact]
+        public void ThreadPoolManagerIsDetected()
+        {
+            var config = new ConfigurationOptions
+            {
+                EndPoints = { { IPAddress.Loopback, 6379 } },
+                SocketManager = SocketManager.ThreadPool
+            };
+            using var muxer = ConnectionMultiplexer.Connect(config);
+            Assert.Same(PipeScheduler.ThreadPool, muxer.SocketManager.Scheduler);
+        }
+
+        [Fact]
+        public void DefaultThreadPoolManagerIsDetected()
+        {
+            var config = new ConfigurationOptions
+            {
+                EndPoints = { { IPAddress.Loopback, 6379 } },
+            };
+            using var muxer = ConnectionMultiplexer.Connect(config);
+            Assert.Same(SocketManager.Shared.Scheduler, muxer.SocketManager.Scheduler);
+        }
     }
 }