Skip to content

Commit 97a9af8

Browse files
soennekersoenneker
committed
Improves performance/allocation
1 parent 55bbfa1 commit 97a9af8

File tree

5 files changed

+111
-109
lines changed

5 files changed

+111
-109
lines changed

‎src/Abstract/IBackgroundQueue.cs‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,5 +42,10 @@ public interface IBackgroundQueue
4242
/// <returns>A <see cref="ValueTask"/> containing a function that represents the dequeued work item, which accepts a <see cref="CancellationToken"/> and returns a <see cref="Task"/>.</returns>
4343
ValueTask<Func<CancellationToken, Task>> DequeueTask(CancellationToken cancellationToken = default);
4444

45+
/// <summary>
46+
/// This is really wait until both queues are empty and their work is done, not just are the queues empty.
47+
/// </summary>
48+
/// <param name="cancellationToken"></param>
49+
/// <returns></returns>
4550
ValueTask WaitUntilEmpty(CancellationToken cancellationToken = default);
4651
}

‎src/BackgroundQueue.cs‎

Lines changed: 60 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@ public sealed class BackgroundQueue : IBackgroundQueue
2525
private readonly ILogger<BackgroundQueue> _logger;
2626
private readonly IQueueInformationUtil _queueInformationUtil;
2727

28+
private long _lastWarnTicks;
29+
2830
private readonly bool _log;
2931

