Updated ConsumerWorkService no longer suffers from high lock conention on modern CPUs. This is resolved by applying a new work pool approach in the CWS itself and creation of a custom Thread per model.

Author: Scooletz

projects/client/RabbitMQ.Client/src/client/api/IConnectionFactory.cs

@@ -128,13 +128,6 @@ public interface IConnectionFactory
         /// <returns></returns>
         IConnection CreateConnection(IList<string> hostnames, String clientProvidedName);
 
-        /// <summary>
-        /// Advanced option.
-        ///
-        /// What task scheduler should consumer dispatcher use.
-        /// </summary>
-        TaskScheduler TaskScheduler { get; set; }
-
         /// <summary>
         /// Amount of time protocol handshake operations are allowed to take before
         /// timing out.

projects/client/RabbitMQ.Client/src/client/impl/Connection.cs

@@ -113,12 +113,12 @@ public class Connection : IConnection
 
         public Connection(IConnectionFactory factory, bool insist, IFrameHandler frameHandler, string clientProvidedName = null)
         {
-            this.ClientProvidedName = clientProvidedName;
+            ClientProvidedName = clientProvidedName;
             KnownHosts = null;
             FrameMax = 0;
             m_factory = factory;
             m_frameHandler = frameHandler;
-            this.ConsumerWorkService = new ConsumerWorkService(factory.TaskScheduler);
+            ConsumerWorkService = new ConsumerWorkService();
 
             m_sessionManager = new SessionManager(this, 0);
             m_session0 = new MainSession(this) { Handler = NotifyReceivedCloseOk };

projects/client/RabbitMQ.Client/src/client/impl/ConsumerWorkService.cs

@@ -1,84 +1,117 @@
-using RabbitMQ.Util;
-using System;
-using System.Collections.Generic;
+using System;
+using System.Collections.Concurrent;
 using System.Threading;
-using System.Threading.Tasks;
 
 namespace RabbitMQ.Client
 {
     public class ConsumerWorkService
     {
-        public const int MAX_THUNK_EXECUTION_BATCH_SIZE = 16;
-        private TaskScheduler scheduler;
-        private BatchingWorkPool<IModel, Action> workPool;
+        readonly ConcurrentDictionary<IModel, WorkPool> workPools = new ConcurrentDictionary<IModel, WorkPool>();
 
-        public ConsumerWorkService() :
-            this(TaskScheduler.Default) { }
-
-        public ConsumerWorkService(TaskScheduler scheduler)
+        public void AddWork(IModel model, Action fn)
         {
-            this.scheduler = scheduler;
-            this.workPool = new BatchingWorkPool<IModel, Action>();
+            WorkPool workPool;
+            if (workPools.TryGetValue(model, out workPool))
+            {
+                workPool.Enqueue(fn);
+            }
         }
 
-        public void ExecuteThunk()
+        public void RegisterKey(IModel model)
         {
-            var actions = new List<Action>(MAX_THUNK_EXECUTION_BATCH_SIZE);
-
-            try
+            // the main model can be skipped, as it will not use CWS anyway
+            if (model.ChannelNumber == 0)
             {
-                IModel key = this.workPool.NextWorkBlock(ref actions, MAX_THUNK_EXECUTION_BATCH_SIZE);
-                if (key == null) { return; }
-
-                try
-                {
-                    foreach (var fn in actions)
-                    {
-                        fn();
-                    }
-                }
-                finally
-                {
-                    if (this.workPool.FinishWorkBlock(key))
-                    {
-                        var t = new Task(new Action(ExecuteThunk));
-                        t.Start(this.scheduler);
-                    }
-                }
+                return;
             }
-            catch (Exception)
+
+            var workPool = new WorkPool(model);
+            if (workPools.TryAdd(model, workPool))
             {
-#if NETFX_CORE
-                // To end a task, return
-                return;
-#else   
-                //Thread.CurrentThread.Interrupt(); //TODO: what to do?
-#endif
+                workPool.Start();
             }
         }
 
-        public void AddWork(IModel model, Action fn)
+        public void StopWork(IModel model)
         {
-            if (this.workPool.AddWorkItem(model, fn))
+            if (model.ChannelNumber == 0)
             {
-                var t = new Task(new Action(ExecuteThunk));
-                t.Start(this.scheduler);
+                return;
             }
-        }
 
-        public void RegisterKey(IModel model)
-        {
-            this.workPool.RegisterKey(model);
+            WorkPool workPool;
+            if (workPools.TryRemove(model, out workPool))
+            {
+                workPool.Stop();
+            }
         }
 
-        public void StopWork(IModel model)
+        public void StopWork()
         {
-            this.workPool.UnregisterKey(model);
+            foreach (var model in workPools.Keys)
+            {
+                StopWork(model);
+            }
         }
 
-        public void StopWork()
+
+        class WorkPool
         {
-            this.workPool.UnregisterAllKeys();
+            readonly ConcurrentQueue<Action> actions;
+            readonly AutoResetEvent messageArrived;
+            readonly TimeSpan waitTime;
+            readonly CancellationTokenSource tokenSource;
+            readonly string name;
+
+            public WorkPool(IModel model)
+            {
+                name = model.ToString();
+                actions = new ConcurrentQueue<Action>();
+                messageArrived = new AutoResetEvent(false);
+                waitTime = TimeSpan.FromMilliseconds(100);
+                tokenSource = new CancellationTokenSource();
+            }
+
+            public void Start()
+            {
+                var thread = new Thread(Loop)
+                {
+                    Name = "WorkPool-" + name,
+                    IsBackground = true
+                };
+                thread.Start();
+            }
+
+            public void Enqueue(Action action)
+            {
+                actions.Enqueue(action);
+                messageArrived.Set();
+            }
+
+            void Loop()
+            {
+                while (tokenSource.IsCancellationRequested == false)
+                {
+                    Action action;
+                    while (actions.TryDequeue(out action))
+                    {
+                        try
+                        {
+                            action();
+                        }
+                        catch (Exception)
+                        {
+                        }
+                    }
+
+                    messageArrived.WaitOne(waitTime);
+                }
+            }
+
+            public void Stop()
+            {
+                tokenSource.Cancel();
+            }
         }
     }
 }