diff --git a/.github/workflows/build-samples.yml b/.github/workflows/build-samples.yml index cca3564..71b1c32 100644 --- a/.github/workflows/build-samples.yml +++ b/.github/workflows/build-samples.yml @@ -21,6 +21,7 @@ on: - "cloud-native/health-resilience-zero-downtime/**" - "cloud-native/polly-resilience/**" - "csharp-language/csharp-12-features/**" + - "csharp-language/high-performance-memory-management/**" - ".github/workflows/build-samples.yml" pull_request: @@ -42,6 +43,7 @@ on: - "cloud-native/health-resilience-zero-downtime/**" - "cloud-native/polly-resilience/**" - "csharp-language/csharp-12-features/**" + - "csharp-language/high-performance-memory-management/**" - ".github/workflows/build-samples.yml" workflow_dispatch: @@ -794,3 +796,101 @@ jobs: )" test "${line_count}" = "7" + + test-high-performance-pipelines: + name: Test UTF-8 pipelines log ingestor + runs-on: ubuntu-latest + + permissions: + contents: read + + steps: + - name: Check out repository + uses: actions/checkout@v5 + + - name: Install .NET 10 SDK + uses: actions/setup-dotnet@v5 + with: + dotnet-version: "10.0.x" + + - name: Restore + run: > + dotnet restore + csharp-language/high-performance-memory-management/HighPerformanceLogIngestor.slnx + + - name: Build + run: > + dotnet build + csharp-language/high-performance-memory-management/HighPerformanceLogIngestor.slnx + --configuration Release + --no-restore + + - name: Test + run: > + dotnet test + csharp-language/high-performance-memory-management/HighPerformanceLogIngestor.slnx + --configuration Release + --no-build + + - name: Verify explicit C# language version + shell: bash + run: | + app_version="$( + dotnet msbuild \ + csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/HighPerformanceLogIngestor.csproj \ + -getProperty:LangVersion + )" + + test_version="$( + dotnet msbuild \ + csharp-language/high-performance-memory-management/tests/HighPerformanceLogIngestor.Tests/HighPerformanceLogIngestor.Tests.csproj \ + -getProperty:LangVersion + )" + + echo "Application LangVersion: ${app_version}" + echo "Test LangVersion: ${test_version}" + + test "${app_version}" = "12.0" + test "${test_version}" = "12.0" + + - name: Run deterministic ingestion sample + shell: bash + run: | + output="$( + dotnet run \ + --project csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/HighPerformanceLogIngestor.csproj \ + --configuration Release \ + --no-build + )" + + printf '%s\n' "${output}" + + printf -v expected '%s\n' \ + 'High-Performance UTF-8 Log Ingestor' \ + 'Lines: total=4, valid=3, invalid=1' \ + 'Accepted event IDs: 1001, 1002, 1004' \ + 'Levels: 1=1, 2=1, 4=1' \ + 'First message: Cache warmed' \ + 'Last message: Worker started' + + expected="${expected%$'\n'}" + + if [[ "${output}" != "${expected}" ]]; then + echo "Console output did not match the expected output." + + diff \ + --unified \ + <(printf '%s\n' "${expected}") \ + <(printf '%s\n' "${output}") \ + || true + + exit 1 + fi + + line_count="$( + printf '%s\n' "${output}" | + wc --lines | + tr --delete ' ' + )" + + test "${line_count}" = "6" diff --git a/README.md b/README.md index 5feb6c9..4a25b50 100644 --- a/README.md +++ b/README.md @@ -26,6 +26,7 @@ Each sample folder contains a focused implementation of one tutorial topic. The | [`cloud-native/health-resilience-zero-downtime`](cloud-native/health-resilience-zero-downtime/) | Focused .NET 10 Orders API demonstrating tagged liveness/readiness checks, structured health output, shutdown readiness draining, typed HttpClient retries, circuit breaking, attempt timeouts, and deterministic integration testing | [.NET 8 Cloud-Native: Health Probes, Polly Resilience & Zero-Downtime Kubernetes Deployments](https://www.dotnet-guide.com/tutorials/cloud-native/health-resilience-zero-downtime/) | | [`cloud-native/polly-resilience`](cloud-native/polly-resilience/) | Focused .NET 10 Polly v8 catalog sample demonstrating explicit stale-cache fallback, timeout-driven degradation, outbound concurrency isolation, strategy event counters, and deterministic integration testing | [Polly Resilience (Polly v8): Timeouts, Retries, Circuits, Bulkheads, Hedging](https://www.dotnet-guide.com/tutorials/cloud-native/polly-resilience/) | | [`csharp-language/csharp-12-features`](csharp-language/csharp-12-features/) | Focused .NET 10 console lab locked to C# 12.0, demonstrating primary constructors with explicit properties, collection expressions and spreads, default lambda parameters, tuple-type aliases, deterministic output, and unit tests | [C# 12 Language Features: Primary Constructors, Collections & More](https://www.dotnet-guide.com/tutorials/csharp-language/csharp-12-features/) | +| [`csharp-language/high-performance-memory-management`](csharp-language/high-performance-memory-management/) | Focused .NET 10 UTF-8 ingestion sample demonstrating PipeReader framing, segmented ReadOnlySequence handling, span-based numeric parsing, bounded MemoryPool ownership, strict input validation, deterministic output, and tests | [High-Performance C#: Span, Memory, SIMD & Pipelines](https://www.dotnet-guide.com/tutorials/csharp-language/high-performance-memory-management/) | ## Companion articles - [Common Microsoft.Extensions.AI mistakes](https://www.dotnet-guide.com/articles/dotnet-ai/microsoft-extensions-ai-common-mistakes/) @@ -269,23 +270,45 @@ tutorials/ | |-- PollyCatalogResilience.Tests.csproj | `-- CatalogResilienceTests.cs |-- csharp-language/ -| `-- csharp-12-features/ -| |-- CSharp12RefactoringLab.slnx -| |-- README.md -| |-- src/ -| | `-- CSharp12RefactoringLab/ -| | |-- CSharp12RefactoringLab.csproj -| | |-- Program.cs -| | |-- Formatting/ -| | | `-- TodoFormatter.cs -| | |-- Models/ -| | | `-- TodoItem.cs -| | `-- Services/ -| | `-- TodoService.cs -| `-- tests/ -| `-- CSharp12RefactoringLab.Tests/ -| |-- CSharp12RefactoringLab.Tests.csproj -| `-- CSharp12FeatureTests.cs +| |-- csharp-12-features/ +| | |-- CSharp12RefactoringLab.slnx +| | |-- README.md +| | |-- src/ +| | | `-- CSharp12RefactoringLab/ +| | | |-- CSharp12RefactoringLab.csproj +| | | |-- Program.cs +| | | |-- Formatting/ +| | | | `-- TodoFormatter.cs +| | | |-- Models/ +| | | | `-- TodoItem.cs +| | | `-- Services/ +| | | `-- TodoService.cs +| | `-- tests/ +| | `-- CSharp12RefactoringLab.Tests/ +| | |-- CSharp12RefactoringLab.Tests.csproj +| | `-- CSharp12FeatureTests.cs +| |-- high-performance-memory-management/ +| | |-- HighPerformanceLogIngestor.slnx +| | |-- README.md +| | |-- src/ +| | | `-- HighPerformanceLogIngestor/ +| | | |-- HighPerformanceLogIngestor.csproj +| | | |-- Program.cs +| | | |-- Models/ +| | | | `-- LogEntry.cs +| | | |-- Parsing/ +| | | | |-- LogLineDecoder.cs +| | | | `-- Utf8LogLineParser.cs +| | | `-- Pipelines/ +| | | |-- LogIngestionResult.cs +| | | `-- LogIngestor.cs +| | `-- tests/ +| | `-- HighPerformanceLogIngestor.Tests/ +| | |-- HighPerformanceLogIngestor.Tests.csproj +| | |-- HighPerformanceLogTests.cs +| | `-- TestSupport/ +| | |-- ChunkedReadStream.cs +| | `-- SegmentedSequence.cs |-- aspnet-core/ | |-- api-security-in-practice/ | | |-- ApiSecurityMinimal.slnx diff --git a/csharp-language/high-performance-memory-management/HighPerformanceLogIngestor.slnx b/csharp-language/high-performance-memory-management/HighPerformanceLogIngestor.slnx new file mode 100644 index 0000000..53684b7 --- /dev/null +++ b/csharp-language/high-performance-memory-management/HighPerformanceLogIngestor.slnx @@ -0,0 +1,8 @@ + + + + + + + + diff --git a/csharp-language/high-performance-memory-management/README.md b/csharp-language/high-performance-memory-management/README.md new file mode 100644 index 0000000..2cc15ca --- /dev/null +++ b/csharp-language/high-performance-memory-management/README.md @@ -0,0 +1,226 @@ +# UTF-8 Pipelines Log Ingestor + +A focused .NET 10 companion demonstrating newline framing with `PipeReader`, +segmented `ReadOnlySequence` handling, span-based numeric parsing, +bounded pooled copies, explicit buffer ownership, and deterministic tests. + +## Full tutorial + +[High-Performance C#: Span, Memory, SIMD & Pipelines](https://www.dotnet-guide.com/tutorials/csharp-language/high-performance-memory-management/) + +## Framework and language note + +The tutorial is framed around .NET 8. + +This companion targets .NET 10 because that is the DOTNET GUIDE repository's +current SDK, but both projects explicitly set C# 12.0. + +The parser keeps `ReadOnlySpan` work in synchronous methods and does not +depend on C# 13's relaxed ref-struct rules for async methods. + +## Input format + +```text +||| +``` + +Example: + +```text +1785924000|2|1001|Cache warmed +``` + +This is a deliberately small pipe-delimited log format. + +It is not a complete CSV or NDJSON implementation. + +## Pipeline flow + +```text +Stream + -> PipeReader + -> ReadOnlySequence + -> newline framing + -> direct single-segment parse + or bounded pooled multi-segment copy + -> Utf8Parser + -> validated LogEntry +``` + +## Single-segment path + +When a complete line occupies one sequence segment, the decoder passes +`FirstSpan` directly to the parser. + +No line-copy buffer is created for that path. + +The valid result still owns a message string and a `LogEntry`. + +## Multi-segment path + +When a line crosses sequence segments: + +1. reject it if it exceeds 4,096 bytes; +2. rent from `MemoryPool.Shared`; +3. use only the requested memory prefix; +4. copy the logical line; +5. parse the copied span; +6. dispose `IMemoryOwner` before returning. + +No pooled reference escapes. + +## Parsing rules + +- all three numeric fields must be consumed completely; +- level must be from 0 through 5; +- event ID must be positive; +- the message must be non-empty valid UTF-8; +- LF and CRLF are supported; +- the last line can omit a trailing newline; +- malformed lines are counted and skipped. + +## Allocation boundary + +This sample does not claim zero allocation. + +A valid result creates an owned message string and `LogEntry`. + +The sample avoids intermediate strings for numeric parsing and avoids copying +contiguous logical lines. + +Measure the real workload before making performance claims. + +## Stream ownership + +The ingestor creates and completes its `PipeReader` but configures it to leave +the caller's stream open. + +Caller cancellation propagates as cancellation. + +## Oversized-record handling + +The ingestor enforces the 4,096-byte logical line limit *before* a newline +arrives, not just when the decoder receives a complete line. + +On each `PipeReader` read, after all complete LF-delimited lines are extracted, +the ingestor examines the length of the current retained, unterminated +`ReadOnlySequence`. Because the sequence already includes bytes retained +from earlier reads, its length is not accumulated again. + +When the retained sequence exceeds 4,097 bytes (4,096 plus one optional CR +byte): + +1. the record is counted exactly once (total and invalid); +2. the ingestor enters discard mode; +3. incoming bytes are consumed but not retained; +4. after the next LF, normal parsing resumes with the following record. + +This prevents an unterminated or oversized record from causing unbounded +retained `PipeReader` data. The sample's deterministic output does not +include oversized records because the sample data stays within the limit. + +## Pooled-copy accounting + +`PooledCopies` counts every logical line that required the multi-segment +pooled-copy path, including malformed lines where the copy was performed +before the decoder determined the line was invalid. + +## Application dependencies + +The application targets `net10.0` and uses only types from the .NET 10 +shared framework. It has no direct NuGet package references. + +`System.IO.Pipelines` is provided by the .NET 10 target framework and does +not need a pinned package reference for this sample. + +## Prerequisite + +- .NET 10 SDK + +## Restore, build, test, and run + +```powershell +dotnet restore ` + .\HighPerformanceLogIngestor.slnx + +dotnet build ` + .\HighPerformanceLogIngestor.slnx ` + --configuration Release ` + --no-restore + +dotnet test ` + .\HighPerformanceLogIngestor.slnx ` + --configuration Release ` + --no-build + +dotnet run ` + --project .\src\HighPerformanceLogIngestor\HighPerformanceLogIngestor.csproj ` + --configuration Release ` + --no-build +``` + +## Expected output + +```text +High-Performance UTF-8 Log Ingestor +Lines: total=4, valid=3, invalid=1 +Accepted event IDs: 1001, 1002, 1004 +Levels: 1=1, 2=1, 4=1 +First message: Cache warmed +Last message: Worker started +``` + +## Project structure + +```text +HighPerformanceLogIngestor.slnx +README.md +src/ +`-- HighPerformanceLogIngestor/ + |-- HighPerformanceLogIngestor.csproj + |-- Program.cs + |-- Models/ + | `-- LogEntry.cs + |-- Parsing/ + | |-- LogLineDecoder.cs + | `-- Utf8LogLineParser.cs + `-- Pipelines/ + |-- LogIngestionResult.cs + `-- LogIngestor.cs +tests/ +`-- HighPerformanceLogIngestor.Tests/ + |-- HighPerformanceLogIngestor.Tests.csproj + |-- HighPerformanceLogTests.cs + `-- TestSupport/ + |-- ChunkedReadStream.cs + `-- SegmentedSequence.cs +``` + +## Deliberately omitted + +- stack allocation; +- unsafe code; +- `MemoryMarshal`; +- SIMD; +- hardware intrinsics; +- BenchmarkDotNet; +- ASP.NET Core; +- NDJSON; +- full CSV quoting; +- performance claims. + +These remain in the complete tutorial or need dedicated measured samples. + +## Verification + +- Companion target framework: .NET 10 +- Explicit language version: C# 12.0 +- System.IO.Pipelines: provided by the .NET 10 shared framework +- NuGet package dependencies: none +- Maximum logical line length: 4,096 bytes +- External services required: none +- Expected tests: 8 +- Last reviewed: 2026-08-05 + +This sample demonstrates ownership and parsing semantics. It does not establish +production throughput, allocation rates, or hardware-specific speedups. \ No newline at end of file diff --git a/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/HighPerformanceLogIngestor.csproj b/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/HighPerformanceLogIngestor.csproj new file mode 100644 index 0000000..27eb670 --- /dev/null +++ b/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/HighPerformanceLogIngestor.csproj @@ -0,0 +1,12 @@ + + + + Exe + net10.0 + 12.0 + enable + enable + true + + + \ No newline at end of file diff --git a/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/Models/LogEntry.cs b/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/Models/LogEntry.cs new file mode 100644 index 0000000..d0daf63 --- /dev/null +++ b/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/Models/LogEntry.cs @@ -0,0 +1,7 @@ +namespace HighPerformanceLogIngestor.Models; + +public sealed record LogEntry( + DateTimeOffset Timestamp, + byte Level, + int EventId, + string Message); \ No newline at end of file diff --git a/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/Parsing/LogLineDecoder.cs b/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/Parsing/LogLineDecoder.cs new file mode 100644 index 0000000..2c42cfa --- /dev/null +++ b/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/Parsing/LogLineDecoder.cs @@ -0,0 +1,90 @@ +using System.Buffers; +using HighPerformanceLogIngestor.Models; + +namespace HighPerformanceLogIngestor.Parsing; + +public sealed class LogLineDecoder( + Utf8LogLineParser parser) +{ + public const int MaximumLineLength = + 4_096; + + public bool TryDecode( + ReadOnlySequence line, + out LogEntry? entry, + out bool usedPooledCopy) + { + entry = + null; + + usedPooledCopy = + false; + + line = + TrimTrailingCarriageReturn( + line); + + if (line.IsEmpty + || + line.Length + > MaximumLineLength) + { + return false; + } + + if (line.IsSingleSegment) + { + return parser.TryParse( + line.FirstSpan, + out entry); + } + + int length = + checked( + (int)line.Length); + + using IMemoryOwner owner = + MemoryPool + .Shared + .Rent( + length); + + Span destination = + owner.Memory + .Span[ + ..length]; + + line.CopyTo( + destination); + + usedPooledCopy = + true; + + return parser.TryParse( + destination, + out entry); + } + + private static ReadOnlySequence + TrimTrailingCarriageReturn( + ReadOnlySequence line) + { + if (line.IsEmpty) + { + return line; + } + + ReadOnlySequence finalByte = + line.Slice( + line.Length + - 1); + + return finalByte.FirstSpan[0] + == (byte)'\r' + ? line.Slice( + 0, + line.Length + - 1) + : line; + } +} \ No newline at end of file diff --git a/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/Parsing/Utf8LogLineParser.cs b/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/Parsing/Utf8LogLineParser.cs new file mode 100644 index 0000000..4d121c8 --- /dev/null +++ b/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/Parsing/Utf8LogLineParser.cs @@ -0,0 +1,200 @@ +using System.Buffers; +using System.Buffers.Text; +using System.Text; +using HighPerformanceLogIngestor.Models; + +namespace HighPerformanceLogIngestor.Parsing; + +public sealed class Utf8LogLineParser +{ + private const long MinimumUnixSeconds = + -62_135_596_800; + + private const long MaximumUnixSeconds = + 253_402_300_799; + + public bool TryParse( + ReadOnlySpan line, + out LogEntry? entry) + { + entry = + null; + + int firstDelimiter = + line.IndexOf( + (byte)'|'); + + if (firstDelimiter <= 0) + { + return false; + } + + ReadOnlySpan remainder = + line[ + (firstDelimiter + 1)..]; + + int secondRelative = + remainder.IndexOf( + (byte)'|'); + + if (secondRelative <= 0) + { + return false; + } + + int secondDelimiter = + firstDelimiter + + 1 + + secondRelative; + + remainder = + line[ + (secondDelimiter + 1)..]; + + int thirdRelative = + remainder.IndexOf( + (byte)'|'); + + if (thirdRelative <= 0) + { + return false; + } + + int thirdDelimiter = + secondDelimiter + + 1 + + thirdRelative; + + ReadOnlySpan timestampBytes = + line[ + ..firstDelimiter]; + + ReadOnlySpan levelBytes = + line[ + (firstDelimiter + 1) + ..secondDelimiter]; + + ReadOnlySpan eventIdBytes = + line[ + (secondDelimiter + 1) + ..thirdDelimiter]; + + ReadOnlySpan messageBytes = + line[ + (thirdDelimiter + 1)..]; + + if (!TryParseInt64Exact( + timestampBytes, + out long unixSeconds) + || + unixSeconds + is < MinimumUnixSeconds + or > MaximumUnixSeconds) + { + return false; + } + + if (!TryParseByteExact( + levelBytes, + out byte level) + || + level > 5) + { + return false; + } + + if (!TryParseInt32Exact( + eventIdBytes, + out int eventId) + || + eventId <= 0) + { + return false; + } + + if (messageBytes.IsEmpty + || + !IsValidUtf8( + messageBytes)) + { + return false; + } + + entry = + new LogEntry( + Timestamp: + DateTimeOffset + .FromUnixTimeSeconds( + unixSeconds), + + Level: + level, + + EventId: + eventId, + + Message: + Encoding.UTF8 + .GetString( + messageBytes)); + + return true; + } + + private static bool TryParseInt64Exact( + ReadOnlySpan value, + out long result) => + Utf8Parser.TryParse( + value, + out result, + out int consumed) + && + consumed + == value.Length; + + private static bool TryParseInt32Exact( + ReadOnlySpan value, + out int result) => + Utf8Parser.TryParse( + value, + out result, + out int consumed) + && + consumed + == value.Length; + + private static bool TryParseByteExact( + ReadOnlySpan value, + out byte result) => + Utf8Parser.TryParse( + value, + out result, + out int consumed) + && + consumed + == value.Length; + + private static bool IsValidUtf8( + ReadOnlySpan value) + { + while (!value.IsEmpty) + { + OperationStatus status = + Rune.DecodeFromUtf8( + value, + out _, + out int consumed); + + if (status + != OperationStatus.Done) + { + return false; + } + + value = + value[consumed..]; + } + + return true; + } +} \ No newline at end of file diff --git a/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/Pipelines/LogIngestionResult.cs b/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/Pipelines/LogIngestionResult.cs new file mode 100644 index 0000000..78f6dbf --- /dev/null +++ b/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/Pipelines/LogIngestionResult.cs @@ -0,0 +1,7 @@ +namespace HighPerformanceLogIngestor.Pipelines; + +public sealed record LogIngestionResult( + int TotalLines, + int ValidLines, + int InvalidLines, + int PooledCopies); \ No newline at end of file diff --git a/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/Pipelines/LogIngestor.cs b/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/Pipelines/LogIngestor.cs new file mode 100644 index 0000000..c908427 --- /dev/null +++ b/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/Pipelines/LogIngestor.cs @@ -0,0 +1,256 @@ +using System.Buffers; +using System.IO.Pipelines; +using HighPerformanceLogIngestor.Models; +using HighPerformanceLogIngestor.Parsing; + +namespace HighPerformanceLogIngestor.Pipelines; + +public sealed class LogIngestor( + LogLineDecoder decoder) +{ + // The maximum bytes of an unterminated record retained while waiting + // for a newline before triggering oversized-record discard. + // Set to MaximumLineLength + 1 so a valid maximum-length record + // can be followed by CR before LF. + private const int MaxRetainedBeforeNewline = + LogLineDecoder.MaximumLineLength + + 1; + + public async Task + IngestAsync( + Stream stream, + Action onEntry, + CancellationToken cancellationToken = + default) + { + ArgumentNullException + .ThrowIfNull( + stream); + + ArgumentNullException + .ThrowIfNull( + onEntry); + + if (!stream.CanRead) + { + throw new ArgumentException( + "The stream must be readable.", + nameof(stream)); + } + + var options = + new StreamPipeReaderOptions( + leaveOpen: + true); + + PipeReader reader = + PipeReader.Create( + stream, + options); + + int totalLines = + 0; + + int validLines = + 0; + + int invalidLines = + 0; + + int pooledCopies = + 0; + + bool discardingOversized = + false; + + Exception? completionError = + null; + + try + { + while (true) + { + ReadResult result = + await reader.ReadAsync( + cancellationToken); + + ReadOnlySequence buffer = + result.Buffer; + + if (result.IsCanceled) + { + reader.AdvanceTo( + buffer.Start, + buffer.End); + + cancellationToken + .ThrowIfCancellationRequested(); + + throw new OperationCanceledException( + "The pipeline read was canceled."); + } + + while (TryReadLine( + ref buffer, + out ReadOnlySequence + line)) + { + if (discardingOversized) + { + // This LF ends the oversized-discard cycle. + discardingOversized = + false; + + continue; + } + + ProcessLine( + line); + } + + if (discardingOversized) + { + if (result.IsCompleted) + { + // Stream ended without an LF; the oversized + // record was already counted once. + reader.AdvanceTo( + result.Buffer.End); + + break; + } + + // Consume all bytes and keep looking for LF. + reader.AdvanceTo( + buffer.End); + + continue; + } + + // Examine the current retained, unterminated sequence. + // The sequence already includes bytes retained from + // earlier reads, so its length is not accumulated again. + if (!buffer.IsEmpty + && buffer.Length + > MaxRetainedBeforeNewline) + { + totalLines++; + invalidLines++; + + discardingOversized = + true; + + reader.AdvanceTo( + buffer.End); + + continue; + } + + if (result.IsCompleted) + { + if (!buffer.IsEmpty) + { + ProcessLine( + buffer); + } + + reader.AdvanceTo( + result.Buffer.End); + + break; + } + + reader.AdvanceTo( + buffer.Start, + buffer.End); + } + } + catch (Exception exception) + { + completionError = + exception; + + throw; + } + finally + { + await reader.CompleteAsync( + completionError); + } + + return new LogIngestionResult( + TotalLines: + totalLines, + + ValidLines: + validLines, + + InvalidLines: + invalidLines, + + PooledCopies: + pooledCopies); + + void ProcessLine( + ReadOnlySequence line) + { + totalLines++; + + bool decoded = + decoder.TryDecode( + line, + out LogEntry? entry, + out bool usedPooledCopy); + + if (usedPooledCopy) + { + pooledCopies++; + } + + if (decoded) + { + validLines++; + + onEntry( + entry!); + } + else + { + invalidLines++; + } + } + } + + private static bool TryReadLine( + ref ReadOnlySequence buffer, + out ReadOnlySequence line) + { + SequencePosition? newline = + buffer.PositionOf( + (byte)'\n'); + + if (newline is null) + { + line = + default; + + return false; + } + + line = + buffer.Slice( + 0, + newline.Value); + + SequencePosition next = + buffer.GetPosition( + 1, + newline.Value); + + buffer = + buffer.Slice( + next); + + return true; + } +} \ No newline at end of file diff --git a/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/Program.cs b/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/Program.cs new file mode 100644 index 0000000..2413a3f --- /dev/null +++ b/csharp-language/high-performance-memory-management/src/HighPerformanceLogIngestor/Program.cs @@ -0,0 +1,81 @@ +using System.Text; +using HighPerformanceLogIngestor.Models; +using HighPerformanceLogIngestor.Parsing; +using HighPerformanceLogIngestor.Pipelines; + +const string SampleData = + """ + 1785924000|2|1001|Cache warmed + 1785924001|4|1002|Request failed + not-a-timestamp|2|1003|Malformed timestamp + 1785924002|1|1004|Worker started + """; + +byte[] utf8Data = + Encoding.UTF8.GetBytes( + SampleData); + +using var stream = + new MemoryStream( + utf8Data, + writable: + false); + +var parser = + new Utf8LogLineParser(); + +var decoder = + new LogLineDecoder( + parser); + +var ingestor = + new LogIngestor( + decoder); + +List accepted = +[ +]; + +LogIngestionResult result = + await ingestor.IngestAsync( + stream, + accepted.Add); + +string eventIds = + string.Join( + ", ", + accepted.Select( + entry => + entry.EventId)); + +string levels = + string.Join( + ", ", + accepted + .GroupBy( + entry => + entry.Level) + .OrderBy( + group => + group.Key) + .Select( + group => + $"{group.Key}={group.Count()}")); + +Console.WriteLine( + "High-Performance UTF-8 Log Ingestor"); + +Console.WriteLine( + $"Lines: total={result.TotalLines}, valid={result.ValidLines}, invalid={result.InvalidLines}"); + +Console.WriteLine( + $"Accepted event IDs: {eventIds}"); + +Console.WriteLine( + $"Levels: {levels}"); + +Console.WriteLine( + $"First message: {accepted[0].Message}"); + +Console.WriteLine( + $"Last message: {accepted[^1].Message}"); \ No newline at end of file diff --git a/csharp-language/high-performance-memory-management/tests/HighPerformanceLogIngestor.Tests/HighPerformanceLogIngestor.Tests.csproj b/csharp-language/high-performance-memory-management/tests/HighPerformanceLogIngestor.Tests/HighPerformanceLogIngestor.Tests.csproj new file mode 100644 index 0000000..c2f99f4 --- /dev/null +++ b/csharp-language/high-performance-memory-management/tests/HighPerformanceLogIngestor.Tests/HighPerformanceLogIngestor.Tests.csproj @@ -0,0 +1,47 @@ + + + + net10.0 + 12.0 + enable + enable + true + false + true + Exe + + + + + + + + + all + + runtime; + build; + native; + contentfiles; + analyzers; + buildtransitive + + + + + + + + + + + + + \ No newline at end of file diff --git a/csharp-language/high-performance-memory-management/tests/HighPerformanceLogIngestor.Tests/HighPerformanceLogTests.cs b/csharp-language/high-performance-memory-management/tests/HighPerformanceLogIngestor.Tests/HighPerformanceLogTests.cs new file mode 100644 index 0000000..4d852bf --- /dev/null +++ b/csharp-language/high-performance-memory-management/tests/HighPerformanceLogIngestor.Tests/HighPerformanceLogTests.cs @@ -0,0 +1,621 @@ +using System.Buffers; +using System.Text; +using HighPerformanceLogIngestor.Models; +using HighPerformanceLogIngestor.Parsing; +using HighPerformanceLogIngestor.Pipelines; +using HighPerformanceLogIngestor.Tests.TestSupport; + +namespace HighPerformanceLogIngestor.Tests; + +public sealed class HighPerformanceLogTests +{ + [Fact] + public void + Parser_reads_exact_utf8_fields() + { + var parser = + new Utf8LogLineParser(); + + bool success = + parser.TryParse( + "1785924000|2|1001|Cache warmed"u8, + out LogEntry? entry); + + Assert.True( + success); + + Assert.NotNull( + entry); + + Assert.Equal( + DateTimeOffset + .FromUnixTimeSeconds( + 1_785_924_000), + entry.Timestamp); + + Assert.Equal( + (byte)2, + entry.Level); + + Assert.Equal( + 1001, + entry.EventId); + + Assert.Equal( + "Cache warmed", + entry.Message); + } + + [Fact] + public void + Parser_rejects_malformed_or_partial_fields() + { + var parser = + new Utf8LogLineParser(); + + byte[][] invalidLines = + [ + "bad|2|1001|Message"u8 + .ToArray(), + + "1785924000|9|1001|Message"u8 + .ToArray(), + + "1785924000|2x|1001|Message"u8 + .ToArray(), + + "1785924000|2|1001x|Message"u8 + .ToArray(), + + "1785924000|2|0|Message"u8 + .ToArray(), + + "1785924000|2|1001|"u8 + .ToArray(), + + [ + .. "1785924000|2|1001|"u8, + 0xC3, + 0x28 + ] + ]; + + foreach (byte[] line in + invalidLines) + { + Assert.False( + parser.TryParse( + line, + out _)); + } + + // ChunkedReadStream constructor validation. + Assert.Throws( + () => + new ChunkedReadStream( + null!, + 1)); + + Assert.Throws( + () => + new ChunkedReadStream( + [], + 0)); + + Assert.Throws( + () => + new ChunkedReadStream( + [], + -1)); + } + + [Fact] + public void + Decoder_uses_pooled_copy_for_multisegment_line() + { + var decoder = + CreateDecoder(); + + // Valid multi-segment line. + ReadOnlySequence line = + SegmentedSequence.Create( + "1785924"u8 + .ToArray(), + + "000|2|10"u8 + .ToArray(), + + "01|Cache warmed"u8 + .ToArray()); + + bool success = + decoder.TryDecode( + line, + out LogEntry? entry, + out bool usedPooledCopy); + + Assert.True( + success); + + Assert.True( + usedPooledCopy); + + Assert.NotNull( + entry); + + Assert.Equal( + 1001, + entry.EventId); + + // Invalid multi-segment line: takes pooled-copy path but + // fails parsing. + ReadOnlySequence invalidLine = + SegmentedSequence.Create( + "invalid"u8 + .ToArray(), + + "|2|"u8 + .ToArray(), + + "1001|msg"u8 + .ToArray()); + + success = + decoder.TryDecode( + invalidLine, + out entry, + out usedPooledCopy); + + Assert.False( + success); + + Assert.True( + usedPooledCopy); + + Assert.Null( + entry); + } + + [Fact] + public void + Decoder_trims_cr_and_rejects_oversized_lines() + { + var decoder = + CreateDecoder(); + + ReadOnlySequence crlfLine = + new( + "1785924000|2|1001|Cache warmed\r"u8 + .ToArray()); + + Assert.True( + decoder.TryDecode( + crlfLine, + out LogEntry? entry, + out bool usedPooledCopy)); + + Assert.NotNull( + entry); + + Assert.False( + usedPooledCopy); + + byte[] oversized = + Enumerable.Repeat( + (byte)'a', + LogLineDecoder + .MaximumLineLength + + 1) + .ToArray(); + + Assert.False( + decoder.TryDecode( + new ReadOnlySequence( + oversized), + out _, + out bool oversizedPooledCopy)); + + Assert.False( + oversizedPooledCopy); + } + + [Fact] + public async Task + Ingestor_handles_chunked_crlf_and_final_line() + { + const string input = + "1785924000|2|1001|First\r\n" + + "1785924001|4|1002|Second\n" + + "1785924002|1|1003|Final"; + + byte[] data = + Encoding.UTF8.GetBytes( + input); + + using var stream = + new ChunkedReadStream( + data, + maximumChunkSize: + 5); + + var accepted = + new List(); + + LogIngestionResult result = + await CreateIngestor() + .IngestAsync( + stream, + accepted.Add, + TestContext.Current + .CancellationToken); + + Assert.Equal( + 3, + result.TotalLines); + + Assert.Equal( + 3, + result.ValidLines); + + Assert.Equal( + 0, + result.InvalidLines); + + Assert.Collection( + accepted, + first => + Assert.Equal( + "First", + first.Message), + second => + Assert.Equal( + "Second", + second.Message), + final => + Assert.Equal( + "Final", + final.Message)); + } + + [Fact] + public async Task + Ingestor_handles_oversized_and_invalid_and_empty_lines() + { + // --- oversized record: 7000 'A's + LF + valid follower --- + // 7000 bytes ensures the retained buffer exceeds + // MaxRetainedBeforeNewline (4097) while no LF is present, + // exercising the ingestor-level discard path. + byte[] oversizedRecord = + Enumerable.Repeat( + (byte)'A', + 7_000) + .ToArray(); + + byte[] oversizedValid = + "1|1|1|validA\n"u8 + .ToArray(); + + byte[] oversizedData = + new byte[ + oversizedRecord.Length + + 1 + + oversizedValid.Length]; + + oversizedRecord.CopyTo( + oversizedData, + 0); + + oversizedData[ + oversizedRecord.Length] = + (byte)'\n'; + + oversizedValid.CopyTo( + oversizedData, + oversizedRecord.Length + + 1); + + using var oversizedStream = + new ChunkedReadStream( + oversizedData, + maximumChunkSize: + 2048); + + var oversizedAccepted = + new List(); + + LogIngestionResult oversizedResult = + await CreateIngestor() + .IngestAsync( + oversizedStream, + oversizedAccepted.Add, + TestContext.Current + .CancellationToken); + + // --- exact 4096-byte LF-terminated record + valid follower --- + // "1785924000|2|1001|" = 22 bytes; pad message to 4096 total. + byte[] prefix = + "1785924000|2|1001|"u8 + .ToArray(); + + const int boundaryLength = + LogLineDecoder.MaximumLineLength; + + int messagePad = + boundaryLength + - prefix.Length; + + byte[] boundaryLF = + new byte[ + prefix.Length + + messagePad + + 1 + + oversizedValid.Length]; + + prefix.CopyTo( + boundaryLF, + 0); + + Array.Fill( + boundaryLF, + (byte)'x', + prefix.Length, + messagePad); + + boundaryLF[ + boundaryLength] = + (byte)'\n'; + + oversizedValid.CopyTo( + boundaryLF, + boundaryLength + + 1); + + using var boundaryLFStream = + new ChunkedReadStream( + boundaryLF, + maximumChunkSize: + 2_048); + + var boundaryLFAccepted = + new List(); + + LogIngestionResult boundaryLFResult = + await CreateIngestor() + .IngestAsync( + boundaryLFStream, + boundaryLFAccepted.Add, + TestContext.Current + .CancellationToken); + + // --- exact 4096-byte CRLF-terminated record + valid follower --- + byte[] boundaryCRLF = + new byte[ + prefix.Length + + messagePad + + 2 + + oversizedValid.Length]; + + prefix.CopyTo( + boundaryCRLF, + 0); + + Array.Fill( + boundaryCRLF, + (byte)'x', + prefix.Length, + messagePad); + + boundaryCRLF[ + boundaryLength] = + (byte)'\r'; + + boundaryCRLF[ + boundaryLength + + 1] = + (byte)'\n'; + + oversizedValid.CopyTo( + boundaryCRLF, + boundaryLength + + 2); + + using var boundaryCRLFStream = + new ChunkedReadStream( + boundaryCRLF, + maximumChunkSize: + 2_048); + + var boundaryCRLFAccepted = + new List(); + + LogIngestionResult boundaryCRLFResult = + await CreateIngestor() + .IngestAsync( + boundaryCRLFStream, + boundaryCRLFAccepted.Add, + TestContext.Current + .CancellationToken); + + // --- invalid-and-empty-lines data --- + const string invalidInput = + "1785924000|2|1001|Valid\n" + + "\n" + + "invalid\n" + + "1785924001|1|1002|Also valid\n"; + + using var invalidStream = + new MemoryStream( + Encoding.UTF8 + .GetBytes( + invalidInput), + writable: + false); + + var invalidAccepted = + new List(); + + LogIngestionResult invalidResult = + await CreateIngestor() + .IngestAsync( + invalidStream, + invalidAccepted.Add, + TestContext.Current + .CancellationToken); + + // --- Oversized-record assertions --- + // total=2 (1 discarded + 1 valid), valid=1, invalid=1 + Assert.Equal( + 2, + oversizedResult.TotalLines); + + Assert.Equal( + 1, + oversizedResult.ValidLines); + + Assert.Equal( + 1, + oversizedResult.InvalidLines); + + Assert.Single( + oversizedAccepted); + + Assert.Equal( + "validA", + oversizedAccepted[0].Message); + + // --- 4096-byte LF boundary assertions --- + Assert.Equal( + 2, + boundaryLFResult.TotalLines); + + Assert.Equal( + 2, + boundaryLFResult.ValidLines); + + Assert.Equal( + 0, + boundaryLFResult.InvalidLines); + + Assert.Equal( + 2, + boundaryLFAccepted.Count); + + Assert.Equal( + "validA", + boundaryLFAccepted[1].Message); + + // --- 4096-byte CRLF boundary assertions --- + Assert.Equal( + 2, + boundaryCRLFResult.TotalLines); + + Assert.Equal( + 2, + boundaryCRLFResult.ValidLines); + + Assert.Equal( + 0, + boundaryCRLFResult.InvalidLines); + + Assert.Equal( + 2, + boundaryCRLFAccepted.Count); + + Assert.Equal( + "validA", + boundaryCRLFAccepted[1].Message); + + // --- Invalid-and-empty-lines assertions --- + Assert.Equal( + 4, + invalidResult.TotalLines); + + Assert.Equal( + 2, + invalidResult.ValidLines); + + Assert.Equal( + 2, + invalidResult.InvalidLines); + + Assert.Equal( + 2, + invalidAccepted.Count); + + Assert.Equal( + "Valid", + invalidAccepted[0].Message); + + Assert.Equal( + "Also valid", + invalidAccepted[1].Message); + } + + [Fact] + public async Task + Ingestor_leaves_caller_stream_open() + { + byte[] data = + "1785924000|2|1001|Valid\n"u8 + .ToArray(); + + using var stream = + new MemoryStream( + data, + writable: + false); + + await CreateIngestor() + .IngestAsync( + stream, + _ => + { + }, + TestContext.Current + .CancellationToken); + + Assert.True( + stream.CanRead); + + stream.Position = + 0; + + Assert.Equal( + (int)'1', + stream.ReadByte()); + } + + [Fact] + public async Task + Ingestor_propagates_caller_cancellation() + { + using var stream = + new BlockingReadStream(); + + using var cancellation = + new CancellationTokenSource(); + + cancellation.Cancel(); + + await Assert.ThrowsAnyAsync< + OperationCanceledException>( + () => + CreateIngestor() + .IngestAsync( + stream, + _ => + { + }, + cancellation.Token)); + } + + private static LogLineDecoder + CreateDecoder() => + new( + new Utf8LogLineParser()); + + private static LogIngestor + CreateIngestor() => + new( + CreateDecoder()); +} \ No newline at end of file diff --git a/csharp-language/high-performance-memory-management/tests/HighPerformanceLogIngestor.Tests/TestSupport/ChunkedReadStream.cs b/csharp-language/high-performance-memory-management/tests/HighPerformanceLogIngestor.Tests/TestSupport/ChunkedReadStream.cs new file mode 100644 index 0000000..626797d --- /dev/null +++ b/csharp-language/high-performance-memory-management/tests/HighPerformanceLogIngestor.Tests/TestSupport/ChunkedReadStream.cs @@ -0,0 +1,197 @@ +namespace HighPerformanceLogIngestor.Tests.TestSupport; + +internal sealed class ChunkedReadStream : Stream +{ + private readonly byte[] _data; + private readonly int _maximumChunkSize; + private int _position; + + internal ChunkedReadStream( + byte[] data, + int maximumChunkSize) + { + ArgumentNullException + .ThrowIfNull( + data); + + if (maximumChunkSize <= 0) + { + throw new ArgumentOutOfRangeException( + nameof(maximumChunkSize), + maximumChunkSize, + "The maximum chunk size must be greater than zero."); + } + + _data = + data; + + _maximumChunkSize = + maximumChunkSize; + } + + public override bool CanRead => + true; + + public override bool CanSeek => + false; + + public override bool CanWrite => + false; + + public override long Length => + _data.Length; + + public override long Position + { + get => + _position; + + set => + throw new NotSupportedException(); + } + + public override int Read( + byte[] buffer, + int offset, + int count) + { + ArgumentNullException + .ThrowIfNull( + buffer); + + return ReadCore( + buffer.AsSpan( + offset, + count)); + } + + public override int Read( + Span buffer) => + ReadCore( + buffer); + + public override ValueTask + ReadAsync( + Memory buffer, + CancellationToken cancellationToken = + default) + { + cancellationToken + .ThrowIfCancellationRequested(); + + return ValueTask.FromResult( + ReadCore( + buffer.Span)); + } + + public override void Flush() + { + } + + public override long Seek( + long offset, + SeekOrigin origin) => + throw new NotSupportedException(); + + public override void SetLength( + long value) => + throw new NotSupportedException(); + + public override void Write( + byte[] buffer, + int offset, + int count) => + throw new NotSupportedException(); + + private int ReadCore( + Span destination) + { + if (_position + >= _data.Length) + { + return 0; + } + + int count = + Math.Min( + _maximumChunkSize, + Math.Min( + destination.Length, + _data.Length + - _position)); + + _data.AsSpan( + _position, + count) + .CopyTo( + destination); + + _position += + count; + + return count; + } +} + +internal sealed class BlockingReadStream : + Stream +{ + public override bool CanRead => + true; + + public override bool CanSeek => + false; + + public override bool CanWrite => + false; + + public override long Length => + throw new NotSupportedException(); + + public override long Position + { + get => + throw new NotSupportedException(); + + set => + throw new NotSupportedException(); + } + + public override int Read( + byte[] buffer, + int offset, + int count) => + throw new NotSupportedException(); + + public override async ValueTask + ReadAsync( + Memory buffer, + CancellationToken cancellationToken = + default) + { + await Task.Delay( + Timeout.InfiniteTimeSpan, + cancellationToken); + + return 0; + } + + public override void Flush() + { + } + + public override long Seek( + long offset, + SeekOrigin origin) => + throw new NotSupportedException(); + + public override void SetLength( + long value) => + throw new NotSupportedException(); + + public override void Write( + byte[] buffer, + int offset, + int count) => + throw new NotSupportedException(); +} \ No newline at end of file diff --git a/csharp-language/high-performance-memory-management/tests/HighPerformanceLogIngestor.Tests/TestSupport/SegmentedSequence.cs b/csharp-language/high-performance-memory-management/tests/HighPerformanceLogIngestor.Tests/TestSupport/SegmentedSequence.cs new file mode 100644 index 0000000..33daaec --- /dev/null +++ b/csharp-language/high-performance-memory-management/tests/HighPerformanceLogIngestor.Tests/TestSupport/SegmentedSequence.cs @@ -0,0 +1,64 @@ +using System.Buffers; + +namespace HighPerformanceLogIngestor.Tests.TestSupport; + +internal static class SegmentedSequence +{ + public static ReadOnlySequence + Create( + params byte[][] segments) + { + ArgumentNullException + .ThrowIfNull( + segments); + + if (segments.Length + == 0) + { + return ReadOnlySequence + .Empty; + } + + var first = + new Segment( + segments[0]); + + Segment last = + first; + + for (int index = 1; + index < segments.Length; + index++) + { + last = + last.Append( + segments[index]); + } + + return new ReadOnlySequence( + first, + 0, + last, + last.Memory.Length); + } + + private sealed class Segment : ReadOnlySequenceSegment + { + internal Segment(ReadOnlyMemory memory) + { + Memory = memory; + } + + public Segment Append(ReadOnlyMemory memory) + { + var next = new Segment(memory) + { + RunningIndex = RunningIndex + Memory.Length + }; + + Next = next; + + return next; + } + } +} \ No newline at end of file