3032
public BackgroundQueue(IConfiguration config, ILogger<BackgroundQueue> logger, IQueueInformationUtil queueInformationUtil)
@@ -51,7 +53,10 @@ public BackgroundQueue(IConfiguration config, ILogger<BackgroundQueue> logger, I
5153

5254
var options = new BoundedChannelOptions(_queueLimit)
5355
{
54-
FullMode = BoundedChannelFullMode.Wait
56+
FullMode = BoundedChannelFullMode.Wait,
57+
SingleReader = true,
58+
SingleWriter = false,
59+
AllowSynchronousContinuations = false
5560
};
5661

5762
_valueTaskChannel = Channel.CreateBounded<Func<CancellationToken, ValueTask>>(options);
@@ -62,34 +67,58 @@ public async ValueTask QueueValueTask(Func<CancellationToken, ValueTask> workIte
6267
{
6368
// TODO: need to redo this, we're going to get too many warnings
6469

65-
int count = await _queueInformationUtil.IncrementValueTaskCounter(cancellationToken).NoSync();
70+
int count = await _queueInformationUtil.IncrementValueTaskCounter(cancellationToken)
71+
.NoSync();
6672

67-
if (count > _queueWarning)
73+
try
74+
{
75+
await _valueTaskChannel.Writer.WriteAsync(workItem, cancellationToken)
76+
.NoSync();
77+
}
78+
catch
6879
{
69-
_logger.LogWarning("ValueTask queue length ({length}) is currently greater than the warning ({_queueWarning}), and will wait after hitting limit ({_queueLimit})", count,
70-
_queueWarning, _queueLimit);
80+
await _queueInformationUtil.DecrementValueTaskCounter(CancellationToken.None)
81+
.NoSync();
82+
throw;
7183
}
7284

73-
if (_log)
74-
_logger.LogDebug("Queuing ValueTask: {name}", workItem.ToString());
85+
if (count > _queueWarning && ShouldWarn())
86+
{
87+
_logger.LogWarning(
88+
"ValueTask queue length ({length}) is currently greater than the warning ({_queueWarning}), and will wait after hitting limit ({_queueLimit})",
89+
count, _queueWarning, _queueLimit);
90+
}
7591

76-
await _valueTaskChannel.Writer.WriteAsync(workItem, cancellationToken).NoSync();
92+
if (_log)
93+
_logger.LogDebug("Queuing ValueTask: {name}", workItem.Method.GetSignature());
7794
}
7895

7996
public async ValueTask QueueTask(Func<CancellationToken, Task> workItem, CancellationToken cancellationToken = default)
8097
{
81-
int count = await _queueInformationUtil.IncrementTaskCounter(cancellationToken).NoSync();
98+
int count = await _queueInformationUtil.IncrementTaskCounter(cancellationToken)
99+
.NoSync();
82100

83-
if (count > _queueWarning)
101+
try
84102
{
85-
_logger.LogWarning("ValueTask queue length ({length}) is currently greater than the warning ({_queueWarning}), and will wait after hitting limit ({_queueLimit})", count,
86-
_queueWarning, _queueLimit);
103+
await _taskChannel.Writer.WriteAsync(workItem, cancellationToken)
104+
.NoSync();
105+
}
106+
catch
107+
{
108+
await _queueInformationUtil.DecrementTaskCounter(CancellationToken.None)
109+
.NoSync();
110+
throw;
111+
}
112+
113+
if (count > _queueWarning && ShouldWarn())
114+
{
115+
_logger.LogWarning(
116+
"Task queue length ({length}) is currently greater than the warning ({_queueWarning}), and will wait after hitting limit ({_queueLimit})",
117+
count, _queueWarning, _queueLimit);
87118
}
88119

89120
if (_log)
90121
_logger.LogDebug("Queuing Task: {name}", workItem.Method.GetSignature());
91-
92-
await _taskChannel.Writer.WriteAsync(workItem, cancellationToken).NoSync();
93122
}
94123

95124
public ValueTask<Func<CancellationToken, ValueTask>> DequeueValueTask(CancellationToken cancellationToken = default)
@@ -110,7 +139,8 @@ public async ValueTask WaitUntilEmpty(CancellationToken cancellationToken = defa
110139

111140
do
112141
{
113-
isProcessing = await _queueInformationUtil.IsProcessing(cancellationToken).ConfigureAwait(false);
142+
isProcessing = await _queueInformationUtil.IsProcessing(cancellationToken)
143+
.ConfigureAwait(false);
114144

115145
if (isProcessing)
116146
{
@@ -119,12 +149,25 @@ public async ValueTask WaitUntilEmpty(CancellationToken cancellationToken = defa
119149
_logger.LogDebug("Delaying for {ms}ms (Background queue emptying)...", delayMs);
120150
}
121151

122-
await DelayUtil.Delay(delayMs, null, cancellationToken).NoSync();
152+
await DelayUtil.Delay(delayMs, null, cancellationToken)
153+
.NoSync();
123154
}
124155
else
125156
{
126157
_logger.LogDebug("Background queue is empty; continuing");
127158
}
128-
} while (isProcessing);
159+
}
160+
while (isProcessing);
161+
}
162+
163+
private bool ShouldWarn()
164+
{
165+
long now = Environment.TickCount64;
166+
long last = Volatile.Read(ref _lastWarnTicks);
167+
168+
if (now - last < 10_000) // 10s
169+
return false;
170+
171+
return Interlocked.CompareExchange(ref _lastWarnTicks, now, last) == last;
129172
}
130173
}

‎src/QueueInformationUtil.cs‎

Lines changed: 31 additions & 85 deletions
Original file line numberDiff line numberDiff line change
@@ -1,123 +1,69 @@
11
using System.Threading;
22
using System.Threading.Tasks;
33
using Microsoft.Extensions.Configuration;
4-
using Nito.AsyncEx;
4+
using Soenneker.Atomics.ValueInts;
55
using Soenneker.Utils.BackgroundQueue.Abstract;
66

77
namespace Soenneker.Utils.BackgroundQueue;
88

