Merge pull request #777 from rabbitmq/rabbitmq-dotnet-client-775-772-5.x

lukebakken · web-flow · commit 3b0ec95f81e1 · 2020-03-25T14:43:53.000-07:00
Back-port several master PRs to 5.x branch
diff --git a/projects/client/RabbitMQ.Client/src/client/impl/AsyncConsumerWorkService.cs b/projects/client/RabbitMQ.Client/src/client/impl/AsyncConsumerWorkService.cs
@@ -1,18 +1,19 @@
-﻿using RabbitMQ.Client.Impl;
-
-using System.Collections.Concurrent;
+﻿using System.Collections.Concurrent;
+using System.Reflection;
 using System.Threading;
 using System.Threading.Tasks;
 
+using RabbitMQ.Client.Impl;
+
 namespace RabbitMQ.Client
 {
-    internal class AsyncConsumerWorkService : ConsumerWorkService
+    internal sealed class AsyncConsumerWorkService : ConsumerWorkService
     {
-        private readonly ConcurrentDictionary<IModel, WorkPool> workPools = new ConcurrentDictionary<IModel, WorkPool>();
+        private readonly ConcurrentDictionary<IModel, WorkPool> _workPools = new ConcurrentDictionary<IModel, WorkPool>();
 
         public void Schedule<TWork>(ModelBase model, TWork work) where TWork : Work
         {
-            workPools.GetOrAdd(model, StartNewWorkPool).Enqueue(work);
+            _workPools.GetOrAdd(model, StartNewWorkPool).Enqueue(work);
         }
 
         private WorkPool StartNewWorkPool(IModel model)
@@ -24,69 +25,94 @@ private WorkPool StartNewWorkPool(IModel model)
 
         public void Stop(IModel model)
         {
-            if (workPools.TryRemove(model, out WorkPool workPool))
+            if (_workPools.TryRemove(model, out WorkPool workPool))
             {
                 workPool.Stop();
             }
         }
 
         public void Stop()
         {
-            foreach (IModel model in workPools.Keys)
+            foreach (IModel model in _workPools.Keys)
             {
                 Stop(model);
             }
         }
 
         class WorkPool
         {
-            readonly ConcurrentQueue<Work> workQueue;
-            readonly CancellationTokenSource tokenSource;
-            readonly ModelBase model;
-            readonly SemaphoreSlim semaphore = new SemaphoreSlim(0);
-            private Task task;
+            readonly ConcurrentQueue<Work> _workQueue;
+            readonly CancellationTokenSource _tokenSource;
+            readonly ModelBase _model;
+            readonly CancellationTokenRegistration _tokenRegistration;
+            volatile TaskCompletionSource<bool> _syncSource = TaskCompletionSourceFactory.Create<bool>();
+            private Task _worker;
 
             public WorkPool(ModelBase model)
             {
-                this.model = model;
-                workQueue = new ConcurrentQueue<Work>();
-                tokenSource = new CancellationTokenSource();
+                _model = model;
+                _workQueue = new ConcurrentQueue<Work>();
+                _tokenSource = new CancellationTokenSource();
+                _tokenRegistration = _tokenSource.Token.Register(() => _syncSource.TrySetCanceled());
             }
 
             public void Start()
             {
-                task = Task.Run(Loop, CancellationToken.None);
+                _worker = Task.Run(Loop, CancellationToken.None);
             }
 
             public void Enqueue(Work work)
             {
-                workQueue.Enqueue(work);
-                semaphore.Release();
+                _workQueue.Enqueue(work);
+                _syncSource.TrySetResult(true);
             }
 
             async Task Loop()
             {
-                while (tokenSource.IsCancellationRequested == false)
+                while (_tokenSource.IsCancellationRequested == false)
                 {
                     try
                     {
-                        await semaphore.WaitAsync(tokenSource.Token).ConfigureAwait(false);
+                        await _syncSource.Task.ConfigureAwait(false);
+                        _syncSource = TaskCompletionSourceFactory.Create<bool>();
                     }
                     catch (TaskCanceledException)
                     {
                         // Swallowing the task cancellation in case we are stopping work.
                     }
 
-                    if (!tokenSource.IsCancellationRequested && workQueue.TryDequeue(out Work work))
+                    while (_tokenSource.IsCancellationRequested == false && _workQueue.TryDequeue(out Work work))
                     {
-                        await work.Execute(model).ConfigureAwait(false);
+                        await work.Execute(_model).ConfigureAwait(false);
                     }
                 }
             }
 
             public void Stop()
             {
-                tokenSource.Cancel();
+                _tokenSource.Cancel();
+                _tokenRegistration.Dispose();
+            }
+
+            static class TaskCompletionSourceFactory
+            {
+#if NETFRAMEWORK
+                static readonly FieldInfo StateField = typeof(Task).GetField("m_stateFlags", BindingFlags.NonPublic | BindingFlags.Instance);
+#endif
+        
+                public static TaskCompletionSource<T> Create<T>(TaskCreationOptions options = TaskCreationOptions.None)
+                {
+#if NETFRAMEWORK
+                    //This lovely hack forces the task scheduler to run continuations asynchronously,
+                    //see https://stackoverflow.com/questions/22579206/how-can-i-prevent-synchronous-continuations-on-a-task/22588431#22588431
+                    var tcs = new TaskCompletionSource<T>(options);
+                    const int TASK_STATE_THREAD_WAS_ABORTED = 134217728;
+                    StateField.SetValue(tcs.Task, (int) StateField.GetValue(tcs.Task) | TASK_STATE_THREAD_WAS_ABORTED);
+                    return tcs;
+#else
+                    return new TaskCompletionSource<T>(options | TaskCreationOptions.RunContinuationsAsynchronously);
+#endif
+                }
             }
         }
     }
diff --git a/projects/client/RabbitMQ.Client/src/client/impl/ConsumerWorkService.cs b/projects/client/RabbitMQ.Client/src/client/impl/ConsumerWorkService.cs
@@ -1,5 +1,6 @@
 ﻿using System;
 using System.Collections.Concurrent;
+using System.Reflection;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -39,58 +40,83 @@ public void StopWork()
 
         class WorkPool
         {
-            readonly ConcurrentQueue<Action> actions;
-            readonly SemaphoreSlim semaphore = new SemaphoreSlim(0);
-            readonly CancellationTokenSource tokenSource;
-            private Task worker;
+            readonly ConcurrentQueue<Action> _actions;
+            readonly CancellationTokenSource _tokenSource;
+            readonly CancellationTokenRegistration _tokenRegistration;
+            volatile TaskCompletionSource<bool> _syncSource = TaskCompletionSourceFactory.Create<bool>();
+            private Task _worker;
 
             public WorkPool(IModel model)
             {
-                actions = new ConcurrentQueue<Action>();
-                tokenSource = new CancellationTokenSource();
+                _actions = new ConcurrentQueue<Action>();
+                _tokenSource = new CancellationTokenSource();
+                _tokenRegistration = _tokenSource.Token.Register(() => _syncSource.TrySetCanceled());
             }
 
             public void Start()
             {
-                worker = Task.Run(Loop);
+                _worker = Task.Run(Loop, CancellationToken.None);
             }
 
             public void Enqueue(Action action)
             {
-                actions.Enqueue(action);
-                semaphore.Release();
+                _actions.Enqueue(action);
+                _syncSource.TrySetResult(true);
             }
 
             async Task Loop()
             {
-                while (tokenSource.IsCancellationRequested == false)
+                while (_tokenSource.IsCancellationRequested == false)
                 {
                     try
                     {
-                        await semaphore.WaitAsync(tokenSource.Token).ConfigureAwait(false);
+                        await _syncSource.Task.ConfigureAwait(false);
+                        _syncSource = TaskCompletionSourceFactory.Create<bool>();
                     }
                     catch (TaskCanceledException)
                     {
-                        // Swallowing the task cancellation exception for the semaphore in case we are stopping.
+                        // Swallowing the task cancellation in case we are stopping work.
                     }
 
-                    if (!tokenSource.IsCancellationRequested && actions.TryDequeue(out Action action))
+                    while (_tokenSource.IsCancellationRequested == false && _actions.TryDequeue(out Action action))
                     {
                         try
                         {
                             action();
                         }
                         catch (Exception)
                         {
+                            // ignored
                         }
                     }
-
                 }
             }
 
             public void Stop()
             {
-                tokenSource.Cancel();
+                _tokenSource.Cancel();
+                _tokenRegistration.Dispose();
+            }
+
+            static class TaskCompletionSourceFactory
+            {
+#if NETFRAMEWORK
+                static readonly FieldInfo StateField = typeof(Task).GetField("m_stateFlags", BindingFlags.NonPublic | BindingFlags.Instance);
+#endif
+        
+                public static TaskCompletionSource<T> Create<T>(TaskCreationOptions options = TaskCreationOptions.None)
+                {
+#if NETFRAMEWORK
+                    //This lovely hack forces the task scheduler to run continuations asynchronously,
+                    //see https://stackoverflow.com/questions/22579206/how-can-i-prevent-synchronous-continuations-on-a-task/22588431#22588431
+                    var tcs = new TaskCompletionSource<T>(options);
+                    const int TASK_STATE_THREAD_WAS_ABORTED = 134217728;
+                    StateField.SetValue(tcs.Task, (int) StateField.GetValue(tcs.Task) | TASK_STATE_THREAD_WAS_ABORTED);
+                    return tcs;
+#else
+                    return new TaskCompletionSource<T>(options | TaskCreationOptions.RunContinuationsAsynchronously);
+#endif
+                }
             }
         }
     }
