Skip to content

Switch WorkPool of ConsumerWorkService to channels #882

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 2 commits into from
Jul 1, 2020
Merged

Switch WorkPool of ConsumerWorkService to channels #882

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7372cda
Switch WorkPool of ConsumerWorkService to channels
danielmarbach Jun 25, 2020
e38fa26
Add hint to catch block
danielmarbach Jun 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
2 changes: 1 addition & 1 deletion projects/RabbitMQ.Client/client/impl/AsyncConsumerWorkService.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,7 +137,7 @@ static async Task HandleConcurrent(Work work, ModelBase model, SemaphoreSlim lim
}
catch (Exception)
{

// ignored
}
finally
{
Expand Down
74 changes: 31 additions & 43 deletions projects/RabbitMQ.Client/client/impl/ConsumerWorkService.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;

namespace RabbitMQ.Client.Impl
Expand Down Expand Up @@ -61,20 +62,16 @@ internal Task StopWorkAsync(IModel model)

class WorkPool
{
readonly ConcurrentQueue<Action> _actions;
readonly CancellationTokenSource _tokenSource;
readonly CancellationTokenRegistration _tokenRegistration;
volatile TaskCompletionSource<bool> _syncSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
private readonly Channel<Action> _channel;
private readonly int _concurrency;
private Task _worker;
CancellationTokenSource _tokenSource;
private SemaphoreSlim _limiter;

public WorkPool(int concurrency)
{
_concurrency = concurrency;
_actions = new ConcurrentQueue<Action>();
_tokenSource = new CancellationTokenSource();
_tokenRegistration = _tokenSource.Token.Register(() => _syncSource.TrySetCanceled());
_channel = Channel.CreateUnbounded<Action>(new UnboundedChannelOptions { SingleReader = true, SingleWriter = false, AllowSynchronousContinuations = false });
}

public void Start()
Expand All @@ -86,37 +83,27 @@ public void Start()
else
{
_limiter = new SemaphoreSlim(_concurrency);
_tokenSource = new CancellationTokenSource();
_worker = Task.Run(() => LoopWithConcurrency(_tokenSource.Token), CancellationToken.None);
}
}

public void Enqueue(Action action)
{
_actions.Enqueue(action);
_syncSource.TrySetResult(true);
_channel.Writer.TryWrite(action);
}

async Task Loop()
{
while (_tokenSource.IsCancellationRequested == false)
while (await _channel.Reader.WaitToReadAsync().ConfigureAwait(false))
{
try
{
await _syncSource.Task.ConfigureAwait(false);
_syncSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
}
catch (TaskCanceledException)
{
// Swallowing the task cancellation exception for the semaphore in case we are stopping.
}

while (_actions.TryDequeue(out Action action))
while (_channel.Reader.TryRead(out Action work))
{
try
{
action();
work();
}
catch (Exception)
catch(Exception)
{
// ignored
}
Expand All @@ -126,36 +113,37 @@ async Task Loop()

async Task LoopWithConcurrency(CancellationToken cancellationToken)
{
while (_tokenSource.IsCancellationRequested == false)
try
{
try
{
await _syncSource.Task.ConfigureAwait(false);
_syncSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
}
catch (TaskCanceledException)
while (await _channel.Reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false))
{
// Swallowing the task cancellation exception for the semaphore in case we are stopping.
}

while (_actions.TryDequeue(out Action action))
{
// Do a quick synchronous check before we resort to async/await with the state-machine overhead.
if(!_limiter.Wait(0))
while (_channel.Reader.TryRead(out Action action))
{
await _limiter.WaitAsync(cancellationToken).ConfigureAwait(false);
}
// Do a quick synchronous check before we resort to async/await with the state-machine overhead.
if(!_limiter.Wait(0))
{
await _limiter.WaitAsync(cancellationToken).ConfigureAwait(false);
}

_ = OffloadToWorkerThreadPool(action, _limiter);
_ = OffloadToWorkerThreadPool(action, _limiter);
}
}
}
catch (OperationCanceledException)
{
// ignored
}
}

static async Task OffloadToWorkerThreadPool(Action action, SemaphoreSlim limiter)
{
try
{
await Task.Run(() => action());
// like Task.Run but doesn't closure allocate
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎉 I like =)

Copy link
Contributor

@sungam3r sungam3r Jul 11, 2020 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unfortunately, it still allocates delegate. IIRC it is well known issue of Roslyn. If you really do not want allocate delegate each call then define static delegate and use it here.
dotnet/roslyn#39869

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@sungam3r thanks for chiming in. You are welcome to submit a PR that changes the comment (and links to the Roslyn issue) or even switched to use a static delegate if it's a safe thing to do in this context. Thank you.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Comparison according to sharplab

    [Serializable]
    [CompilerGenerated]
    private sealed class <>c
    {
        public static readonly <>c <>9 = new <>c();

        public static Action<object> <>9__0_0;

        internal void <TaskFactory>b__0_0(object state)
        {
            ((Action)state)();
        }
    }

awaiter = Task.Factory.StartNew(<>c.<>9__0_0 ?? (<>c.<>9__0_0 = new Action<object>(<>c.<>9.<TaskFactory>b__0_0)), action, CancellationToken.None, TaskCreationOptions.DenyChildAttach, TaskScheduler.Default).GetAwaiter();

so the Action allocation happens once and is then cached for further execution in a static Action field.

vs Task.Run(() => action())

                    <>c__DisplayClass1_0 <>c__DisplayClass1_ = new <>c__DisplayClass1_0();
                    <>c__DisplayClass1_.action = action;
                    awaiter = Task.Run(new Action(<>c__DisplayClass1_.<TaskRun>b__0)).GetAwaiter();

display class creation for every run plus action allocation for every run

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was wrong. Such lambda usage is OK. I re-read the Roslyn issue and remembered details.

await Task.Factory.StartNew(state =>
{
((Action)state)();
}, action, CancellationToken.None, TaskCreationOptions.DenyChildAttach, TaskScheduler.Default);
}
catch (Exception)
{
Expand All @@ -169,8 +157,8 @@ static async Task OffloadToWorkerThreadPool(Action action, SemaphoreSlim limiter

public Task Stop()
{
_tokenSource.Cancel();
_tokenRegistration.Dispose();
_channel.Writer.Complete();
_tokenSource?.Cancel();
_limiter?.Dispose();
return _worker;
}
Expand Down