Skip to content

Back-port several master PRs to 5.x branch #777

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Mar 25, 2020
Merged

Back-port several master PRs to 5.x branch #777

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
57f50b0
Merge pull request #775 from danielmarbach/fix-async-consumer-loop
michaelklishin Mar 24, 2020
487236e
Merge pull request #772 from danielmarbach/fix-consumer-loop
michaelklishin Mar 25, 2020
c6395dc
Update APIApproval expectation file
michaelklishin Mar 25, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
74 changes: 50 additions & 24 deletions projects/client/RabbitMQ.Client/src/client/impl/AsyncConsumerWorkService.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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)
Expand All @@ -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
}
}
}
}
Expand Down
56 changes: 41 additions & 15 deletions projects/client/RabbitMQ.Client/src/client/impl/ConsumerWorkService.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
using System;
using System.Collections.Concurrent;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;

Expand Down Expand Up @@ -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
}
}
}
}
Expand Down
1 change: 0 additions & 1 deletion projects/client/Unit/src/unit/APIApproval.Approve.approved.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down