Skip to content

Commit f990ba3

Browse files
author
Pedram Rezaei
committed
Keep recovery admission separate from reader contention and repair empty queues
1 parent e61d42d commit f990ba3

9 files changed

Lines changed: 47 additions & 32 deletions

File tree

‎.github/workflows/build.yml‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -12,19 +12,18 @@ jobs:
1212
strategy:
1313
matrix:
1414
os: ["ubuntu-latest", "windows-latest", "macos-latest"]
15-
configuration: [Debug, Release]
16-
name: ${{ matrix.configuration }} on ${{ matrix.os }}
17-
15+
name: Release on ${{ matrix.os }}
16+
1817
steps:
1918

2019
- name: Checkout
2120
uses: actions/checkout@v7
22-
21+
2322
- name: Setup .NET
2423
uses: actions/setup-dotnet@v6
2524
with:
2625
dotnet-version: |
2726
10.0.x
2827
2928
- name: Build and Test
30-
run: dotnet test src/Interprocess.sln --configuration ${{ matrix.configuration }}
29+
run: dotnet test src/Interprocess.sln --configuration Release

‎.github/workflows/publish.yml‎

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ name: publish
33
on:
44
push:
55
branches: [ main ]
6-
6+
77
jobs:
88
build:
99
permissions:
@@ -15,21 +15,18 @@ jobs:
1515

1616
- name: Checkout
1717
uses: actions/checkout@v7
18-
18+
1919
- name: Setup .NET
2020
uses: actions/setup-dotnet@v6
2121
with:
2222
dotnet-version: |
2323
10.0.x
2424
25-
- name: Test - Debug
26-
run: dotnet test src/Interprocess.sln
27-
2825
- name: Test - Release
2926
run: dotnet test -c Release -p:VersionSuffix=${{github.run_number}} src/Interprocess.sln
3027

3128
- name: Pack
32-
run: dotnet pack -c Release --no-build --output . --version-suffix ${{github.run_number}} src/Interprocess/Interprocess.csproj
29+
run: dotnet pack -c Release --no-build --output . --version-suffix ${{github.run_number}} src/Interprocess/Interprocess.csproj
3330

3431
- name: NuGet login
3532
uses: NuGet/login@v1

‎src/Interprocess.Tests/PublisherRecoveryTests.cs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -200,7 +200,7 @@ internal unsafe void LeaveDeadRegistrationInLastSlot(QueueOptions registrationOp
200200
internal long Register() => RegisterParticipant();
201201
internal PublisherLease Lease(long id) => Publishers.Register(id);
202202
internal bool AnyActive() => Publishers.AnyActive();
203-
internal unsafe void CloseAdmission() => Interlocked.Exchange(ref Header->ReadLockOwner, long.MinValue | 1);
204-
internal unsafe void OpenAdmission() => Interlocked.Exchange(ref Header->ReadLockOwner, 0);
203+
internal void CloseAdmission() => Publishers.CloseAdmission();
204+
internal void OpenAdmission() => Publishers.OpenAdmission();
205205
}
206206
}

‎src/Interprocess.Tests/ReaderOwnershipTests.cs‎

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -41,10 +41,11 @@ public void RegistrationExhaustionDoesNotWrapOrBreakExistingParticipants()
4141
}
4242

4343
[Theory]
44-
[InlineData(false, false)]
45-
[InlineData(true, false)]
46-
[InlineData(true, true)]
47-
public async Task ReaderProcessLifetimeControlsRecoveryAsync(bool crash, bool admissionClosed)
44+
[InlineData(false, false, false)]
45+
[InlineData(true, false, false)]
46+
[InlineData(true, true, false)]
47+
[InlineData(true, true, true)]
48+
public async Task ReaderProcessLifetimeControlsRecoveryAsync(bool crash, bool admissionClosed, bool emptied)
4849
{
4950
var options = new QueueOptions(Guid.NewGuid().ToStringInvariant("N")[..16], fixture.Path, 1024);
5051
var factory = new QueueFactory();
@@ -83,10 +84,13 @@ public async Task ReaderProcessLifetimeControlsRecoveryAsync(bool crash, bool ad
8384
if (admissionClosed)
8485
probe.CloseAdmission();
8586

87+
if (emptied)
88+
probe.EmptyQueue();
89+
8690
child.Kill();
8791
await child.WaitForExitAsync().WaitAsync(TimeSpan.FromSeconds(5));
8892
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(30));
89-
while (probe.ReadOffset == 0)
93+
while (probe.ReadOffset == 0 || probe.Owner != 0)
9094
{
9195
survivor.TryDequeue(out _).Should().BeFalse("the crashed read cannot be delivered again");
9296
await Task.Delay(10, timeout.Token);
@@ -263,7 +267,13 @@ internal unsafe int LastParticipantId
263267
set => Volatile.Write(ref Header->LastParticipantId, value);
264268
}
265269

266-
internal unsafe void CloseAdmission() => Interlocked.Or(ref Header->ReadLockOwner, long.MinValue);
270+
internal void CloseAdmission() => Publishers.CloseAdmission();
271+
272+
internal unsafe void EmptyQueue()
273+
{
274+
Buffer.Clear(Header->ReadOffset, Header->WriteOffset - Header->ReadOffset);
275+
Interlocked.Exchange(ref Header->ReadOffset, Header->WriteOffset);
276+
}
267277
}
268278

269279
private sealed class PausedMemory(ManualResetEventSlim entered, ManualResetEventSlim resume) : MemoryManager<byte>