9-
///<inheritdoc cref="IQueueInformationUtil"/>
9+
/// <inheritdoc cref="IQueueInformationUtil"/>
1010
public sealed class QueueInformationUtil : IQueueInformationUtil
1111
{
12-
private readonly bool _lockCounts;
12+
private readonly bool _trackCounts;
1313

14-
private readonly AsyncLock? _asyncLock;
15-
16-
private int _taskCount;
17-
private int _valueTaskCount;
14+
private ValueAtomicInt _taskCount;
15+
private ValueAtomicInt _valueTaskCount;
1816

1917
public QueueInformationUtil(IConfiguration config)
2018
{
21-
_lockCounts = config.GetValue<bool>("Background:LockCounts");
22-
23-
if (_lockCounts)
24-
_asyncLock = new AsyncLock();
19+
_trackCounts = config.GetValue<bool>("Background:LockCounts");
2520
}
2621

27-
public async ValueTask<(int TaskLength, int ValueTaskLength)> GetCountsOfProcessing(CancellationToken cancellationToken = default)
22+
public ValueTask<(int TaskLength, int ValueTaskLength)> GetCountsOfProcessing(CancellationToken cancellationToken = default)
2823
{
29-
if (!_lockCounts)
30-
return (_taskCount, _valueTaskCount);
24+
if (!_trackCounts)
25+
return ValueTask.FromResult((0, 0));
3126

32-
using (await _asyncLock!.LockAsync(cancellationToken).ConfigureAwait(false))
33-
{
34-
return (_taskCount, _valueTaskCount);
35-
}
27+
return ValueTask.FromResult((_taskCount.Value, _valueTaskCount.Value));
3628
}
3729

38-
public async ValueTask<bool> IsProcessing(CancellationToken cancellationToken = default)
30+
public ValueTask<bool> IsProcessing(CancellationToken cancellationToken = default)
3931
{
40-
if (!_lockCounts)
41-
{
42-
if (_valueTaskCount > 0 || _taskCount > 0)
43-
return true;
44-
45-
return false;
46-
}
32+
if (!_trackCounts)
33+
return ValueTask.FromResult(false);
4734

48-
using (await _asyncLock!.LockAsync(cancellationToken).ConfigureAwait(false))
49-
{
50-
if (_valueTaskCount > 0 || _taskCount > 0)
51-
return true;
52-
53-
return false;
54-
}
35+
return ValueTask.FromResult(_taskCount.Value > 0 || _valueTaskCount.Value > 0);
5536
}
5637

57-
public async ValueTask<int> IncrementValueTaskCounter(CancellationToken cancellationToken = default)
38+
public ValueTask<int> IncrementValueTaskCounter(CancellationToken cancellationToken = default)
5839
{
59-
if (!_lockCounts)
60-
{
61-
Interlocked.Increment(ref _valueTaskCount);
62-
63-
return _valueTaskCount;
64-
}
65-
66-
using (await _asyncLock!.LockAsync(cancellationToken).ConfigureAwait(false))
67-
{
68-
_valueTaskCount++;
40+
if (!_trackCounts)
41+
return ValueTask.FromResult(0);
6942

70-
return _valueTaskCount;
71-
}
43+
return ValueTask.FromResult(_valueTaskCount.Increment());
7244
}
7345

74-
public async ValueTask<int> DecrementValueTaskCounter(CancellationToken cancellationToken = default)
46+
public ValueTask<int> DecrementValueTaskCounter(CancellationToken cancellationToken = default)
7547
{
76-
if (!_lockCounts)
77-
{
78-
Interlocked.Decrement(ref _valueTaskCount);
48+
if (!_trackCounts)
49+
return ValueTask.FromResult(0);
7950

80-
return _valueTaskCount;
81-
}
82-
83-
using (await _asyncLock!.LockAsync(cancellationToken).ConfigureAwait(false))
84-
{
85-
_valueTaskCount--;
86-
87-
return _valueTaskCount;
88-
}
51+
return ValueTask.FromResult(_valueTaskCount.Decrement());
8952
}
9053

91-
public async ValueTask<int> IncrementTaskCounter(CancellationToken cancellationToken = default)
54+
public ValueTask<int> IncrementTaskCounter(CancellationToken cancellationToken = default)
9255
{
93-
if (!_lockCounts)
94-
{
95-
Interlocked.Increment(ref _taskCount);
96-
97-
return _taskCount;
98-
}
56+
if (!_trackCounts)
57+
return ValueTask.FromResult(0);
9958

100-
using (await _asyncLock!.LockAsync(cancellationToken).ConfigureAwait(false))
101-
{
102-
_taskCount++;
103-
104-
return _taskCount;
105-
}
59+
return ValueTask.FromResult(_taskCount.Increment());
10660
}
10761

108-
public async ValueTask<int> DecrementTaskCounter(CancellationToken cancellationToken = default)
62+
public ValueTask<int> DecrementTaskCounter(CancellationToken cancellationToken = default)
10963
{
110-
if (!_lockCounts)
111-
{
112-
Interlocked.Decrement(ref _taskCount);
113-
return _taskCount;
114-
}
115-
116-
using (await _asyncLock!.LockAsync(cancellationToken).ConfigureAwait(false))
117-
{
118-
_taskCount--;
119-
120-
return _taskCount;
121-
}
64+
if (!_trackCounts)
65+
return ValueTask.FromResult(0);
66+
67+
return ValueTask.FromResult(_taskCount.Decrement());
12268
}
12369
}