diff --git a/projects/client/Unit/src/unit/APIApproval.Approve.approved.txt b/projects/client/Unit/src/unit/APIApproval.Approve.approved.txt
@@ -3184,7 +3184,6 @@ namespace RabbitMQ.Client.Impl
         void HandleBasicDeliver(RabbitMQ.Client.IBasicConsumer consumer, string consumerTag, ulong deliveryTag, bool redelivered, string exchange, string routingKey, RabbitMQ.Client.IBasicProperties basicProperties, byte[] body);
         void HandleModelShutdown(RabbitMQ.Client.IBasicConsumer consumer, RabbitMQ.Client.ShutdownEventArgs reason);
         void Quiesce();
-        void Shutdown();
         void Shutdown(RabbitMQ.Client.IModel model);
     }
     public interface IFrameHandler

Original file line number	Diff line number	Diff line change
`@@ -1,18 +1,19 @@`
`1`		`-using RabbitMQ.Client.Impl;`
`2`		`-`
`3`		`-using System.Collections.Concurrent;`
	`1`	`+using System.Collections.Concurrent;`
	`2`	`+using System.Reflection;`
`4`	`3`	`using System.Threading;`
`5`	`4`	`using System.Threading.Tasks;`
`6`	`5`
	`6`	`+using RabbitMQ.Client.Impl;`
	`7`	`+`
