diff --git a/README.md b/README.md index 3234fe8..048074d 100644 --- a/README.md +++ b/README.md @@ -39,7 +39,7 @@ Creating a message queue publisher: ```csharp var options = new QueueOptions( queueName: "my-queue", - bytesCapacity: 1024 * 1024); + capacity: 1024 * 1024); using var publisher = factory.CreatePublisher(options); publisher.TryEnqueue(message); @@ -50,7 +50,7 @@ Creating a message queue subscriber: ```csharp options = new QueueOptions( queueName: "my-queue", - bytesCapacity: 1024 * 1024); + capacity: 1024 * 1024); using var subscriber = factory.CreateSubscriber(options); subscriber.TryDequeue(messageBuffer, cancellationToken, out var message); @@ -71,7 +71,7 @@ Creating a message queue publisher using an instance of `IQueueFactory` retrieve ```csharp var options = new QueueOptions( queueName: "my-queue", - bytesCapacity: 1024 * 1024); + capacity: 1024 * 1024); using var publisher = factory.CreatePublisher(options); publisher.TryEnqueue(message); @@ -82,7 +82,7 @@ Creating a message queue subscriber using an instance of `IQueueFactory` retriev ```csharp var options = new QueueOptions( queueName: "my-queue", - bytesCapacity: 1024 * 1024); + capacity: 1024 * 1024); using var subscriber = factory.CreateSubscriber(options); subscriber.TryDequeue(messageBuffer, cancellationToken, out var message); @@ -112,7 +112,7 @@ Please note that you can start multiple publishers and subscribers sending and r A lot has gone into optimizing the implementation of this library. For instance, it is mostly heap-memory allocation free, reducing the need for garbage collection induced pauses. -**Summary**: A full enqueue followed by a dequeue takes `~250 ns` on Linux, `~650 ns` on macOS, and `~300 ns` on Windows. +**Latest native macOS measurement**: a three-byte enqueue/dequeue round trip with a reused buffer averaged **210.0 ns** on an Apple M5 Max. Only the macOS results below were refreshed on September 12, 2026; the Windows and Linux sections retain their historical measurements. **Details**: To benchmark the performance and memory usage, we use [BenchmarkDotNet][BenchmarkOrg] and perform the following runs: @@ -124,7 +124,13 @@ A lot has gone into optimizing the implementation of this library. For instance, You can replicate the results by running the following command: ```sh -dotnet run Interprocess.Benchmark.csproj -c Release +dotnet run --project src/Interprocess.Benchmark -c Release -- --filter '*QueueBenchmark*' +``` + +To compare throughput with one subscriber versus four concurrent subscribers: + +```sh +dotnet run --project src/Interprocess.Benchmark -c Release -- --filter '*SubscriberBenchmark*' --iterationCount 8 ``` --- @@ -152,20 +158,37 @@ Results: ### On macOS -Host: +Measured **September 12, 2026**, running directly on the Mac: ```text -BenchmarkDotNet v0.14.0, macOS Sequoia 15.2 (24C101) [Darwin 24.2.0] -Apple M3 Max, 1 CPU, 16 logical and 16 physical cores -.NET SDK 9.0.101 - [Host] : .NET 9.0.0 (9.0.24.52809), Arm64 RyuJIT AdvSIMD - .NET 9.0 : .NET 9.0.0 (9.0.24.52809), Arm64 RyuJIT AdvSIMD +BenchmarkDotNet v0.15.8, macOS Tahoe 26.6.2 (25G83) [Darwin 25.6.0] +Apple M5 Max, 1 CPU, 18 logical and 18 physical cores +.NET SDK 10.0.401 +.NET runtime 10.0.12, Arm64 RyuJIT +Release build; 3 warm-up iterations; 8 measured iterations; 1 launch ``` -| Method | Mean (ns) | Error (ns) | StdDev | Gen0 | Allocated | -|-------------------------------------------------- |----------:|-----------:|-------:|---------:|----------:| -| 'Message enqueue and dequeue' | `249.2` | `0.74` | `0.62` | `-` | `-` | -| 'Message enqueue and dequeue - no message buffer' | `252.1` | `4.10` | `3.83` | `0.0038` | `32 B` | +All seven cases completed. Times are means in nanoseconds, normalized per operation. For enqueue/dequeue rows, an operation is one complete round trip. Concurrent-delivery rows report amortized time per delivered message. + +| Workload | Mean (ns) | StdDev (ns) | Allocated per operation | +| --- | ---: | ---: | ---: | +| Enqueue, 3 bytes | 182.3 | 5.01 | 0 B | +| Enqueue + dequeue, 3 bytes, reused buffer | 210.0 | 0.46 | 0 B | +| Enqueue + dequeue, 3 bytes, new result array | 214.9 | 0.81 | 32 B | +| Enqueue + dequeue, 50 bytes, reused buffer | 214.6 | 1.33 | 0 B | +| Enqueue + dequeue, 50 bytes, ring-wrap workload | 223.8 | 1.08 | 0 B | +| Concurrent delivery, 8 bytes, 1 subscriber | 246.5 | 2.20 | Not measured | +| Concurrent delivery, 8 bytes, 4 subscribers | 344.7 | 1.81 | Not measured | + +The enqueue case batches 320,000 messages and drains the queue outside the timed body. The ring-wrap case uses a 120-byte queue so padded 64-byte records repeatedly cross the end of the buffer; two round trips per invocation are normalized to one. Concurrent delivery uses one publisher and dedicated subscriber threads to transfer batches of 65,536 messages, including worker startup and completion in the timing. + +These are in-process microbenchmarks, not end-to-end latency between separate applications. The concurrent cases measure throughput under contention, not individual message latency; their allocations were not measured. See the [complete native Mac reports and methodology](docs/benchmarks/2026-09-12/README.md) for source revision, errors, and reproduction details. + +Run all cases from the repository root: + +```sh +dotnet run --project src/Interprocess.Benchmark -c Release -- --filter '*' --warmupCount 3 --iterationCount 8 --artifacts BenchmarkDotNet.Artifacts +``` --- diff --git a/docs/benchmarks/2026-09-12/README.md b/docs/benchmarks/2026-09-12/README.md new file mode 100644 index 0000000..eb9b21a --- /dev/null +++ b/docs/benchmarks/2026-09-12/README.md @@ -0,0 +1,31 @@ +# Native macOS benchmark measurements — September 12, 2026 + +All seven benchmark cases were run directly on an Apple M5 Max using benchmark and library source at [`c7d0755`](https://github.com/cloudtoid/interprocess/commit/c7d07559a48045c82b63940da9f7910b888d9e82). The later README update does not change the measured code. + +[Complete BenchmarkDotNet reports](macos-local.md) + +## Environment + +- macOS Tahoe 26.6.2 (25G83), Darwin 25.6.0. +- Apple M5 Max, arm64, 18 logical and physical cores reported. +- BenchmarkDotNet 0.15.8; .NET SDK 10.0.401; .NET runtime 10.0.12; Release configuration. +- Three warm-up iterations and eight measured iterations per case, one launch, with BenchmarkDotNet's usual pilot, overhead, and outlier handling. +- These are in-process microbenchmarks. They do not measure end-to-end latency between separate applications, idle CPU usage, or latency percentiles. + +## Workloads + +- **Enqueue:** 320,000 three-byte messages per invocation, normalized to one enqueue. Queue capacity is 5,120,000 bytes. Draining and validation happen outside the timed body. Each enqueue checks that it succeeded. +- **Three-byte round trips:** enqueue followed by dequeue on the calling thread, with either a reused buffer or a newly allocated result array; queue capacity 128 bytes. +- **50-byte round trips:** the same calling-thread pattern with a reused buffer and a 128-byte queue. +- **Ring-wrap workload:** 50-byte messages in a 120-byte queue, so padded 64-byte records repeatedly cross the end of the ring. Each invocation performs two enqueue/dequeue pairs; reported time is normalized to one pair. +- **Concurrent delivery:** one publisher sends 65,536 eight-byte messages to one or four dedicated subscriber threads sharing a 65,536-byte queue. Each subscriber receives an equal share. Time is normalized per delivered message and includes per-batch worker startup and completion. + +Memory diagnostics reported 0 B/op for enqueue and reused-buffer round trips, and 32 B/op for the three-byte result-array case. Allocation diagnostics were not enabled for concurrent delivery, which creates workers and tasks per batch. + +## Reproduce + +From the repository root: + +```sh +dotnet run --project src/Interprocess.Benchmark -c Release -- --filter '*' --warmupCount 3 --iterationCount 8 --artifacts BenchmarkDotNet.Artifacts +``` diff --git a/docs/benchmarks/2026-09-12/macos-local.md b/docs/benchmarks/2026-09-12/macos-local.md new file mode 100644 index 0000000..a083a02 --- /dev/null +++ b/docs/benchmarks/2026-09-12/macos-local.md @@ -0,0 +1,80 @@ +# Local macOS — Apple M5 Max + +Measured September 12, 2026, using source commit [`c7d0755`](https://github.com/cloudtoid/interprocess/commit/c7d07559a48045c82b63940da9f7910b888d9e82). + +See [methodology and reproduction commands](README.md). The tables below are BenchmarkDotNet exports. + +## EnqueueBenchmark + +``` + +BenchmarkDotNet v0.15.8, macOS Tahoe 26.6.2 (25G83) [Darwin 25.6.0] +Apple M5 Max, 1 CPU, 18 logical and 18 physical cores +.NET SDK 10.0.401 + [Host] : .NET 10.0.12 (10.0.12, 10.0.1226.42308), Arm64 RyuJIT armv8.0-a + .NET 10.0 : .NET 10.0.12 (10.0.12, 10.0.1226.42308), Arm64 RyuJIT armv8.0-a + +Job=.NET 10.0 Runtime=.NET 10.0 InvocationCount=1 +IterationCount=8 UnrollFactor=1 WarmupCount=3 + +``` +| Method | Mean | Error | StdDev | Allocated | +|------------------ |---------:|--------:|--------:|----------:| +| 'Message enqueue' | 182.3 ns | 9.57 ns | 5.01 ns | - | + +## QueueBenchmark + +``` + +BenchmarkDotNet v0.15.8, macOS Tahoe 26.6.2 (25G83) [Darwin 25.6.0] +Apple M5 Max, 1 CPU, 18 logical and 18 physical cores +.NET SDK 10.0.401 + [Host] : .NET 10.0.12 (10.0.12, 10.0.1226.42308), Arm64 RyuJIT armv8.0-a + .NET 10.0 : .NET 10.0.12 (10.0.12, 10.0.1226.42308), Arm64 RyuJIT armv8.0-a + +Job=.NET 10.0 Runtime=.NET 10.0 IterationCount=8 +WarmupCount=3 + +``` +| Method | Mean | Error | StdDev | Gen0 | Allocated | +|-------------------------------------------------- |---------:|--------:|--------:|-------:|----------:| +| 'Message enqueue and dequeue - no message buffer' | 214.9 ns | 1.54 ns | 0.81 ns | 0.0038 | 32 B | +| 'Message enqueue and dequeue' | 210.0 ns | 1.03 ns | 0.46 ns | - | - | + +## QueueExtendedBenchmark + +``` + +BenchmarkDotNet v0.15.8, macOS Tahoe 26.6.2 (25G83) [Darwin 25.6.0] +Apple M5 Max, 1 CPU, 18 logical and 18 physical cores +.NET SDK 10.0.401 + [Host] : .NET 10.0.12 (10.0.12, 10.0.1226.42308), Arm64 RyuJIT armv8.0-a + .NET 10.0 : .NET 10.0.12 (10.0.12, 10.0.1226.42308), Arm64 RyuJIT armv8.0-a + +Job=.NET 10.0 Runtime=.NET 10.0 IterationCount=8 +WarmupCount=3 + +``` +| Method | Mean | Error | StdDev | Allocated | +|--------------------------------------------------- |---------:|--------:|--------:|----------:| +| 'Message enqueue and dequeue - long message' | 214.6 ns | 2.55 ns | 1.33 ns | - | +| 'Message enqueue and dequeue - ring-wrap workload' | 223.8 ns | 2.43 ns | 1.08 ns | - | + +## SubscriberBenchmark + +``` + +BenchmarkDotNet v0.15.8, macOS Tahoe 26.6.2 (25G83) [Darwin 25.6.0] +Apple M5 Max, 1 CPU, 18 logical and 18 physical cores +.NET SDK 10.0.401 + [Host] : .NET 10.0.12 (10.0.12, 10.0.1226.42308), Arm64 RyuJIT armv8.0-a + ShortRun : .NET 10.0.12 (10.0.12, 10.0.1226.42308), Arm64 RyuJIT armv8.0-a + +Job=ShortRun IterationCount=8 LaunchCount=1 +WarmupCount=3 + +``` +| Method | SubscriberCount | Mean | Error | StdDev | +|------------------------- |---------------- |---------:|--------:|--------:| +| **ReceiveConcurrentlyAsync** | **1** | **246.5 ns** | **4.21 ns** | **2.20 ns** | +| **ReceiveConcurrentlyAsync** | **4** | **344.7 ns** | **3.45 ns** | **1.81 ns** | diff --git a/src/Interprocess.Benchmark/Program.cs b/src/Interprocess.Benchmark/Program.cs index d466157..cab6510 100644 --- a/src/Interprocess.Benchmark/Program.cs +++ b/src/Interprocess.Benchmark/Program.cs @@ -4,5 +4,5 @@ namespace Cloudtoid.Interprocess.Benchmark; public sealed class Program { - public static void Main() => _ = BenchmarkRunner.Run(typeof(Program).Assembly); + public static void Main(string[] args) => BenchmarkSwitcher.FromAssembly(typeof(Program).Assembly).Run(args); } \ No newline at end of file diff --git a/src/Interprocess.Benchmark/Queue/EnqueueBenchmark.cs b/src/Interprocess.Benchmark/Queue/EnqueueBenchmark.cs index 2164d68..4f8ab75 100644 --- a/src/Interprocess.Benchmark/Queue/EnqueueBenchmark.cs +++ b/src/Interprocess.Benchmark/Queue/EnqueueBenchmark.cs @@ -3,11 +3,12 @@ namespace Cloudtoid.Interprocess.Benchmark; -[SimpleJob(RuntimeMoniker.Net90)] +[SimpleJob(RuntimeMoniker.Net10_0)] [MemoryDiagnoser] [MarkdownExporterAttribute.GitHub] public class EnqueueBenchmark { + private const int MessageCount = 320000; private static readonly byte[] Message = [100, 110, 120]; private static readonly Memory MessageBuffer = new byte[Message.Length]; #pragma warning disable CS8618 @@ -19,8 +20,8 @@ public class EnqueueBenchmark public void Setup() { var queueFactory = new QueueFactory(); - publisher = queueFactory.CreatePublisher(new QueueOptions("qn", Path.GetTempPath(), 5120000)); - subscriber = queueFactory.CreateSubscriber(new QueueOptions("qn", Path.GetTempPath(), 5120000)); + publisher = queueFactory.CreatePublisher(new QueueOptions("qn", Path.GetTempPath(), MessageCount * 16)); + subscriber = queueFactory.CreateSubscriber(new QueueOptions("qn", Path.GetTempPath(), MessageCount * 16)); } [GlobalCleanup] @@ -33,15 +34,21 @@ public void Cleanup() [IterationCleanup] public void DrainQueue() { - for (int i = 8; i < 320000; i++) - subscriber.Dequeue(MessageBuffer, default); + for (var i = 0; i < MessageCount; i++) + { + if (!subscriber.TryDequeue(MessageBuffer, default, out _)) + throw new InvalidOperationException("The benchmark did not enqueue the expected number of messages."); + } } // Expecting that there are NO managed heap allocations. - [Benchmark(Description = "Message enqueue (320,000 times)")] + [Benchmark(Description = "Message enqueue", OperationsPerInvoke = MessageCount)] public void Enqueue() { - for (int i = 8; i < 320000; i++) - publisher.TryEnqueue(Message); + for (var i = 0; i < MessageCount; i++) + { + if (!publisher.TryEnqueue(Message)) + throw new InvalidOperationException("The benchmark queue is full."); + } } } \ No newline at end of file diff --git a/src/Interprocess.Benchmark/Queue/QueueBenchmark.cs b/src/Interprocess.Benchmark/Queue/QueueBenchmark.cs index a40330f..dec9528 100644 --- a/src/Interprocess.Benchmark/Queue/QueueBenchmark.cs +++ b/src/Interprocess.Benchmark/Queue/QueueBenchmark.cs @@ -3,7 +3,7 @@ namespace Cloudtoid.Interprocess.Benchmark; -[SimpleJob(RuntimeMoniker.Net90)] +[SimpleJob(RuntimeMoniker.Net10_0)] [MemoryDiagnoser] [MarkdownExporterAttribute.GitHub] public class QueueBenchmark diff --git a/src/Interprocess.Benchmark/Queue/QueueExtendedBenchmark.cs b/src/Interprocess.Benchmark/Queue/QueueExtendedBenchmark.cs index bed7be0..d38199c 100644 --- a/src/Interprocess.Benchmark/Queue/QueueExtendedBenchmark.cs +++ b/src/Interprocess.Benchmark/Queue/QueueExtendedBenchmark.cs @@ -3,7 +3,8 @@ namespace Cloudtoid.Interprocess.Benchmark; -[SimpleJob(RuntimeMoniker.Net90)] +[SimpleJob(RuntimeMoniker.Net10_0)] +[MemoryDiagnoser] [MarkdownExporterAttribute.GitHub] public class QueueExtendedBenchmark { @@ -14,13 +15,11 @@ public class QueueExtendedBenchmark private ISubscriber subscriber; #pragma warning restore CS8618 - [GlobalSetup] - public void Setup() - { - var queueFactory = new QueueFactory(); - publisher = queueFactory.CreatePublisher(new QueueOptions("qn", Path.GetTempPath(), 128)); - subscriber = queueFactory.CreateSubscriber(new QueueOptions("qn", Path.GetTempPath(), 128)); - } + [GlobalSetup(Target = nameof(EnqueueDequeue_LongMessage))] + public void Setup() => SetupQueue(128); + + [GlobalSetup(Target = nameof(EnqueueDequeue_WrappedMessages))] + public void SetupWrapped() => SetupQueue(120); [GlobalCleanup] public void Cleanup() @@ -38,7 +37,9 @@ public ReadOnlyMemory EnqueueDequeue_LongMessage() return subscriber.Dequeue(MessageBuffer, default); } - [Benchmark(Description = "Message enqueue and dequeue - wrapped message in circular buffer")] + // A padded message occupies 64 bytes. A 120-byte ring makes message bodies cross + // the end of the buffer; a 128-byte ring only cycles between aligned slots. + [Benchmark(Description = "Message enqueue and dequeue - ring-wrap workload", OperationsPerInvoke = 2)] public ReadOnlyMemory EnqueueDequeue_WrappedMessages() { if (!publisher.TryEnqueue(Message)) @@ -51,4 +52,11 @@ public ReadOnlyMemory EnqueueDequeue_WrappedMessages() return subscriber.Dequeue(MessageBuffer, default); } + + private void SetupQueue(long capacity) + { + var queueFactory = new QueueFactory(); + publisher = queueFactory.CreatePublisher(new QueueOptions("qn", Path.GetTempPath(), capacity)); + subscriber = queueFactory.CreateSubscriber(new QueueOptions("qn", Path.GetTempPath(), capacity)); + } } \ No newline at end of file diff --git a/src/Interprocess.Benchmark/Queue/SubscriberBenchmark.cs b/src/Interprocess.Benchmark/Queue/SubscriberBenchmark.cs new file mode 100644 index 0000000..c123fc7 --- /dev/null +++ b/src/Interprocess.Benchmark/Queue/SubscriberBenchmark.cs @@ -0,0 +1,65 @@ +using BenchmarkDotNet.Attributes; + +namespace Cloudtoid.Interprocess.Benchmark; + +[ShortRunJob] +[MarkdownExporterAttribute.GitHub] +public class SubscriberBenchmark +{ + private const int MessageCount = 65536; + private readonly QueueFactory factory = new(); + private readonly QueueOptions options = new("subscriber-bench", 65536); + private IPublisher publisher = null!; + private ISubscriber[] subscribers = []; + + [Params(1, 4)] + public int SubscriberCount { get; set; } + + [GlobalSetup] + public void Setup() + { + publisher = factory.CreatePublisher(options); + subscribers = Enumerable.Range(0, SubscriberCount).Select(_ => factory.CreateSubscriber(options)).ToArray(); + } + + [GlobalCleanup] + public void Cleanup() + { + foreach (var subscriber in subscribers) + subscriber.Dispose(); + + publisher.Dispose(); + } + + [Benchmark(OperationsPerInvoke = MessageCount)] + public async Task ReceiveConcurrentlyAsync() + { + using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + var readers = new Task[subscribers.Length]; + for (var reader = 0; reader < subscribers.Length; reader++) + { + var subscriber = subscribers[reader]; + readers[reader] = Task.Factory.StartNew( + () => + { + var buffer = new byte[8]; + for (var i = 0; i < MessageCount / SubscriberCount; i++) + subscriber.Dequeue(buffer, cancellation.Token); + }, + cancellation.Token, + TaskCreationOptions.LongRunning, + TaskScheduler.Default); + } + + for (var i = 0; i < MessageCount; i++) + { + while (!publisher.TryEnqueue("message!"u8)) + { + cancellation.Token.ThrowIfCancellationRequested(); + Thread.Yield(); + } + } + + await Task.WhenAll(readers); + } +} \ No newline at end of file diff --git a/src/Interprocess.Tests/SubscriberTests.cs b/src/Interprocess.Tests/SubscriberTests.cs new file mode 100644 index 0000000..f35a46a --- /dev/null +++ b/src/Interprocess.Tests/SubscriberTests.cs @@ -0,0 +1,173 @@ +namespace Cloudtoid.Interprocess.Tests; + +public sealed class SubscriberTests(UniquePathFixture fixture) : IClassFixture +{ + private readonly QueueOptions options = new(Guid.NewGuid().ToStringInvariant("N")[..16], fixture.Path, 256); + private readonly QueueFactory factory = new(); + + [Fact] + public async Task TryDequeueDoesNotWaitForAnotherSubscriberAsync() + { + using var probe = new QueueProbe(options); + using var publisher = factory.CreatePublisher(options); + using var subscriber = factory.CreateSubscriber(options); + publisher.TryEnqueue("message!"u8).Should().BeTrue(); + probe.LockReads(); + try + { + var attempt = Task.Run(() => subscriber.TryDequeue(default, out _)); + (await attempt.WaitAsync(TimeSpan.FromSeconds(1))).Should().BeFalse(); + } + finally + { + probe.UnlockReads(); + } + + subscriber.TryDequeue(default, out _).Should().BeTrue(); + } + + [Fact] + public void ExpiredSubscriberLockCanBeRecovered() + { + using var probe = new QueueProbe(options); + using var publisher = factory.CreatePublisher(options); + using var subscriber = factory.CreateSubscriber(options); + publisher.TryEnqueue("message!"u8).Should().BeTrue(); + probe.AbandonReadLock(); + subscriber.TryDequeue(default, out var message).Should().BeTrue(); + message.ToArray().Should().Equal("message!"u8.ToArray()); + probe.ReadsAreLocked.Should().BeFalse(); + } + + [Fact] + public async Task BlockingDequeueRetriesContendedReadAsync() + { + using var probe = new QueueProbe(options); + using var publisher = factory.CreatePublisher(options); + using var subscriber = factory.CreateSubscriber(options); + using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + publisher.TryEnqueue("message!"u8).Should().BeTrue(); + probe.LockReads(); + var read = Task.Run(() => subscriber.Dequeue(cancellation.Token)); + try + { + await Task.Delay(20); + read.IsCompleted.Should().BeFalse(); + } + finally + { + probe.UnlockReads(); + } + + (await read.WaitAsync(TimeSpan.FromSeconds(1))).ToArray().Should().Equal("message!"u8.ToArray()); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task CancellationInterruptsUnfinishedMessageAsync(bool blocking) + { + using var probe = new QueueProbe(options); + using var subscriber = factory.CreateSubscriber(options); + using var cancellation = new CancellationTokenSource(); + probe.ReserveUnfinishedMessage(); + var read = Task.Run(() => + { + if (blocking) + subscriber.Dequeue(cancellation.Token); + else + subscriber.TryDequeue(cancellation.Token, out _); + }); + + try + { + SpinWait.SpinUntil(() => probe.ReadsAreLocked, TimeSpan.FromSeconds(1)).Should().BeTrue(); + await cancellation.CancelAsync(); + await Assert.ThrowsAnyAsync( + async () => await read.WaitAsync(TimeSpan.FromSeconds(1))); + probe.ReadsAreLocked.Should().BeFalse(); + } + finally + { + await cancellation.CancelAsync(); + } + } + + [Fact] + public async Task EmptyBlockingDequeueCanBeCancelledAsync() + { + using var subscriber = factory.CreateSubscriber(options); + using var cancellation = new CancellationTokenSource(); + var read = Task.Run(() => subscriber.Dequeue(cancellation.Token)); + await Task.Delay(20); + await cancellation.CancelAsync(); + await Assert.ThrowsAnyAsync( + async () => await read.WaitAsync(TimeSpan.FromSeconds(1))); + } + + [Theory] + [InlineData(1)] + [InlineData(4)] + public async Task ConcurrentSubscribersReceiveEveryMessageExactlyOnceAsync(int subscriberCount) + { + const int count = 4000; + var received = new int[count]; + using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + // Keep a participant alive while the workers start and finish. + using var anchor = factory.CreatePublisher(options); + var readers = new Task[subscriberCount]; + for (var reader = 0; reader < subscriberCount; reader++) + { + readers[reader] = Task.Run(() => + { + using var subscriber = factory.CreateSubscriber(options); + var buffer = new byte[8]; + for (var i = 0; i < count / subscriberCount; i++) + { + var message = subscriber.Dequeue(buffer, cancellation.Token); + message.Length.Should().Be(8); + var id = BitConverter.ToInt32(message.Span); + id.Should().BeInRange(0, count - 1); + Interlocked.Increment(ref received[id]); + } + }); + } + + var writers = new Task[2]; + for (var writer = 0; writer < writers.Length; writer++) + { + var firstId = writer; + writers[writer] = Task.Run(() => + { + using var publisher = factory.CreatePublisher(options); + var buffer = new byte[8]; + for (var id = firstId; id < count; id += 2) + { + BitConverter.TryWriteBytes(buffer, id).Should().BeTrue(); + while (!publisher.TryEnqueue(buffer)) + { + cancellation.Token.ThrowIfCancellationRequested(); + Thread.Yield(); + } + } + }); + } + + await Task.WhenAll(readers.Concat(writers)).WaitAsync(TimeSpan.FromSeconds(15)); + received.Should().OnlyContain(value => value == 1); + } + + private sealed class QueueProbe(QueueOptions options) : Queue(options, NullLoggerFactory.Instance) + { + internal unsafe bool ReadsAreLocked => Interlocked.Read(ref Header->ReadLockTimestamp) != 0; + + internal unsafe void LockReads() => Interlocked.Exchange(ref Header->ReadLockTimestamp, DateTime.UtcNow.Ticks); + + internal unsafe void AbandonReadLock() => + Interlocked.Exchange(ref Header->ReadLockTimestamp, DateTime.UtcNow.Ticks - TimeSpan.FromSeconds(11).Ticks); + + internal unsafe void UnlockReads() => Interlocked.Exchange(ref Header->ReadLockTimestamp, 0); + + internal unsafe void ReserveUnfinishedMessage() => Interlocked.Exchange(ref Header->WriteOffset, 16); + } +} \ No newline at end of file diff --git a/src/Interprocess/Queue/Subscriber.cs b/src/Interprocess/Queue/Subscriber.cs index 13f7c3f..dc2aa71 100644 --- a/src/Interprocess/Queue/Subscriber.cs +++ b/src/Interprocess/Queue/Subscriber.cs @@ -83,18 +83,23 @@ private ReadOnlyMemory DequeueCore(Memory? resultBuffer, Cancellatio try { - int i = -5; + SpinWait spin = default; while (true) { if (TryDequeueImpl(resultBuffer, cancellation, out var message)) return message; - if (i > 10) - signal.Wait(millisecondsTimeout: 10); - else if (i++ > 0) - signal.Wait(millisecondsTimeout: i); + // Retry briefly in user space while another reader finishes. Once spinning + // would yield, wait for a signal instead of burning CPU on an idle queue. + if (spin.NextSpinWillYield) + { + signal.Wait(millisecondsTimeout: 5); + spin.Reset(); + } else - Thread.Yield(); + { + spin.SpinOnce(); + } } } finally @@ -159,6 +164,7 @@ private unsafe bool TryDequeueImpl( Interlocked.Exchange(ref Header->ReadOffset, writeOffset); return false; } + cancellationSource.ThrowIfCancellationRequested(cancellation); Thread.Yield(); }