‎src/QueuedHostedService.cs‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -89,7 +89,8 @@ private async Task TaskProcessing(CancellationToken cancellationToken)
8989
}
9090
finally
9191
{
92-
await _queueInformationUtil.DecrementTaskCounter(cancellationToken).NoSync();
92+
if (workItem != null)
93+
await _queueInformationUtil.DecrementTaskCounter(CancellationToken.None).NoSync();
9394
}
9495
}
9596
}
@@ -129,7 +130,8 @@ private async Task ValueTaskProcessing(CancellationToken cancellationToken)
129130
}
130131
finally
131132
{
132-
await _queueInformationUtil.DecrementValueTaskCounter(cancellationToken).NoSync();
133+
if (workItem != null)
134+
await _queueInformationUtil.DecrementValueTaskCounter(CancellationToken.None).NoSync();
133135
}
134136
}
135137
}

‎src/Soenneker.Utils.BackgroundQueue.csproj‎

Lines changed: 11 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,6 @@
1-
<?xml version="1.0" encoding="utf-8"?><Project Sdk="Microsoft.NET.Sdk">
1+
<?xml version="1.0" encoding="utf-8"?>
2+
3+
<Project Sdk="Microsoft.NET.Sdk">
24

35
<PropertyGroup>
46
<TargetFramework>net10.0</TargetFramework>
@@ -30,19 +32,23 @@
3032
<LangVersion>latest</LangVersion>
3133
<PackageReadmeFile>README.md</PackageReadmeFile>
3234
<PackageIcon>icon.png</PackageIcon>
33-
<ContinuousIntegrationBuild>true</ContinuousIntegrationBuild></PropertyGroup>
34-
35+
<ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
36+
</PropertyGroup>
37+
3538
<ItemGroup>
3639
<None Include="..\README.md" Pack="true" PackagePath="\" />
3740
<None Include="..\LICENSE" Pack="true" PackagePath="\" />
3841
<None Include="..\icon.png" Pack="true" PackagePath="\" />
42+
</ItemGroup>
43+
44+
<ItemGroup>
3945
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="10.0.1" />
40-
<PackageReference Include="Nito.AsyncEx" Version="5.1.2" />
46+
<PackageReference Include="Soenneker.Asyncs.Locks" Version="4.0.3" />
4147
<PackageReference Include="Soenneker.Extensions.Configuration" Version="4.0.745" />
4248
<PackageReference Include="Soenneker.Extensions.Double" Version="4.0.447" />
4349
<PackageReference Include="Soenneker.Extensions.MethodInfo" Version="4.0.350" />
4450
<PackageReference Include="Soenneker.Extensions.Task" Version="4.0.108" />
4551
<PackageReference Include="Soenneker.Extensions.ValueTask" Version="4.0.98" />
4652
<PackageReference Include="Soenneker.Utils.Delay" Version="4.0.42" />
4753
</ItemGroup>
48-
</Project>
54+
</Project>

0 commit comments

Comments
 (0)