`7`	`8`	`namespace RabbitMQ.Client`
`8`	`9`	`{`
`9`		`- internal class AsyncConsumerWorkService : ConsumerWorkService`
	`10`	`+ internal sealed class AsyncConsumerWorkService : ConsumerWorkService`
`10`	`11`	`{`
`11`		`- private readonly ConcurrentDictionary<IModel, WorkPool> workPools = new ConcurrentDictionary<IModel, WorkPool>();`
	`12`	`+ private readonly ConcurrentDictionary<IModel, WorkPool> _workPools = new ConcurrentDictionary<IModel, WorkPool>();`
`12`	`13`
`13`	`14`	`public void Schedule<TWork>(ModelBase model, TWork work) where TWork : Work`
`14`	`15`	`{`
`15`		`- workPools.GetOrAdd(model, StartNewWorkPool).Enqueue(work);`
	`16`	`+ _workPools.GetOrAdd(model, StartNewWorkPool).Enqueue(work);`
`16`	`17`	`}`
`17`	`18`
`18`	`19`	`private WorkPool StartNewWorkPool(IModel model)`
`@@ -24,69 +25,94 @@ private WorkPool StartNewWorkPool(IModel model)`
`24`	`25`
`25`	`26`	`public void Stop(IModel model)`
`26`	`27`	`{`
`27`		`- if (workPools.TryRemove(model, out WorkPool workPool))`
	`28`	`+ if (_workPools.TryRemove(model, out WorkPool workPool))`
