From 1a80b22886e78173c0f1baccc34d4a779e242ec9 Mon Sep 17 00:00:00 2001 From: Simon Cropp Date: Sat, 29 Aug 2026 17:29:27 +1000 Subject: [PATCH 1/2] Overlap the two reads when comparing binary snapshots The verified and received streams were filled strictly one after the other, in 8KB chunks, so a passing run read both files end to end sequentially for every binary snapshot. The two streams are independent, so the fills now overlap, which roughly halves the wall time. The chunk is 64KB, still under the large object heap threshold and inside the array pool's buckets, so there are 8x fewer round trips. On net6.0 and above the Memory overload of ReadAsync avoids the Task allocation the byte[] overload makes per call against a FileStream opened for async IO. The equal sized chunk alignment the length comparison depends on is unchanged, and WhenAll rather than sequential awaits so a failure on one side cannot leave the other unobserved. docs/comparer.md is regenerated from the snippet. --- docs/comparer.md | 23 +++++++++++++++++++---- src/Verify/Compare/StreamComparer.cs | 21 ++++++++++++++++++--- 2 files changed, 37 insertions(+), 7 deletions(-) diff --git a/docs/comparer.md b/docs/comparer.md index 112676bdcd..63ca864ad8 100644 --- a/docs/comparer.md +++ b/docs/comparer.md @@ -211,7 +211,9 @@ The flag must be set on the source target, and that target must precede the deri ```cs -const int bufferSize = 1024 * sizeof(long); +// Large enough that a snapshot of any usual size is a couple of reads, and small enough +// to stay under the large object heap threshold and inside the array pool's buckets. +const int bufferSize = 64 * 1024; public static async Task AreEqual(Stream stream1, Stream stream2) { @@ -224,8 +226,15 @@ public static async Task AreEqual(Stream stream1, Stream stream2) { while (true) { - var count1 = await ReadBufferAsync(stream1, buffer1); - var count2 = await ReadBufferAsync(stream2, buffer2); + // The two streams are independent, so the reads overlap instead of running + // one after the other. Both files are read in full whenever they match, + // which is the usual case on a passing run. WhenAll rather than awaiting in + // sequence, so a failure on one side cannot leave the other unobserved. + var read1 = ReadBufferAsync(stream1, buffer1); + var read2 = ReadBufferAsync(stream2, buffer2); + await Task.WhenAll(read1, read2); + var count1 = read1.Result; + var count2 = read2.Result; // Callers do not always guarantee the streams are the same length // (e.g. a non-seekable received stream), so a length difference must @@ -272,7 +281,13 @@ static async Task ReadBufferAsync(Stream stream, byte[] buffer) var bytesRead = 0; while (bytesRead < bufferSize) { +#if NET6_0_OR_GREATER + // Memory overload: the byte[] overload wraps every call in a Task on a + // FileStream opened for async IO. + var read = await stream.ReadAsync(buffer.AsMemory(bytesRead, bufferSize - bytesRead)); +#else var read = await stream.ReadAsync(buffer, bytesRead, bufferSize - bytesRead); +#endif if (read == 0) { // Reached end of stream. @@ -285,7 +300,7 @@ static async Task ReadBufferAsync(Stream stream, byte[] buffer) return bytesRead; } ``` -snippet source | anchor +snippet source | anchor diff --git a/src/Verify/Compare/StreamComparer.cs b/src/Verify/Compare/StreamComparer.cs index ec8931ead0..76c918db06 100644 --- a/src/Verify/Compare/StreamComparer.cs +++ b/src/Verify/Compare/StreamComparer.cs @@ -2,7 +2,9 @@ { #region DefualtCompare - const int bufferSize = 1024 * sizeof(long); + // Large enough that a snapshot of any usual size is a couple of reads, and small enough + // to stay under the large object heap threshold and inside the array pool's buckets. + const int bufferSize = 64 * 1024; public static async Task AreEqual(Stream stream1, Stream stream2) { @@ -15,8 +17,15 @@ public static async Task AreEqual(Stream stream1, Stream stream2) { while (true) { - var count1 = await ReadBufferAsync(stream1, buffer1); - var count2 = await ReadBufferAsync(stream2, buffer2); + // The two streams are independent, so the reads overlap instead of running + // one after the other. Both files are read in full whenever they match, + // which is the usual case on a passing run. WhenAll rather than awaiting in + // sequence, so a failure on one side cannot leave the other unobserved. + var read1 = ReadBufferAsync(stream1, buffer1); + var read2 = ReadBufferAsync(stream2, buffer2); + await Task.WhenAll(read1, read2); + var count1 = read1.Result; + var count2 = read2.Result; // Callers do not always guarantee the streams are the same length // (e.g. a non-seekable received stream), so a length difference must @@ -63,7 +72,13 @@ static async Task ReadBufferAsync(Stream stream, byte[] buffer) var bytesRead = 0; while (bytesRead < bufferSize) { +#if NET6_0_OR_GREATER + // Memory overload: the byte[] overload wraps every call in a Task on a + // FileStream opened for async IO. + var read = await stream.ReadAsync(buffer.AsMemory(bytesRead, bufferSize - bytesRead)); +#else var read = await stream.ReadAsync(buffer, bytesRead, bufferSize - bytesRead); +#endif if (read == 0) { // Reached end of stream. From ea822ae1dd447411799019be76d6982b3a98567b Mon Sep 17 00:00:00 2001 From: Simon Cropp Date: Sat, 29 Aug 2026 22:59:54 +1000 Subject: [PATCH 2/2] Add benchmarks and read loop tests for the binary snapshot compare StreamComparerBenchmarks measures the previous commit against the implementation it replaced, kept in LegacyStreamComparer with the buffer size and read overload lifted to parameters. The four rungs isolate each part of the change: 8KB sequential byte[] reads, then 64KB, then the Memory overload, then the shipped overlapped version. Both sides are real async FileStreams opened the way IoHelpers.OpenRead opens them, and each size carries its own baseline so the ratios compare like for like. The 1MB compare drops to 0.13 of the old time and a ninth of the allocation, and 64KB to 0.26. At 2KB the buffer size does nothing, one read either way, so the whole 0.77 there is the overlapping. A mismatch that exits on the first chunk is 0.51. The buffered category is a MemoryStream against a FileStream, not a MemoryStream against another one. InnerCompare always opens the verified side with IoHelpers.OpenRead, so a pair of MemoryStreams never reaches the comparer, and measuring that shape reported a regression that cannot occur. Against the real shape it is 0.36. MixedEqualSpanningMultipleBuffers covers that buffered shape, which had only not-equal coverage. EqualWithShortReads covers the accumulation loop in ReadBufferAsync: a local FileStream returns the full buffer every time, so replacing that loop with a single read leaves every other test passing. A stream that always reads short keeps both sides chunk aligned under test, where a desync would surface as a spurious NotEqual on a network share. --- src/Benchmarks/LegacyStreamComparer.cs | 73 +++++++ src/Benchmarks/StreamComparerBenchmarks.cs | 242 +++++++++++++++++++++ src/Verify.Tests/StreamComparerTests.cs | 54 +++++ 3 files changed, 369 insertions(+) create mode 100644 src/Benchmarks/LegacyStreamComparer.cs create mode 100644 src/Benchmarks/StreamComparerBenchmarks.cs diff --git a/src/Benchmarks/LegacyStreamComparer.cs b/src/Benchmarks/LegacyStreamComparer.cs new file mode 100644 index 0000000000..c4d3413686 --- /dev/null +++ b/src/Benchmarks/LegacyStreamComparer.cs @@ -0,0 +1,73 @@ +using System.Buffers; + +// StreamComparer as it stood before the current branch, with the two things the branch +// changed inside ReadBufferAsync lifted to parameters, so each step can be measured on +// its own: +// * bufferSize: 1024 * sizeof(long) (8K) was the old constant, 64K is the new one +// * useMemoryOverload: the byte[] ReadAsync overload wraps every call in a Task on a +// FileStream opened for async IO, the Memory overload does not +// The reads still run one after the other, which is the part the branch replaced with +// an overlapped pair. +static class LegacyStreamComparer +{ + public static async Task AreEqual(Stream stream1, Stream stream2, int bufferSize, bool useMemoryOverload) + { + var buffer1 = ArrayPool.Shared.Rent(bufferSize); + var buffer2 = ArrayPool.Shared.Rent(bufferSize); + try + { + while (true) + { + var count1 = await ReadBufferAsync(stream1, buffer1, bufferSize, useMemoryOverload); + var count2 = await ReadBufferAsync(stream2, buffer2, bufferSize, useMemoryOverload); + + if (count1 != count2) + { + return CompareResult.NotEqual(); + } + + if (count1 == 0) + { + return CompareResult.Equal; + } + + if (!buffer1.AsSpan(0, count1).SequenceEqual(buffer2.AsSpan(0, count1))) + { + return CompareResult.NotEqual(); + } + } + } + finally + { + ArrayPool.Shared.Return(buffer1); + ArrayPool.Shared.Return(buffer2); + } + } + + static async Task ReadBufferAsync(Stream stream, byte[] buffer, int bufferSize, bool useMemoryOverload) + { + var bytesRead = 0; + while (bytesRead < bufferSize) + { + int read; + if (useMemoryOverload) + { + read = await stream.ReadAsync(buffer.AsMemory(bytesRead, bufferSize - bytesRead)); + } + else + { + read = await stream.ReadAsync(buffer, bytesRead, bufferSize - bytesRead); + } + + if (read == 0) + { + // Reached end of stream. + return bytesRead; + } + + bytesRead += read; + } + + return bytesRead; + } +} diff --git a/src/Benchmarks/StreamComparerBenchmarks.cs b/src/Benchmarks/StreamComparerBenchmarks.cs new file mode 100644 index 0000000000..708cae06b3 --- /dev/null +++ b/src/Benchmarks/StreamComparerBenchmarks.cs @@ -0,0 +1,242 @@ +using BenchmarkDotNet.Configs; + +// StreamComparer is the default comparer for every binary snapshot, and on a passing run +// both files are read end to end, so the read path is the whole cost. The branch changed +// three things, and the ladder below isolates each one: +// * Legacy_* 8K buffer, byte[] overload, sequential reads. The implementation as +// it stood on main. +// * Legacy64K_* only the buffer size changed, so a snapshot of any usual size is a +// couple of reads instead of dozens. +// * LegacyMemory_* 64K plus the Memory ReadAsync overload, which drops the Task +// the byte[] overload allocates per call on an async FileStream. +// * Current_* the shipped implementation: the above plus the two reads overlapped. +// +// The verified side is always an async FileStream, because InnerCompare opens it with +// IoHelpers.OpenRead. The received side is a FileStream in the usual case, and a +// MemoryStream when FileComparer had to buffer a non seekable one, which is the Buffered +// category. Both sides are opened once and rewound per invocation, matching FileComparer, +// which reads both streams from position 0. +// +// The OS file cache is warm after the first iteration, so these measure the async read +// machinery rather than the disk. Overlapping is worth more than this shows when the +// reads reach storage. +// +// Grouped by category so each file size carries its own baseline. A ratio against one +// shared baseline would be comparing a 2K compare to a 1M one. +[MemoryDiagnoser] +[SimpleJob(iterationCount: 10, warmupCount: 3)] +[GroupBenchmarksBy(BenchmarkLogicalGroupRule.ByCategory)] +[CategoriesColumn] +public class StreamComparerBenchmarks +{ + const int oldBufferSize = 1024 * sizeof(long); + const int newBufferSize = 64 * 1024; + + // Sized to the snapshots actually seen: a text snapshot is a couple of KB, a small + // image or serialized document lands around the new buffer size, and a large binary + // snapshot runs to megabytes. + const int smallSize = 2 * 1024; + const int mediumSize = 64 * 1024; + const int largeSize = 1024 * 1024; + + string directory = null!; + + Pair small = null!; + Pair medium = null!; + Pair large = null!; + Pair mismatch = null!; + + MemoryStream buffered = null!; + + [GlobalSetup] + public void Setup() + { + directory = Path.Combine(Path.GetTempPath(), "VerifyStreamComparerBenchmarks"); + if (Directory.Exists(directory)) + { + Directory.Delete(directory, true); + } + + Directory.CreateDirectory(directory); + + small = BuildPair("small", smallSize, differAtStart: false); + medium = BuildPair("medium", mediumSize, differAtStart: false); + large = BuildPair("large", largeSize, differAtStart: false); + + // Same length, first byte differs. FileComparer short circuits on a length + // difference, so a mismatch that reaches here is usually an equal length one. + mismatch = BuildPair("mismatch", largeSize, differAtStart: true); + + // FileComparer buffers a non seekable received stream into a MemoryStream, then + // compares it against the verified file. So the received side reads synchronously + // while the verified side is still async file IO. A MemoryStream on both sides + // never reaches StreamComparer: InnerCompare always opens the verified side with + // IoHelpers.OpenRead. Same content as the medium pair, so this compares equal + // against that pair's verified file. + buffered = new(BuildContent(mediumSize, seed: mediumSize)); + } + + [GlobalCleanup] + public void Cleanup() + { + small.Dispose(); + medium.Dispose(); + large.Dispose(); + mismatch.Dispose(); + buffered.Dispose(); + Directory.Delete(directory, true); + } + + [BenchmarkCategory("Small")] + [Benchmark(Baseline = true)] + public async Task Legacy_Small() => + (await LegacyStreamComparer.AreEqual(small.Rewind(), small.Verified, oldBufferSize, useMemoryOverload: false)).IsEqual; + + [BenchmarkCategory("Small")] + [Benchmark] + public async Task Legacy64K_Small() => + (await LegacyStreamComparer.AreEqual(small.Rewind(), small.Verified, newBufferSize, useMemoryOverload: false)).IsEqual; + + [BenchmarkCategory("Small")] + [Benchmark] + public async Task LegacyMemory_Small() => + (await LegacyStreamComparer.AreEqual(small.Rewind(), small.Verified, newBufferSize, useMemoryOverload: true)).IsEqual; + + [BenchmarkCategory("Small")] + [Benchmark] + public async Task Current_Small() => + (await StreamComparer.AreEqual(small.Rewind(), small.Verified)).IsEqual; + + [BenchmarkCategory("Medium")] + [Benchmark(Baseline = true)] + public async Task Legacy_Medium() => + (await LegacyStreamComparer.AreEqual(medium.Rewind(), medium.Verified, oldBufferSize, useMemoryOverload: false)).IsEqual; + + [BenchmarkCategory("Medium")] + [Benchmark] + public async Task Legacy64K_Medium() => + (await LegacyStreamComparer.AreEqual(medium.Rewind(), medium.Verified, newBufferSize, useMemoryOverload: false)).IsEqual; + + [BenchmarkCategory("Medium")] + [Benchmark] + public async Task LegacyMemory_Medium() => + (await LegacyStreamComparer.AreEqual(medium.Rewind(), medium.Verified, newBufferSize, useMemoryOverload: true)).IsEqual; + + [BenchmarkCategory("Medium")] + [Benchmark] + public async Task Current_Medium() => + (await StreamComparer.AreEqual(medium.Rewind(), medium.Verified)).IsEqual; + + [BenchmarkCategory("Large")] + [Benchmark(Baseline = true)] + public async Task Legacy_Large() => + (await LegacyStreamComparer.AreEqual(large.Rewind(), large.Verified, oldBufferSize, useMemoryOverload: false)).IsEqual; + + [BenchmarkCategory("Large")] + [Benchmark] + public async Task Legacy64K_Large() => + (await LegacyStreamComparer.AreEqual(large.Rewind(), large.Verified, newBufferSize, useMemoryOverload: false)).IsEqual; + + [BenchmarkCategory("Large")] + [Benchmark] + public async Task LegacyMemory_Large() => + (await LegacyStreamComparer.AreEqual(large.Rewind(), large.Verified, newBufferSize, useMemoryOverload: true)).IsEqual; + + [BenchmarkCategory("Large")] + [Benchmark] + public async Task Current_Large() => + (await StreamComparer.AreEqual(large.Rewind(), large.Verified)).IsEqual; + + // A failing run: both sides are read once and the compare stops at the first chunk. + [BenchmarkCategory("NotEqual")] + [Benchmark(Baseline = true)] + public async Task Legacy_Large_NotEqual() => + (await LegacyStreamComparer.AreEqual(mismatch.Rewind(), mismatch.Verified, oldBufferSize, useMemoryOverload: false)).IsEqual; + + [BenchmarkCategory("NotEqual")] + [Benchmark] + public async Task Current_Large_NotEqual() => + (await StreamComparer.AreEqual(mismatch.Rewind(), mismatch.Verified)).IsEqual; + + // The received side reads synchronously, so its read completes inline before the + // verified read is even issued. The overlapping has nothing to hide behind here. + [BenchmarkCategory("Buffered")] + [Benchmark(Baseline = true)] + public async Task Legacy_Medium_Buffered() => + (await LegacyStreamComparer.AreEqual(RewindBuffered(), medium.Verified, oldBufferSize, useMemoryOverload: false)).IsEqual; + + [BenchmarkCategory("Buffered")] + [Benchmark] + public async Task Legacy64K_Medium_Buffered() => + (await LegacyStreamComparer.AreEqual(RewindBuffered(), medium.Verified, newBufferSize, useMemoryOverload: false)).IsEqual; + + [BenchmarkCategory("Buffered")] + [Benchmark] + public async Task LegacyMemory_Medium_Buffered() => + (await LegacyStreamComparer.AreEqual(RewindBuffered(), medium.Verified, newBufferSize, useMemoryOverload: true)).IsEqual; + + [BenchmarkCategory("Buffered")] + [Benchmark] + public async Task Current_Medium_Buffered() => + (await StreamComparer.AreEqual(RewindBuffered(), medium.Verified)).IsEqual; + + MemoryStream RewindBuffered() + { + buffered.Position = 0; + medium.Verified.Position = 0; + return buffered; + } + + Pair BuildPair(string name, int size, bool differAtStart) + { + var content = BuildContent(size, seed: size); + var receivedPath = Path.Combine(directory, name + ".received.bin"); + var verifiedPath = Path.Combine(directory, name + ".verified.bin"); + File.WriteAllBytes(verifiedPath, content); + + if (differAtStart) + { + var copy = (byte[]) content.Clone(); + copy[0] ^= 0xFF; + File.WriteAllBytes(receivedPath, copy); + } + else + { + File.WriteAllBytes(receivedPath, content); + } + + return new(Open(receivedPath), Open(verifiedPath)); + } + + // Matches IoHelpers.OpenRead, which is how the verified side is opened in production. + static FileStream Open(string path) => + new(path, FileMode.Open, FileAccess.Read, FileShare.Read, bufferSize: 4096, useAsync: true); + + static byte[] BuildContent(int size, int seed) + { + var content = new byte[size]; + new Random(seed).NextBytes(content); + return content; + } + + sealed class Pair(FileStream received, FileStream verified) : + IDisposable + { + public FileStream Verified { get; } = verified; + + // StreamComparer requires both streams at position 0. Rewinding costs the same + // for every variant, so it does not skew the comparison. + public FileStream Rewind() + { + received.Position = 0; + Verified.Position = 0; + return received; + } + + public void Dispose() + { + received.Dispose(); + Verified.Dispose(); + } + } +} diff --git a/src/Verify.Tests/StreamComparerTests.cs b/src/Verify.Tests/StreamComparerTests.cs index 757bd64422..8d00eae533 100644 --- a/src/Verify.Tests/StreamComparerTests.cs +++ b/src/Verify.Tests/StreamComparerTests.cs @@ -28,6 +28,60 @@ public async Task EqualWithLengthNotMultipleOfEight() Assert.True(result.IsEqual); } + [Fact] + public async Task MixedEqualSpanningMultipleBuffers() + { + // The shape FileComparer produces for a non-seekable received stream: it is + // buffered into a MemoryStream and compared against the verified file, which is + // always opened for async IO. So one side completes every read inline and the + // other does not. Large enough to span several buffers, and deliberately not a + // multiple of the buffer size, so the final block is partial. + var bytes = new byte[256 * 1024 + 13]; + new Random(1).NextBytes(bytes); + + var path = Path.Combine(Path.GetTempPath(), $"StreamComparerTests_{Guid.NewGuid():N}.bin"); + File.WriteAllBytes(path, bytes); + try + { + using var received = new MemoryStream((byte[]) bytes.Clone()); + // ReSharper disable once UseAwaitUsing + using var verified = IoHelpers.OpenRead(path); + var result = await StreamComparer.AreEqual(received, verified); + Assert.True(result.IsEqual); + } + finally + { + File.Delete(path); + } + } + + [Fact] + public async Task EqualWithShortReads() + { + // A stream is free to return fewer bytes than asked for, and ReadBufferAsync + // accumulates until the buffer is full so both sides stay chunk aligned. A real + // FileStream rarely short reads, so it takes a stream that always does to cover + // that loop. Both sides are equal, so a mis-aligned chunk would surface as a + // spurious NotEqual. + var bytes = new byte[256 * 1024 + 13]; + new Random(2).NextBytes(bytes); + + using var received = new ShortReadStream(bytes, maxRead: 1023); + using var verified = new ShortReadStream((byte[]) bytes.Clone(), maxRead: 337); + var result = await StreamComparer.AreEqual(received, verified); + Assert.True(result.IsEqual); + } + + // Returns at most maxRead bytes per call, regardless of how many are asked for. Only + // the byte[] overload needs overriding: because this is a derived type, MemoryStream + // routes the span and Memory reads back through it rather than using its fast path. + class ShortReadStream(byte[] bytes, int maxRead) : + MemoryStream(bytes) + { + public override int Read(byte[] buffer, int offset, int count) => + base.Read(buffer, offset, Math.Min(count, maxRead)); + } + [Fact] public async Task NotEqualInPartialFinalBlock() {