Merge pull request #253 from Scooletz/no-lock-contention-at-consumer-work-service

No lock contention at consumer work service

Author: kjnilsson

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

@@ -238,6 +238,7 @@ public TimeSpan ContinuationTimeout
         /// Task scheduler connections created by this factory will use when
         /// dispatching consumer operations, such as message deliveries.
         /// </summary>
+        [Obsolete("This scheduler is no longer used for dispatching consumer operations and will be removed in the next major version.", false)]
         public TaskScheduler TaskScheduler { get; set; } = TaskScheduler.Default;
 
         /// <summary>

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

@@ -133,6 +133,7 @@ public interface IConnectionFactory
         ///
         /// What task scheduler should consumer dispatcher use.
         /// </summary>
+        [Obsolete("This scheduler is no longer used for dispatching consumer operations and will be removed in the next major version.", false)]
         TaskScheduler TaskScheduler { get; set; }
 
         /// <summary>

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

@@ -13,7 +13,6 @@ public ConcurrentConsumerDispatcher(ModelBase model, ConsumerWorkService ws)
         {
             this.model = model;
             this.workService = ws;
-            this.workService.RegisterKey(model);
             this.IsShutdown = false;
         }
 

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,112 @@
-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)
-        {
-            this.scheduler = scheduler;
-            this.workPool = new BatchingWorkPool<IModel, Action>();
-        }
-
-        public void ExecuteThunk()
+        public void AddWork(IModel model, Action fn)
         {
-            var actions = new List<Action>(MAX_THUNK_EXECUTION_BATCH_SIZE);
+            // two step approach is taken, as TryGetValue does not aquire locks
+            // if this fails, GetOrAdd is called, which takes a lock
 
-            try
+            WorkPool workPool;
+            if (workPools.TryGetValue(model, out workPool) == false)
             {
-                IModel key = this.workPool.NextWorkBlock(ref actions, MAX_THUNK_EXECUTION_BATCH_SIZE);
-                if (key == null) { return; }
+                var newWorkPool = new WorkPool(model);
+                workPool = workPools.GetOrAdd(model, newWorkPool);
 
-                try
-                {
-                    foreach (var fn in actions)
-                    {
-                        fn();
-                    }
-                }
-                finally
+                // start if it's only the workpool that has been just created
+                if (newWorkPool == workPool)
                 {
-                    if (this.workPool.FinishWorkBlock(key))
-                    {
-                        var t = new Task(new Action(ExecuteThunk));
-                        t.Start(this.scheduler);
-                    }
+                    newWorkPool.Start();
                 }
             }
-            catch (Exception)
-            {
-#if NETFX_CORE
-                // To end a task, return
-                return;
-#else   
-                //Thread.CurrentThread.Interrupt(); //TODO: what to do?
-#endif
-            }
+
+            workPool.Enqueue(fn);
         }
 
-        public void AddWork(IModel model, Action fn)
+        public void StopWork(IModel model)
         {
-            if (this.workPool.AddWorkItem(model, fn))
+            WorkPool workPool;
+            if (workPools.TryRemove(model, out workPool))
             {
-                var t = new Task(new Action(ExecuteThunk));
-                t.Start(this.scheduler);
+                workPool.Stop();
             }
         }
 
-        public void RegisterKey(IModel model)
+        public void StopWork()
         {
-            this.workPool.RegisterKey(model);
+            foreach (var model in workPools.Keys)
+            {
+                StopWork(model);
+            }
         }
 
-        public void StopWork(IModel model)
+        class WorkPool
         {
-            this.workPool.UnregisterKey(model);
-        }
+            readonly ConcurrentQueue<Action> actions;
+            readonly AutoResetEvent messageArrived;
+            readonly TimeSpan waitTime;
+            readonly CancellationTokenSource tokenSource;
+            readonly string name;
 
-        public void StopWork()
-        {
-            this.workPool.UnregisterAllKeys();
+            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()
+            {
+#if NETFX_CORE
+                Task.Factory.StartNew(Loop, TaskCreationOptions.LongRunning);
+#else
+                var thread = new Thread(Loop)
+                {
+                    Name = "WorkPool-" + name,
+                    IsBackground = true
+                };
+                thread.Start();
+#endif
+            }
+
+            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();
+            }
         }
     }
 }