`28`	`29`	`{`
`29`	`30`	`workPool.Stop();`
`30`	`31`	`}`
`31`	`32`	`}`
`32`	`33`
`33`	`34`	`public void Stop()`
`34`	`35`	`{`
`35`		`- foreach (IModel model in workPools.Keys)`
	`36`	`+ foreach (IModel model in _workPools.Keys)`
`36`	`37`	`{`
`37`	`38`	`Stop(model);`
`38`	`39`	`}`
`39`	`40`	`}`
`40`	`41`
`41`	`42`	`class WorkPool`
`42`	`43`	`{`
`43`		`- readonly ConcurrentQueue<Work> workQueue;`
`44`		`- readonly CancellationTokenSource tokenSource;`
`45`		`- readonly ModelBase model;`
`46`		`- readonly SemaphoreSlim semaphore = new SemaphoreSlim(0);`
`47`		`- private Task task;`
	`44`	`+ readonly ConcurrentQueue<Work> _workQueue;`
	`45`	`+ readonly CancellationTokenSource _tokenSource;`
	`46`	`+ readonly ModelBase _model;`
	`47`	`+ readonly CancellationTokenRegistration _tokenRegistration;`
	`48`	`+ volatile TaskCompletionSource<bool> _syncSource = TaskCompletionSourceFactory.Create<bool>();`
	`49`	`+ private Task _worker;`
`48`	`50`
`49`	`51`	`public WorkPool(ModelBase model)`
`50`	`52`	`{`
`51`		`- this.model = model;`
`52`		`- workQueue = new ConcurrentQueue<Work>();`
`53`		`- tokenSource = new CancellationTokenSource();`
	`53`	`+ _model = model;`
	`54`	`+ _workQueue = new ConcurrentQueue<Work>();`
	`55`	`+ _tokenSource = new CancellationTokenSource();`
	`56`	`+ _tokenRegistration = _tokenSource.Token.Register(() => _syncSource.TrySetCanceled());`
`54`	`57`	`}`
`55`	`58`
`56`	`59`	`public void Start()`
`57`	`60`	`{`
`58`		`- task = Task.Run(Loop, CancellationToken.None);`
	`61`	`+ _worker = Task.Run(Loop, CancellationToken.None);`
`59`	`62`	`}`
`60`	`63`
`61`	`64`	`public void Enqueue(Work work)`
`62`	`65`	`{`
`63`		`- workQueue.Enqueue(work);`
`64`		`- semaphore.Release();`
	`66`	`+ _workQueue.Enqueue(work);`
	`67`	`+ _syncSource.TrySetResult(true);`
`65`	`68`	`}`
`66`	`69`
`67`	`70`	`async Task Loop()`
`68`	`71`	`{`
`69`		`- while (tokenSource.IsCancellationRequested == false)`
	`72`	`+ while (_tokenSource.IsCancellationRequested == false)`
`70`	`73`	`{`
`71`	`74`	`try`
`72`	`75`	`{`
`73`		`- await semaphore.WaitAsync(tokenSource.Token).ConfigureAwait(false);`
	`76`	`+ await _syncSource.Task.ConfigureAwait(false);`
	`77`	`+ _syncSource = TaskCompletionSourceFactory.Create<bool>();`
`74`	`78`	`}`
`75`	`79`	`catch (TaskCanceledException)`
`76`	`80`	`{`
`77`	`81`	`// Swallowing the task cancellation in case we are stopping work.`
`78`	`82`	`}`
`79`	`83`
`80`		`- if (!tokenSource.IsCancellationRequested && workQueue.TryDequeue(out Work work))`
	`84`	`+ while (_tokenSource.IsCancellationRequested == false && _workQueue.TryDequeue(out Work work))`
`81`	`85`	`{`
`82`		`- await work.Execute(model).ConfigureAwait(false);`
	`86`	`+ await work.Execute(_model).ConfigureAwait(false);`