‎src/Interprocess/Contracts/ISubscriber.cs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,8 @@ public interface ISubscriber : IDisposable
3030
bool TryDequeue(out ReadOnlyMemory<byte> message);
3131

3232
/// <summary>
33-
/// Attempts to dequeue the next message if it is ready. It does not wait for a message. This method populates the <paramref name="buffer"/> that is passed in.
33+
/// Attempts to dequeue the next message if it is ready. It does not wait for a message.
34+
/// This method populates the <paramref name="buffer"/> that is passed in.
3435
/// Make sure that the buffer is large enough to receive the entire message, or the message is truncated to fit the buffer.
3536
/// </summary>
3637
/// <param name="buffer">The memory buffer that is populated with the message. Make sure

‎src/Interprocess/Contracts/QueueHeader.cs‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@ internal struct QueueHeader
1919

2020
/// <summary>
2121
/// The unique subscriber registration holding the read lock, or zero.
22-
/// The sign bit closes publication admission during recovery.
2322
/// </summary>
2423
[FieldOffset(16)]
2524
internal long ReadLockOwner;

‎src/Interprocess/Queue/Publisher.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,7 @@ public unsafe bool TryEnqueue(ReadOnlySpan<byte> message)
3737
try
3838
{
3939
// Recovery closes admission before inspecting shared in-flight counters.
40-
return Volatile.Read(ref Header->ReadLockOwner) >= 0 && TryEnqueueCore(message);
40+
return !Publishers.IsAdmissionClosed && TryEnqueueCore(message);
4141
}
4242
finally
4343
{

‎src/Interprocess/Queue/PublisherRegistry.cs‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,14 @@ internal sealed unsafe class PublisherRegistry(QueueOptions options, byte* queue
55
{
66
internal const int MaximumPublishers = 2048;
77
internal const int SlotSize = 128;
8-
internal const int TableOffset = 128; // Keep counters apart from the 32-byte queue header.
8+
internal const int TableOffset = 256; // Header, recovery gate, and counters occupy separate cache lines.
99
internal const int BufferOffset = TableOffset + (MaximumPublishers * SlotSize);
1010

11+
internal bool IsAdmissionClosed => Volatile.Read(ref *(int*)(queue + 128)) != 0;
12+
13+
internal void CloseAdmission() => Interlocked.Exchange(ref *(int*)(queue + 128), 1);
14+
internal void OpenAdmission() => Interlocked.Exchange(ref *(int*)(queue + 128), 0);
15+
1116
internal PublisherLease Register(long id)
1217
{
1318
var lifetime = new ReaderLease(options, id);

‎src/Interprocess/Queue/Subscriber.cs‎

Lines changed: 13 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -163,17 +163,15 @@ private unsafe bool TryDequeueImpl(
163163
{
164164
message = ReadOnlyMemory<byte>.Empty;
165165
var header = Header;
166-
if (header->IsEmpty())
167-
return false;
168-
169166
var owner = header->ReadLockOwner;
170167
if (owner != 0)
171168
{
172169
TryRecoverReader(owner);
173170
return false;
174171
}
175172

176-
if (Interlocked.CompareExchange(ref header->ReadLockOwner, readerId, 0L) != 0)
173+
// A dead recovering reader can leave admission closed even when the queue is empty.
174+
if (header->IsEmpty() || Interlocked.CompareExchange(ref header->ReadLockOwner, readerId, 0L) != 0)
177175
return false;
178176

179177
try
@@ -207,7 +205,7 @@ private unsafe bool TryDequeueImpl(
207205
{
208206
// Clear through the captured tail before publishers can reuse the space.
209207
// Otherwise discarded ready headers could be consumed on a later lap.
210-
Interlocked.Exchange(ref Header->ReadLockOwner, readerId | long.MinValue);
208+
Publishers.CloseAdmission();
211209
try
212210
{
213211
if (Publishers.AnyActive())
@@ -222,7 +220,7 @@ private unsafe bool TryDequeueImpl(
222220
}
223221
finally
224222
{
225-
Interlocked.Exchange(ref Header->ReadLockOwner, readerId);
223+
Publishers.OpenAdmission();
226224
}
227225
}
228226

@@ -271,16 +269,22 @@ private unsafe bool TryDequeueImpl(
271269
private unsafe void TryRecoverReader(long owner)
272270
{
273271
// Another call on this subscriber is still alive. Only its owner can release it.
274-
if ((owner & long.MaxValue) == readerId)
272+
if (owner == readerId)
275273
return;
276274

277275
var next = Volatile.Read(ref nextRecoveryCheck);
278276
var now = Stopwatch.GetTimestamp();
279277
if (now < next || Interlocked.CompareExchange(ref nextRecoveryCheck, now + RecoveryInterval, next) != next)
280278
return;
281279

282-
if (!ReaderLease.IsAlive(options, owner & long.MaxValue))
283-
Interlocked.CompareExchange(ref Header->ReadLockOwner, 0L, owner);
280+
// Take ownership before reopening admission. A delayed recovery attempt must
281+
// not reopen the gate after another reader has started its own recovery.
282+
if (!ReaderLease.IsAlive(options, owner)
283+
&& Interlocked.CompareExchange(ref Header->ReadLockOwner, readerId, owner) == owner)
284+
{
285+
Publishers.OpenAdmission();
286+
Interlocked.CompareExchange(ref Header->ReadLockOwner, 0L, readerId);
287+
}
284288
}
285289

286290
[MethodImpl(MethodImplOptions.AggressiveInlining)]

0 commit comments

Comments
 (0)