`83`	`87`	`}`
`84`	`88`	`}`
`85`	`89`	`}`
`86`	`90`
`87`	`91`	`public void Stop()`
`88`	`92`	`{`
`89`		`- tokenSource.Cancel();`
	`93`	`+ _tokenSource.Cancel();`
	`94`	`+ _tokenRegistration.Dispose();`
	`95`	`+ }`
	`96`	`+`
	`97`	`+ static class TaskCompletionSourceFactory`
	`98`	`+ {`
	`99`	`+#if NETFRAMEWORK`
	`100`	`+ static readonly FieldInfo StateField = typeof(Task).GetField("m_stateFlags", BindingFlags.NonPublic \| BindingFlags.Instance);`
	`101`	`+#endif`
	`102`	`+`
	`103`	`+ public static TaskCompletionSource<T> Create<T>(TaskCreationOptions options = TaskCreationOptions.None)`
	`104`	`+ {`
	`105`	`+#if NETFRAMEWORK`
	`106`	`+ //This lovely hack forces the task scheduler to run continuations asynchronously,`
	`107`	`+ //see https://stackoverflow.com/questions/22579206/how-can-i-prevent-synchronous-continuations-on-a-task/22588431#22588431`
	`108`	`+ var tcs = new TaskCompletionSource<T>(options);`
	`109`	`+ const int TASK_STATE_THREAD_WAS_ABORTED = 134217728;`
	`110`	`+ StateField.SetValue(tcs.Task, (int) StateField.GetValue(tcs.Task) \| TASK_STATE_THREAD_WAS_ABORTED);`
	`111`	`+ return tcs;`
	`112`	`+#else`
	`113`	`+ return new TaskCompletionSource<T>(options \| TaskCreationOptions.RunContinuationsAsynchronously);`
	`114`	`+#endif`
	`115`	`+ }`
`90`	`116`	`}`
`91`	`117`	`}`
`92`	`118`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,6 @@`
`1`	`1`	`using System;`
`2`	`2`	`using System.Collections.Concurrent;`
	`3`	`+using System.Reflection;`
`3`	`4`	`using System.Threading;`
`4`	`5`	`using System.Threading.Tasks;`
`5`	`6`
`@@ -39,58 +40,83 @@ public void StopWork()`
`39`	`40`
`40`	`41`	`class WorkPool`
`41`	`42`	`{`
`42`		`- readonly ConcurrentQueue<Action> actions;`
`43`		`- readonly SemaphoreSlim semaphore = new SemaphoreSlim(0);`
`44`		`- readonly CancellationTokenSource tokenSource;`
`45`		`- private Task worker;`
	`43`	`+ readonly ConcurrentQueue<Action> _actions;`
	`44`	`+ readonly CancellationTokenSource _tokenSource;`
	`45`	`+ readonly CancellationTokenRegistration _tokenRegistration;`
	`46`	`+ volatile TaskCompletionSource<bool> _syncSource = TaskCompletionSourceFactory.Create<bool>();`
	`47`	`+ private Task _worker;`
`46`	`48`
`47`	`49`	`public WorkPool(IModel model)`
`48`	`50`	`{`
`49`		`- actions = new ConcurrentQueue<Action>();`
`50`		`- tokenSource = new CancellationTokenSource();`
	`51`	`+ _actions = new ConcurrentQueue<Action>();`
	`52`	`+ _tokenSource = new CancellationTokenSource();`
	`53`	`+ _tokenRegistration = _tokenSource.Token.Register(() => _syncSource.TrySetCanceled());`
`51`	`54`	`}`
`52`	`55`
`53`	`56`	`public void Start()`
`54`	`57`	`{`
`55`		`- worker = Task.Run(Loop);`
	`58`	`+ _worker = Task.Run(Loop, CancellationToken.None);`
`56`	`59`	`}`
`57`	`60`
`58`	`61`	`public void Enqueue(Action action)`
`59`	`62`	`{`
`60`		`- actions.Enqueue(action);`
`61`		`- semaphore.Release();`
	`63`	`+ _actions.Enqueue(action);`
	`64`	`+ _syncSource.TrySetResult(true);`
`62`	`65`	`}`
`63`	`66`
`64`	`67`	`async Task Loop()`
`65`	`68`	`{`
`66`		`- while (tokenSource.IsCancellationRequested == false)`
	`69`	`+ while (_tokenSource.IsCancellationRequested == false)`
`67`	`70`	`{`
`68`	`71`	`try`
`69`	`72`	`{`
`70`		`- await semaphore.WaitAsync(tokenSource.Token).ConfigureAwait(false);`
	`73`	`+ await _syncSource.Task.ConfigureAwait(false);`
	`74`	`+ _syncSource = TaskCompletionSourceFactory.Create<bool>();`
`71`	`75`	`}`
`72`	`76`	`catch (TaskCanceledException)`
`73`	`77`	`{`
`74`		`- // Swallowing the task cancellation exception for the semaphore in case we are stopping.`
	`78`	`+ // Swallowing the task cancellation in case we are stopping work.`
`75`	`79`	`}`
`76`	`80`
`77`		`- if (!tokenSource.IsCancellationRequested && actions.TryDequeue(out Action action))`
	`81`	`+ while (_tokenSource.IsCancellationRequested == false && _actions.TryDequeue(out Action action))`
`78`	`82`	`{`
`79`	`83`	`try`
`80`	`84`	`{`
`81`	`85`	`action();`
`82`	`86`	`}`
`83`	`87`	`catch (Exception)`
`84`	`88`	`{`
	`89`	`+ // ignored`
`85`	`90`	`}`
`86`	`91`	`}`
`87`		`-`
`88`	`92`	`}`
`89`	`93`	`}`
`90`	`94`
`91`	`95`	`public void Stop()`
`92`	`96`	`{`
`93`		`- tokenSource.Cancel();`
	`97`	`+ _tokenSource.Cancel();`
	`98`	`+ _tokenRegistration.Dispose();`
	`99`	`+ }`
	`100`	`+`
	`101`	`+ static class TaskCompletionSourceFactory`
	`102`	`+ {`
	`103`	`+#if NETFRAMEWORK`
	`104`	`+ static readonly FieldInfo StateField = typeof(Task).GetField("m_stateFlags", BindingFlags.NonPublic \| BindingFlags.Instance);`
	`105`	`+#endif`
	`106`	`+`
	`107`	`+ public static TaskCompletionSource<T> Create<T>(TaskCreationOptions options = TaskCreationOptions.None)`
	`108`	`+ {`
	`109`	`+#if NETFRAMEWORK`
	`110`	`+ //This lovely hack forces the task scheduler to run continuations asynchronously,`
	`111`	`+ //see https://stackoverflow.com/questions/22579206/how-can-i-prevent-synchronous-continuations-on-a-task/22588431#22588431`
	`112`	`+ var tcs = new TaskCompletionSource<T>(options);`
	`113`	`+ const int TASK_STATE_THREAD_WAS_ABORTED = 134217728;`
	`114`	`+ StateField.SetValue(tcs.Task, (int) StateField.GetValue(tcs.Task) \| TASK_STATE_THREAD_WAS_ABORTED);`
	`115`	`+ return tcs;`
	`116`	`+#else`
	`117`	`+ return new TaskCompletionSource<T>(options \| TaskCreationOptions.RunContinuationsAsynchronously);`
	`118`	`+#endif`
	`119`	`+ }`
`94`	`120`	`}`
`95`	`121`	`}`
`96`	`122`	`}`
Original file line number	Diff line number	Diff line change
`@@ -3184,7 +3184,6 @@ namespace RabbitMQ.Client.Impl`
`3184`	`3184`	`void HandleBasicDeliver(RabbitMQ.Client.IBasicConsumer consumer, string consumerTag, ulong deliveryTag, bool redelivered, string exchange, string routingKey, RabbitMQ.Client.IBasicProperties basicProperties, byte[] body);`
`3185`	`3185`	`void HandleModelShutdown(RabbitMQ.Client.IBasicConsumer consumer, RabbitMQ.Client.ShutdownEventArgs reason);`
`3186`	`3186`	`void Quiesce();`
`3187`		`- void Shutdown();`
`3188`	`3187`	`void Shutdown(RabbitMQ.Client.IModel model);`
`3189`	`3188`	`}`
`3190`	`3189`	`public interface IFrameHandler`