diff --git a/.github/workflows/benchmark-report.yml b/.github/workflows/benchmark-report.yml new file mode 100644 index 00000000..51e5fca4 --- /dev/null +++ b/.github/workflows/benchmark-report.yml @@ -0,0 +1,21 @@ +name: .NET SDK benchmark report + +on: + workflow_run: + workflows: ['.NET SDK benchmarks'] + types: [completed] + +permissions: {} + +jobs: + report: + permissions: + contents: read + actions: read + pull-requests: write + uses: PostHog/.github/.github/workflows/sdk-benchmark-report.yml@2a6111d08d784c0ca84f7f97f32a13fdd6d31449 # SDK benchmark reporting + with: + workflow-path: .github/workflows/benchmarks.yml + artifact-name: dotnet-benchmark-results + report-key: posthog-dotnet + sdk: posthog-dotnet diff --git a/.github/workflows/benchmarks.yml b/.github/workflows/benchmarks.yml new file mode 100644 index 00000000..2055a22e --- /dev/null +++ b/.github/workflows/benchmarks.yml @@ -0,0 +1,89 @@ +name: .NET SDK benchmarks + +on: [pull_request] + +permissions: {} + +concurrency: + group: ${{ github.workflow }}-${{ github.event.pull_request.number }} + cancel-in-progress: true + +jobs: + benchmark: + runs-on: ubuntu-24.04 + timeout-minutes: 12 + permissions: + contents: read + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.event.pull_request.head.sha }} + fetch-depth: 0 + persist-credentials: false + + - uses: actions/setup-dotnet@a98b56852c35b8e3190ac28c8c2271da59106c68 # v6.0.0 + with: + dotnet-version: '9.0.109' + + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: '3.12.12' + + - name: Prepare pinned baseline + env: + TARGET_SHA: ${{ github.event.pull_request.base.sha }} + HEAD_SHA: ${{ github.event.pull_request.head.sha }} + run: | + set -euo pipefail + export BASE_WORKTREE="$RUNNER_TEMP/dotnet-benchmark-base" + export RESULTS="$RUNNER_TEMP/dotnet-benchmark-results" + printf 'BASE_WORKTREE=%s\nRESULTS=%s\n' "$BASE_WORKTREE" "$RESULTS" >> "$GITHUB_ENV" + BASE_SHA=$(git merge-base "$TARGET_SHA" "$HEAD_SHA") + git worktree add --detach "$BASE_WORKTREE" "$BASE_SHA" + mkdir -p "$RESULTS" + python -c 'import json, os, pathlib, sys; pathlib.Path(os.environ["RESULTS"], "revisions.json").write_text(json.dumps({"target": os.environ["TARGET_SHA"], "head": os.environ["HEAD_SHA"], "base": sys.argv[1]}, indent=2))' "$BASE_SHA" + + - name: Test coordinator controls + run: python -B -m unittest discover -s benchmarks -p 'test_*.py' + + - name: Build, test native controls and measure paired base and head + id: measure + timeout-minutes: 10 + run: | + python -B benchmarks/run.py \ + --base "$BASE_WORKTREE" \ + --head "$GITHUB_WORKSPACE" \ + --build-directory "$RUNNER_TEMP/dotnet-benchmark-build" \ + --output "$RESULTS" \ + --repetitions 6 + + - name: Render benchmark comparison + if: ${{ !cancelled() && (steps.measure.outcome == 'success' || steps.measure.outcome == 'failure') }} + uses: PostHog/.github/.github/actions/benchmark-report@04f8421ec4daa7f4066eaeb4c53621ba79ddc3ca # SDK benchmark reporting + with: + mode: render + source-status: ${{ steps.measure.outcome }} + base-results: ${{ runner.temp }}/dotnet-benchmark-results/base.json + head-results: ${{ runner.temp }}/dotnet-benchmark-results/head.json + output-directory: ${{ runner.temp }}/dotnet-benchmark-report + + - name: Upload rendered comparison + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: dotnet-benchmark-report + path: ${{ runner.temp }}/dotnet-benchmark-report + if-no-files-found: ignore + retention-days: 14 + + - name: Upload raw benchmark results + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: dotnet-benchmark-results + path: | + ${{ runner.temp }}/dotnet-benchmark-results/base.json + ${{ runner.temp }}/dotnet-benchmark-results/head.json + ${{ runner.temp }}/dotnet-benchmark-results/revisions.json + if-no-files-found: warn + retention-days: 14 diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 58a8f434..9ddf9921 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -45,6 +45,113 @@ The test projects target both `net8.0` and `netcoreapp3.1`. While .NET Core 3.1 This testing approach ensures broad compatibility without requiring users to install legacy runtimes in production. +## Performance benchmarks + +Every opened, reopened, or updated pull request runs `.github/workflows/benchmarks.yml`. +It compares the merge base with the exact PR head, using the head's benchmark code for +both revisions. Results are informational: slow measurements do not fail the job, but +build, execution, fixture correctness, or delivery failures do. The separate trusted +`workflow_run` publisher posts a sticky PR comment once its workflow is on the default +branch. Reporting is provided by [PostHog/.github](https://github.com/PostHog/.github/blob/main/.github/workflows/sdk-benchmark-report.md). + +The 20 workloads follow the shared [SDK benchmark specification](https://github.com/PostHog/.github/blob/main/.github/workflows/sdk-benchmark-specification.md) +(`suite.version=2`, `suite.specification=sdk-benchmarks/v1`). They cover nine single-flag +cases (16,000 evaluations each), all-flags at 10/100/1,000 definitions (1,600/160/160 +evaluations), 1,600 load evaluations using at most eight ThreadPool tasks, 40 definition +reloads, and six capture workloads with 1,000 events each. Capture uses 32-byte and +4-KiB payloads for enqueue, steady delivery (explicit flush every 100), and burst +delivery (one final explicit flush). Enqueue has timing/allocation passes; both delivery +patterns have timing/allocation/memory passes. Single/all-flags have timing/allocation +passes; load and refresh have memory passes. Flag cases include percentage rollouts, +variants, properties, regexes, cohorts, early/late/no-match groups and inactive flags. +The suite exercises supported legacy getters; it does not measure the newer +`EvaluateFlagsAsync` snapshot API. + +### Run locally + +Install .NET SDK 9.0.109 (including runtime 9.0.8) and Python 3.12. The coordinator uses +only Python's standard library; all instrumentation and SDK workloads execute in C#. +Use a fresh build directory for each invocation: + +```bash +python3 -B -m unittest discover -s benchmarks -p 'test_*.py' +RUN=$(mktemp -d) +python3 -B benchmarks/run.py \ + --base /path/to/baseline-checkout \ + --head "$PWD" \ + --build-directory "$RUN/build" \ + --output "$RUN/results" \ + --repetitions 6 +``` + +Use `--smoke --repetitions 1` for correctness checks across all 20 workloads: 160 +single-flag evaluations, 16 all-flags evaluations, 32 load evaluations, four refreshes, +and 200 events per capture workload. Smoke is not repeatability evidence. Native +controls run automatically before pairs and check the inventory, fixtures, phase reset, +flush schedules, reload generations, deterministic checkpoints and failure detection. +For a same-revision noise control, point both `--base` and `--head` at the same checkout. +Run repeatability checks on an otherwise idle machine; concurrent validation is only +correctness evidence. Do not compare results from different hosts or languages. + +The coordinator copies the same standalone benchmark project outside the checkout, +builds it separately against each revision's public SDK project with locked NuGet +restore, and fingerprints the loaded SDK assembly and dependency graph. It uses the +SDK's `net8.0` target on pinned .NET 9.0.8. Benchmark workers use workstation GC and +optimized, non-tiered JIT compilation to avoid tier transitions during measurements. +This is one explicit runtime configuration, not a claim about all .NET deployments. +Each workload/repetition/pass starts a fresh process; base/head order alternates. +Warmup is 1,024 single evaluations, 16 all/load evaluations, two refreshes, or 100 +captures, followed by a 25-ms native settling delay and full GC. Warmup captures are +flushed and verified before counters and bounded sequence state reset. Every measured +flag workload starts with generation 0 reloaded and verified. Full flag correctness +sweeps occur outside measurement, including after the memory post-settling snapshot; +measured refreshes only reload definitions and check that each made a request. + +### Measurement boundaries + +- Elapsed time uses `Stopwatch`; CPU uses whole-process `TotalProcessorTime`. Values + are expressed in nanoseconds, but CPU-counter resolution is OS-dependent. +- Allocation volume uses `GC.GetTotalAllocatedBytes(precise: true)`, including managed + objects collected during the operation. It excludes native allocations and does not + provide exact object counts; counts are reported as unsupported, never zero. +- Idle and post-settling managed heap use `GC.GetTotalMemory(false)` after explicit + full collections. These are retained-heap estimates, not allocation volume. +- RSS checkpoints follow every 32 completed load evaluations, each refresh, or every + 100 captures, including the final operation if needed. Steady checkpoints follow the + scheduled flush; burst checkpoints precede the final flush. `observed_workload_peak_rss` + is the maximum of before-work, checkpoints and after-work readings, not a continuous + maximum. `process_lifetime_peak_rss` (`PeakWorkingSet64`) includes startup and warmup. + `idle_process_rss`, `post_settling_process_rss` and `post_settling_heap_used` retain their + distinct scopes. No forced GC occurs within measured work or checkpoint callbacks; + full collections bracket memory measurement. GC heap bytes and RSS are not + interchangeable; releasing objects need not reduce RSS. +- Public `IHttpClientFactory` injection supplies offline definitions and acknowledges + batches. Measurements include in-process transport JSON parsing and validation, with + compression disabled through public options. This is not a network or remote + ingestion benchmark. No project credentials are required. +- Caller fixtures are constructed outside timing/allocation boundaries. Event `benchmark` + uses 16 rotating identities and separate per-event property dictionaries containing + ASCII `payload`, `plan=pro` and integer `sequence`, starting at zero for each phase. + The receiver checks exact payload, plan and sequence-to-identity mapping using bounded, + preallocated uniqueness state: balanced duplicate/loss fails; order is not required. +- Queue capacity (100,001), automatic flush count (100,000), and maximum count-based + batch size (100,000) exceed the workload; polling and flush intervals are one hour. + Native background consumers + and size-based transport behavior remain active. The patterns specify caller flushes, + not HTTP request counts. Enqueue measures synchronous `Capture` API return + (`completion_boundary=api_return`), excluding its later flush and verification; + process-wide CPU/allocation counters may include concurrent background work. Delivery + includes capture preparation, serialization and public `FlushAsync` completion + (`completion_boundary=completed_delivery`). Verification never adds a rescuing flush. + +CI keeps the suite on one runner, with a 12-minute job deadline and a 10-minute +build/measurement step deadline to leave time for reporting and artifact upload. +Each worker has a 120-second deadline and a 512-MiB **managed heap** limit, not an RSS +limit. After two worker timeouts the coordinator finishes the current pair, saves +partial diagnostics and fails the run. A killed process is not automatically classified +as OOM. Raw `base.json` and `head.json` retain statuses, metadata, repetitions and memory +checkpoints; the PR report suppresses numeric comparisons when the source run failed. + ## Public API changes Public API is hard to change once it ships, so agree on it before writing the implementation. Our [SDK guidelines](https://posthog.com/handbook/engineering/sdks/guidelines) explain how we design it. diff --git a/benchmarks/.gitignore b/benchmarks/.gitignore new file mode 100644 index 00000000..c18dd8d8 --- /dev/null +++ b/benchmarks/.gitignore @@ -0,0 +1 @@ +__pycache__/ diff --git a/benchmarks/Benchmark.csproj b/benchmarks/Benchmark.csproj new file mode 100644 index 00000000..5c46b6d7 --- /dev/null +++ b/benchmarks/Benchmark.csproj @@ -0,0 +1,18 @@ + + + + Exe + net9.0 + 9.0.8 + Disable + enable + enable + true + false + false + false + + + + + diff --git a/benchmarks/Controls.cs b/benchmarks/Controls.cs new file mode 100644 index 00000000..1f595288 --- /dev/null +++ b/benchmarks/Controls.cs @@ -0,0 +1,241 @@ +using System.Diagnostics; +using System.Runtime.CompilerServices; +using System.Text.Json; + +namespace PostHog.Benchmarks; + +internal static class Controls +{ + [MethodImpl(MethodImplOptions.NoInlining)] + static void AllocateChurn() + { + for (var i = 0; i < 32; i++) GC.KeepAlive(new byte[1024 * 1024]); + } + + sealed class ExtendedFlagOptions : FeatureFlagOptions + { + public bool UnrelatedDefault { get; init; } + } + + static void CheckFixtureIdentity() + { + var spec = Fixtures.Workloads(smoke: true)[0]; + var baseline = Fixtures.FixtureDigest(spec); + var extended = Fixtures.Options.Select(option => new ExtendedFlagOptions + { + PersonProperties = new(option.PersonProperties!), + OnlyEvaluateLocally = option.OnlyEvaluateLocally, + SendFeatureFlagEvents = option.SendFeatureFlagEvents, + }).ToArray(); + using var sdkJson = JsonDocument.Parse(JsonSerializer.Serialize(extended, Program.Json)); + Program.Require(sdkJson.RootElement[0].TryGetProperty("unrelated_default", out _), "Extension control must change the serialized SDK schema"); + Program.Require(Fixtures.FixtureDigest(spec, extended) == baseline, "SDK schema extension changed explicit fixture identity"); + + using var inputs = JsonDocument.Parse(Fixtures.FixtureInputs()); + string[] expectedKeys = ["distinct_id", "person_properties", "only_evaluate_locally", "send_feature_flag_events"]; + foreach (var input in inputs.RootElement.EnumerateArray()) + { + Program.Require(input.EnumerateObject().Select(p => p.Name).Order().SequenceEqual(expectedKeys.Order()), + "Fixture identity must contain only benchmark-owned explicit inputs"); + } + extended[0].PersonProperties!["plan"] = "changed-plan"; + Program.Require(Fixtures.FixtureDigest(spec, extended) != baseline, "Person-property change did not change fixture identity"); + Program.Require(Fixtures.FixtureDigest(spec with { FlagCount = spec.FlagCount + 1 }) != baseline, + "Definition change did not change fixture identity"); + } + + static void CheckInventory() + { + foreach (var smoke in new[] { false, true }) + { + var workloads = Fixtures.Workloads(smoke); + var expectedIds = new[] { "boolean", "rollout", "variant", "property", "regex", "cohort", "early", "late", "no-match" } + .Select(kind => $"flags.single.{kind}") + .Concat(new[] { "flags.all.10", "flags.all.100", "flags.all.1000", "flags.load.1000", "flags.refresh.1000", + "capture.enqueue.32", "capture.enqueue.4096", "capture.drain.steady.32", "capture.drain.steady.4096", + "capture.drain.burst.32", "capture.drain.burst.4096" }).ToArray(); + Program.Require(workloads.Select(w => w.Id).SequenceEqual(expectedIds), "Incorrect workload inventory"); + foreach (var spec in workloads) + { + var expectedOperations = spec.Mode switch + { + "single" => smoke ? 160 : 16000, + "refresh" => smoke ? 4 : 40, + "all" when spec.Concurrency == 8 => smoke ? 32 : 1600, + "all" => smoke ? 16 : spec.FlagCount == 10 ? 1600 : 160, + _ => smoke ? 200 : 1000, + }; + string[] expectedPasses = spec.Mode == "drain" ? ["timing", "allocation", "memory"] + : spec.Mode == "refresh" || spec.Concurrency == 8 ? ["memory"] : ["timing", "allocation"]; + Program.Require(spec.Operations == expectedOperations && spec.Passes.SequenceEqual(expectedPasses), "Incorrect operations/passes"); + Program.Require(spec.Concurrency == (spec.Id == "flags.load.1000" ? 8 : 1), "Incorrect concurrency"); + if (spec.Mode == "single") + Program.Require(spec.FlagCount == 10 && Fixtures.Cases[spec.FlagIndex] == spec.Id.Split('.')[2], "Incorrect single fixture"); + else if (spec.FlagCount > 0) + Program.Require(spec.FlagCount == int.Parse(spec.Id.Split('.')[2]), "Incorrect flag count"); + if (spec.PayloadBytes > 0) + { + Program.Require(spec.PayloadBytes == int.Parse(spec.Id.Split('.')[^1]), "Incorrect payload size"); + Program.Require(spec.CompletionBoundary == (spec.Mode == "capture" ? "api_return" : "completed_delivery"), "Incorrect completion boundary"); + Program.Require(spec.FlushEvery == (spec.DrainStrategy == "steady" ? 100 : spec.Mode == "drain" ? spec.Operations : null), "Incorrect flush interval"); + Program.Require(spec.DrainStrategy == (spec.Mode == "drain" ? spec.Id.Split('.')[2] : null), "Incorrect drain strategy"); + } + } + } + } + + static JsonElement Event(int sequence, int payloadBytes = 32, string plan = "pro", string? user = null) => + JsonSerializer.SerializeToElement(new + { + @event = Fixtures.EventName, + properties = new Dictionary + { + ["distinct_id"] = user ?? Fixtures.Users[(sequence % 16 + 16) % 16], + ["payload"] = new string('x', payloadBytes), + ["plan"] = plan, + ["sequence"] = sequence, + }, + }); + + static void Reject(Action action, string message) + { + var rejected = false; + try { action(); } + catch (InvalidOperationException) { rejected = true; } + Program.Require(rejected, message); + } + + static void CheckReceiver() + { + using var transport = new Transport(Fixtures.Workloads(true).Single(s => s.Id == "capture.enqueue.32")); + transport.ResetDelivery(32); + for (var i = 31; i >= 0; i--) transport.ValidateEvent(Event(i)); + Program.Require(transport.Delivered == 32, "Receiver must accept reordered delivery"); + transport.ResetDelivery(32); + // Replacing sequence 16 with 0 preserves both total and per-user counts. + for (var i = 0; i < 16; i++) transport.ValidateEvent(Event(i)); + Reject(() => transport.ValidateEvent(Event(0)), "Balanced duplicate/loss accepted"); + foreach (var invalid in new[] { Event(-1), Event(32), Event(16, plan: "free"), Event(16, payloadBytes: 31), Event(16, user: Fixtures.Users[1]) }) + Reject(() => transport.ValidateEvent(invalid), "Invalid capture fixture accepted"); + var properties = Fixtures.CaptureProperties(32, 32); + properties[1]["sequence"] = 999; + Program.Require((int)properties[0]["sequence"] == 0 && (int)properties[2]["sequence"] == 2, "Caller property dictionaries are shared"); + } + + static async Task CheckSchedulesAndRefresh() + { + foreach (var smoke in new[] { false, true }) + { + foreach (var spec in Fixtures.Workloads(smoke).Where(s => s.Mode == "drain")) + { + await using var scenario = new Scenario(spec); + await scenario.Prepare(); + Program.Require(scenario.Transport.Delivered == 0 && scenario.Transport.BatchRequests == 0 && scenario.ExplicitFlushes == 0, + "Warmup delivery phase did not reset"); + var flushes = new List(); + var checkpoints = new List(); + await scenario.Execute(spec.Operations, completed => + { + checkpoints.Add(completed); + Program.Require(scenario.ExplicitFlushes == (spec.DrainStrategy == "steady" ? completed / 100 : 0), + "Checkpoint on wrong side of explicit flush"); + }, flushes.Add); + var expectedCheckpoints = Enumerable.Range(1, spec.Operations / 100).Select(i => i * 100).ToArray(); + Program.Require(checkpoints.SequenceEqual(expectedCheckpoints), "Incorrect capture checkpoints"); + Program.Require(flushes.SequenceEqual(spec.DrainStrategy == "steady" ? expectedCheckpoints : [spec.Operations]), "Incorrect explicit flush schedule"); + var count = scenario.ExplicitFlushes; + await scenario.VerifyCompletion(spec.Operations); + Program.Require(scenario.ExplicitFlushes == count, "Delivery verification performed a rescuing flush"); + } + } + await using var refresh = new Scenario(Fixtures.Workloads(true).Single(s => s.Mode == "refresh")); + await refresh.Prepare(); + Program.Require(refresh.Transport.Generation == 0, "Measured flags did not start at generation zero"); + for (var generation = 1; generation <= 4; generation++) + { + var requests = refresh.Transport.DefinitionRequests; + await refresh.Execute(1); + Program.Require(refresh.Transport.Generation == generation && refresh.Transport.DefinitionRequests == requests + 1, "Incorrect reload generation/request count"); + await refresh.VerifyFlags(); + } + } + + static async Task CheckFailedDelivery() + { + foreach (var strategy in new[] { "steady", "burst" }) + { + await using var dropped = new Scenario(Fixtures.Workloads(true).Single(s => s.Id == $"capture.drain.{strategy}.32")); + await dropped.Prepare(); + dropped.Transport.DropEvents = true; + await dropped.Execute(dropped.Spec.Operations); + dropped.Transport.DropEvents = false; + var flushes = dropped.ExplicitFlushes; + var rejected = false; + try { await dropped.VerifyCompletion(dropped.Spec.Operations); } + catch (InvalidOperationException) { rejected = true; } + Program.Require(rejected && dropped.ExplicitFlushes == flushes, "Incomplete measured delivery rescued/accepted by verification"); + } + } + + internal static async Task Run() + { + CheckFixtureIdentity(); + CheckInventory(); + CheckReceiver(); + await CheckSchedulesAndRefresh(); + await CheckFailedDelivery(); + foreach (var spec in Fixtures.Workloads(smoke: true)) + { + foreach (var pass in spec.Passes) + { + // Each pass normally gets a fresh worker; use a fresh scenario here too. + await using var measured = new Scenario(spec); + Program.Require(measured.Options.FlushAt == 100000 && measured.Options.MaxQueueSize == 100001 && + measured.Options.MaxBatchSize == 100000, "Incorrect controlled count limits"); + Program.Require(!measured.Options.EnableCompression && measured.Options.FlushInterval > TimeSpan.FromSeconds(120) && + measured.Options.FeatureFlagPollInterval > TimeSpan.FromSeconds(120), "Incorrect compression or automatic polling/flush configuration"); + var result = await Measurement.Run(measured, pass); + Program.Require((string)result["status"] == "ok", "Control workload failed"); + if (pass == "allocation") Program.Require((double)result["allocated_bytes_per_op"] > 0, "Missing allocation signal"); + if (pass == "timing") + { + Program.Require((double)result["elapsed_ns"] > 0, "Missing elapsed-time signal"); + Program.Require((double)result["process_cpu_time_per_op"] >= 0, "Invalid CPU signal"); + } + if (pass == "memory") + { + var points = JsonSerializer.SerializeToElement(result["checkpoints"]); + var positions = points.EnumerateArray().Select(p => p.GetProperty("operations").GetInt32()).ToArray(); + Program.Require(positions.SequenceEqual(positions.Distinct().Order()), "Checkpoint positions must increase strictly"); + var interval = spec.Mode == "refresh" ? 1 : spec.Mode == "drain" ? 100 : 32; + var expected = Enumerable.Range(1, spec.Operations).Where(i => i % interval == 0 || i == spec.Operations); + Program.Require(positions.SequenceEqual(expected), "Incorrect deterministic checkpoints"); + Program.Require((long)result["post_settling_heap_used"] > 0, "Missing retained-heap signal"); + } + } + } + Measurement.Collect(); + var before = GC.GetTotalAllocatedBytes(true); + AllocateChurn(); + Measurement.Collect(); + Program.Require(GC.GetTotalAllocatedBytes(true) - before >= 32 * 1024 * 1024, "Allocation counter lost collected objects"); + var heap = GC.GetTotalMemory(false); + var retained = new byte[32 * 1024 * 1024]; + Measurement.Collect(); + Program.Require(GC.GetTotalMemory(false) - heap >= retained.Length, "Retained-heap counter missed live data"); + GC.KeepAlive(retained); + using var process = Process.GetCurrentProcess(); + Program.Require(Measurement.Rss(process) > 0, "RSS unavailable"); + + await using var dropped = new Scenario(Fixtures.Workloads(true).Single(s => s.Id == "capture.enqueue.32")); + await dropped.Prepare(); + dropped.Transport.DropEvents = true; + await dropped.Execute(1); + await dropped.Client.FlushAsync(); + var rejected = false; + try { dropped.VerifyDelivery(1); } + catch (InvalidOperationException) { rejected = true; } + Program.Require(rejected, "Delivery control accepted missing events"); + Console.WriteLine("20 workloads: inventory, fixture identity, receiver uniqueness, flush schedules, phase reset, refresh generations, checkpoints, failed delivery, allocation churn, retained heap and RSS controls passed."); + } +} diff --git a/benchmarks/Fixtures.cs b/benchmarks/Fixtures.cs new file mode 100644 index 00000000..c2e45155 --- /dev/null +++ b/benchmarks/Fixtures.cs @@ -0,0 +1,120 @@ +using System.Globalization; +using System.Security.Cryptography; +using System.Text; +using System.Text.Json; + +namespace PostHog.Benchmarks; + +internal record Workload(string Id, string Mode, int Operations, string[] Passes, + int FlagCount = 0, int FlagIndex = 0, int Concurrency = 1, int PayloadBytes = 0, + string? DrainStrategy = null, int? FlushEvery = null, string? CompletionBoundary = null); + +internal static class Fixtures +{ + internal const string EventName = "benchmark"; + internal static readonly string[] Cases = + ["boolean", "rollout", "variant", "property", "regex", "cohort", "early", "late", "no-match", "inactive"]; + internal static readonly string[] Users = Enumerable.Range(0, 16).Select(i => $"benchmark-user-{i}").ToArray(); + internal static readonly FeatureFlagOptions[] Options = Enumerable.Range(0, 16).Select(i => new FeatureFlagOptions + { + OnlyEvaluateLocally = true, + SendFeatureFlagEvents = false, + PersonProperties = new() { ["plan"] = i % 2 == 0 ? "pro" : "free", ["email"] = $"user-{i}@example.test", ["score"] = i }, + }).ToArray(); + + internal static string Key(int i) => $"{Cases[i % Cases.Length]}-{i}"; + internal static string Digest(byte[] bytes) => Convert.ToHexStringLower(SHA256.HashData(bytes)); + internal static string Digest(string value) => Digest(Encoding.UTF8.GetBytes(value)); + internal static string FixtureInputs(IEnumerable? options = null) => JsonSerializer.Serialize( + (options ?? Options).Select((option, i) => new + { + DistinctId = Users[i], + option.PersonProperties, + option.OnlyEvaluateLocally, + option.SendFeatureFlagEvents, + }), Program.Json); + + internal static string FixtureDigest(Workload spec, IEnumerable? options = null) => + Digest(Definitions(spec.FlagCount) + FixtureInputs(options) + JsonSerializer.Serialize(spec, Program.Json) + + (spec.PayloadBytes > 0 ? EventName + JsonSerializer.Serialize(CaptureProperties(spec.PayloadBytes, spec.Operations), Program.Json) : "")); + + internal static Dictionary[] CaptureProperties(int payloadBytes, int operations) + { + var payload = new string('x', payloadBytes); + return Enumerable.Range(0, operations).Select(i => new Dictionary + { + ["payload"] = payload, + ["plan"] = "pro", + ["sequence"] = i, + }).ToArray(); + } + + static object Property(string key, object value, string op = "exact") => new { key, value, @operator = op, type = "person" }; + static object Group(object[] properties, int rollout = 100) => new { properties, rollout_percentage = rollout }; + + internal static string Definitions(int count, int generation = 0) + { + var flags = Enumerable.Range(0, count).Select(i => + { + var kind = Cases[i % Cases.Length]; + var match = Group([Property("plan", new[] { "pro" })]); + var miss = Group([Property("plan", new[] { "never" })]); + object[] groups = kind switch + { + "rollout" => [Group([], 50)], + "property" => [match], + "regex" => [Group([Property("email", $"^(user|refresh-{generation})-[0-9]+@example[.]test$", "regex")])], + "cohort" => [Group([new { key = "id", value = 1, type = "cohort" }])], + "early" => [match, .. Enumerable.Repeat(miss, 7)], + "late" => [.. Enumerable.Repeat(miss, 7), match], + "no-match" => Enumerable.Repeat(miss, 8).ToArray(), + _ => [Group([])], + }; + var filters = new Dictionary { ["groups"] = groups }; + if (kind == "variant") + filters["multivariate"] = new { variants = new[] { new { key = "control", rollout_percentage = 50 }, new { key = "test", rollout_percentage = 50 } } }; + return new { id = i + 1, key = Key(i), name = Key(i), active = kind != "inactive" && (kind != "boolean" || generation % 2 == 0), filters }; + }).ToArray(); + return JsonSerializer.Serialize(new + { + flags, + group_type_mapping = new Dictionary(), + cohorts = new Dictionary + { + ["1"] = new { type = "AND", values = new[] { Property("plan", new[] { "pro" }), Property("score", generation % 2 == 0 ? -1 : 100, "gt") } }, + }, + }); + } + + // Independent fixture oracle: no SDK evaluator or private SDK state is used. + static double Bucket(string key, string user, string salt = "") + { + var hash = Convert.ToHexString(SHA1.HashData(Encoding.UTF8.GetBytes($"{key}.{user}{salt}"))); + return long.Parse(hash[..15], NumberStyles.HexNumber, CultureInfo.InvariantCulture) / Math.Pow(2, 60); + } + + internal static object Expected(int index, int user, int generation = 0) => Cases[index % Cases.Length] switch + { + "rollout" => Bucket(Key(index), Users[user]) < 0.5, + "variant" => Bucket(Key(index), Users[user], "variant") < 0.5 ? "control" : "test", + "boolean" => generation % 2 == 0, + "cohort" => generation % 2 == 0 && user % 2 == 0, + "property" or "early" or "late" => user % 2 == 0, + "no-match" or "inactive" => false, + _ => true, + }; + + internal static Workload[] Workloads(bool smoke) => + [ + .. Cases.Take(9).Select((kind, i) => new Workload($"flags.single.{kind}", "single", smoke ? 160 : 16000, ["timing", "allocation"], 10, i)), + .. new[] { 10, 100, 1000 }.Select(count => new Workload($"flags.all.{count}", "all", smoke ? 16 : Math.Max(160, 16000 / count), ["timing", "allocation"], count)), + new("flags.load.1000", "all", smoke ? 32 : 1600, ["memory"], 1000, Concurrency: 8), + new("flags.refresh.1000", "refresh", smoke ? 4 : 40, ["memory"], 1000), + .. new[] { 32, 4096 }.Select(size => new Workload($"capture.enqueue.{size}", "capture", smoke ? 200 : 1000, + ["timing", "allocation"], PayloadBytes: size, CompletionBoundary: "api_return")), + .. new[] { "steady", "burst" }.SelectMany(strategy => new[] { 32, 4096 }.Select(size => + new Workload($"capture.drain.{strategy}.{size}", "drain", smoke ? 200 : 1000, + ["timing", "allocation", "memory"], PayloadBytes: size, DrainStrategy: strategy, + FlushEvery: strategy == "steady" ? 100 : smoke ? 200 : 1000, CompletionBoundary: "completed_delivery"))), + ]; +} diff --git a/benchmarks/Measurement.cs b/benchmarks/Measurement.cs new file mode 100644 index 00000000..078866f7 --- /dev/null +++ b/benchmarks/Measurement.cs @@ -0,0 +1,78 @@ +using System.Diagnostics; + +namespace PostHog.Benchmarks; + +internal static class Measurement +{ + internal static void Collect() + { + GC.Collect(GC.MaxGeneration, GCCollectionMode.Forced, blocking: true, compacting: true); + GC.WaitForPendingFinalizers(); + GC.Collect(GC.MaxGeneration, GCCollectionMode.Forced, blocking: true, compacting: true); + } + + internal static long Rss(Process process) + { + process.Refresh(); + var bytes = process.WorkingSet64; + Program.Require(bytes > 0, "Current RSS is unavailable on this platform"); + return bytes; + } + + internal static async Task> Run(Scenario scenario, string pass) + { + await scenario.Prepare(); + var spec = scenario.Spec; + var result = new Dictionary + { + ["pass"] = pass, ["status"] = "ok", ["operations"] = spec.Operations, + ["warmup_operations"] = scenario.WarmupOperations, + }; + using var process = Process.GetCurrentProcess(); + if (pass == "timing") + { + var cpu = process.TotalProcessorTime; + var start = Stopwatch.GetTimestamp(); + await scenario.Execute(spec.Operations); + var elapsed = Stopwatch.GetElapsedTime(start).TotalNanoseconds; + result["elapsed_ns"] = elapsed; + result["elapsed_time_per_op"] = elapsed / spec.Operations; + result["process_cpu_time_per_op"] = (process.TotalProcessorTime - cpu).TotalNanoseconds / spec.Operations; + } + else if (pass == "allocation") + { + var before = GC.GetTotalAllocatedBytes(precise: true); + await scenario.Execute(spec.Operations); + result["allocated_bytes_per_op"] = (GC.GetTotalAllocatedBytes(precise: true) - before) / (double)spec.Operations; + } + else if (pass == "memory") + { + result["idle_managed_heap_bytes"] = GC.GetTotalMemory(forceFullCollection: false); + result["idle_process_rss"] = Rss(process); + var checkpoints = new List(50); + var peak = (long)result["idle_process_rss"]; + void Checkpoint(int completed) + { + var rss = Rss(process); + peak = Math.Max(peak, rss); + checkpoints.Add(new { operations = completed, rss_bytes = rss, managed_heap_bytes = GC.GetTotalMemory(false) }); + } + await scenario.Execute(spec.Operations, Checkpoint); + var completedRss = Rss(process); + result["after_operations_rss_bytes"] = completedRss; + result["observed_workload_peak_rss"] = Math.Max(peak, completedRss); + Collect(); + result["post_settling_heap_used"] = GC.GetTotalMemory(false); + result["post_settling_process_rss"] = Rss(process); + var lifetime = process.PeakWorkingSet64; + if (lifetime > 0) result["process_lifetime_peak_rss"] = lifetime; + result["checkpoints"] = checkpoints; + } + else throw new ArgumentException($"Unknown pass: {pass}"); + + // Correctness and delivery checks are outside the measurement boundary. + if (spec.FlagCount > 0) await scenario.VerifyFlags(); + else await scenario.VerifyCompletion(spec.Operations); + return result; + } +} diff --git a/benchmarks/Program.cs b/benchmarks/Program.cs new file mode 100644 index 00000000..104a7bc0 --- /dev/null +++ b/benchmarks/Program.cs @@ -0,0 +1,57 @@ +using System.Diagnostics; +using System.Runtime.InteropServices; +using System.Text.Json; + +namespace PostHog.Benchmarks; + +internal static class Program +{ + internal static readonly JsonSerializerOptions Json = new() { PropertyNamingPolicy = JsonNamingPolicy.SnakeCaseLower }; + internal static void Require(bool condition, string message) + { + if (!condition) throw new InvalidOperationException(message); + } + + static async Task Main(string[] args) + { + try + { + if (args[0] == "describe") + { + var entry = typeof(PostHogClient).Assembly.Location; + using var client = new PostHogClient(new PostHogOptions { ProjectToken = "benchmark-project-token", Disabled = true }); + Console.WriteLine(JsonSerializer.Serialize(new + { + runtime = ".NET", runtime_version = Environment.Version.ToString(), + os = RuntimeInformation.OSDescription, arch = RuntimeInformation.ProcessArchitecture.ToString(), + kernel = Environment.OSVersion.VersionString, entry, + sdk_version = client.Version, build_digest = Fixtures.Digest(File.ReadAllBytes(entry)), + measurement_tools = new { clr = Environment.Version.ToString() }, + workloads = Fixtures.Workloads(args.Contains("--smoke")).Select(spec => new + { + spec, + fixture_digest = Fixtures.FixtureDigest(spec), + }), + }, Json)); + } + else if (args[0] == "controls") await Controls.Run(); + else if (args[0] == "worker") + { + var spec = JsonSerializer.Deserialize(File.ReadAllText(args[1]), Json)!; + await using var scenario = new Scenario(spec); + var result = await Measurement.Run(scenario, args[2]); + var entry = typeof(PostHogClient).Assembly.Location; + result["entry"] = entry; + result["build_digest"] = Fixtures.Digest(File.ReadAllBytes(entry)); + await File.WriteAllTextAsync(args[3], JsonSerializer.Serialize(result, Json)); + } + else throw new ArgumentException("Expected describe, controls or worker"); + return 0; + } + catch (Exception error) + { + Console.Error.WriteLine(error); + return 1; + } + } +} diff --git a/benchmarks/Workload.cs b/benchmarks/Workload.cs new file mode 100644 index 00000000..61a1a4b8 --- /dev/null +++ b/benchmarks/Workload.cs @@ -0,0 +1,279 @@ +using System.Net; +using System.Text; +using System.Text.Json; +using PostHog.Features; + +namespace PostHog.Benchmarks; + +#pragma warning disable CS0618 // Measure supported legacy getters, matching the Node and Python suites. + +// Public HTTP injection keeps this offline. Parsing/validation is part of the drain +// workload, not an estimate of socket or remote ingestion performance. +internal sealed class Transport(Workload spec) : HttpMessageHandler, IHttpClientFactory +{ + readonly string _payload = new('x', spec.PayloadBytes); + readonly bool[] _seen = new bool[Math.Max(100, spec.Operations)]; + readonly object _deliveryLock = new(); + int _expected; + + internal void ResetDelivery(int operations) + { + lock (_deliveryLock) + { + Program.Require(operations <= _seen.Length, "Receiver capacity exceeded"); + Array.Clear(_seen); + _expected = operations; + Delivered = BatchRequests = UnexpectedRequests = 0; + Failure = null; + } + } + + internal void ValidateEvent(JsonElement item) + { + lock (_deliveryLock) + { + var properties = item.GetProperty("properties"); + var sequence = properties.GetProperty("sequence").GetInt32(); + Program.Require(item.GetProperty("event").GetString() == Fixtures.EventName, "Unexpected event"); + Program.Require(sequence >= 0 && sequence < _expected, "Sequence outside phase"); + Program.Require(properties.GetProperty("distinct_id").GetString() == Fixtures.Users[sequence % 16], "Wrong distinct ID"); + Program.Require(properties.GetProperty("payload").GetString() == _payload, "Wrong event payload"); + Program.Require(properties.GetProperty("plan").GetString() == "pro", "Wrong event plan"); + Program.Require(!_seen[sequence], "Duplicate sequence"); + if (!DropEvents) + { + _seen[sequence] = true; + Delivered++; + } + } + } + internal int Generation; + internal int DefinitionRequests; + internal int BatchRequests; + internal int Delivered; + internal int UnexpectedRequests; + internal string? Failure; + internal bool DropEvents; + + public HttpClient CreateClient(string name) => new(this, disposeHandler: false); + + protected override async Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) + { + if (request.Method == HttpMethod.Get && request.RequestUri?.AbsolutePath.TrimEnd('/') == "/flags/definitions") + { + Interlocked.Increment(ref DefinitionRequests); + return Response(Fixtures.Definitions(spec.FlagCount, Generation)); + } + if (request.Method == HttpMethod.Post && request.RequestUri?.AbsolutePath.TrimEnd('/') == "/batch") + { + Interlocked.Increment(ref BatchRequests); + try + { + Program.Require(!request.Content!.Headers.ContentEncoding.Any(), "Compression must be disabled"); + await using var content = await request.Content.ReadAsStreamAsync(cancellationToken); + using var body = await JsonDocument.ParseAsync(content, cancellationToken: cancellationToken); + foreach (var item in body.RootElement.GetProperty("batch").EnumerateArray()) ValidateEvent(item); + } + catch (Exception error) + { + Failure = error.Message; + throw; + } + return Response("{\"status\":1}"); + } + Interlocked.Increment(ref UnexpectedRequests); + throw new InvalidOperationException($"Unexpected request: {request.Method} {request.RequestUri?.AbsolutePath}"); + } + + static HttpResponseMessage Response(string json) => new(HttpStatusCode.OK) + { + Content = new StringContent(json, Encoding.UTF8, "application/json"), + }; +} + +internal sealed class Scenario : IAsyncDisposable +{ + internal readonly Workload Spec; + internal readonly Transport Transport; + internal readonly PostHogOptions Options; + internal readonly PostHogClient Client; + Dictionary[] _properties = []; + readonly string[] _keys; + internal long Checksum; + internal int WarmupOperations; + internal int ExplicitFlushes; + + internal Scenario(Workload spec) + { + Spec = spec; + Transport = new(spec); + Options = new PostHogOptions + { + ProjectToken = "benchmark-project-token", + SecretKey = spec.FlagCount > 0 ? "benchmark-secret" : null, + HostUrl = new Uri("http://benchmark.invalid"), + FeatureFlagPollInterval = TimeSpan.FromHours(1), + FlushInterval = TimeSpan.FromHours(1), + FlushAt = 100000, + MaxQueueSize = 100001, + MaxBatchSize = 100000, + EnableCompression = false, + }; + Client = new(Options, httpClientFactory: Transport); + _keys = Enumerable.Range(0, spec.FlagCount).Select(Fixtures.Key).ToArray(); + } + + internal async Task Prepare() + { + if (Spec.FlagCount > 0) + { + await Client.LoadFeatureFlagsAsync(CancellationToken.None); + await VerifyFlags(); + } + WarmupOperations = Spec.Mode switch { "single" => 1024, "all" => 16, "refresh" => 2, _ => 100 }; + if (Spec.PayloadBytes > 0) PrepareCapturePhase(WarmupOperations); + await Execute(WarmupOperations); + if (Spec.FlagCount > 0) + { + await VerifyFlags(); + Transport.Generation = 0; + await Client.LoadFeatureFlagsAsync(CancellationToken.None); + await VerifyFlags(); + } + else + { + await VerifyCompletion(WarmupOperations); + PrepareCapturePhase(Spec.Operations); + } + Checksum = 0; + ExplicitFlushes = 0; + await Task.Delay(25); // Let startup/background tasks reach their waits before measurement. + Measurement.Collect(); + } + + internal void PrepareCapturePhase(int operations) + { + // Each call owns its properties; submitted dictionaries are never mutated or reused. + _properties = Fixtures.CaptureProperties(Spec.PayloadBytes, operations); + Transport.ResetDelivery(operations); + } + + internal async Task Flush() + { + ExplicitFlushes++; + await Client.FlushAsync(); + } + + internal async Task VerifyCompletion(int operations) + { + if (Spec.Mode == "capture") await Flush(); + VerifyDelivery(operations); + } + + internal async Task Execute(int operations, Action? checkpoint = null, Action? flushed = null) + { + if (Spec.Mode is "capture" or "drain") + { + for (var i = 0; i < operations; i++) + { + Program.Require(Client.Capture(Fixtures.Users[i % 16], Fixtures.EventName, _properties[i]), "Capture rejected event"); + if (Spec.DrainStrategy == "steady" && ((i + 1) % Spec.FlushEvery == 0 || i + 1 == operations)) + { + await Flush(); + flushed?.Invoke(i + 1); + } + if ((i + 1) % 100 == 0 || i + 1 == operations) checkpoint?.Invoke(i + 1); + } + if (Spec.DrainStrategy == "burst") + { + await Flush(); + flushed?.Invoke(operations); + } + return; + } + if (Spec.Mode == "refresh") + { + for (var i = 0; i < operations; i++) + { + Transport.Generation++; + var requests = Transport.DefinitionRequests; + await Client.LoadFeatureFlagsAsync(CancellationToken.None); + Program.Require(Transport.DefinitionRequests == requests + 1, "Reload did not fetch new definitions"); + checkpoint?.Invoke(i + 1); + } + return; + } + if (Spec.Concurrency > 1) + { + // Task.Run is deliberate: cached async getters can complete synchronously. + // Merely collecting their Tasks would not produce overlapping evaluations. + var completed = 0; + var checkpointLock = new object(); + await Task.WhenAll(Enumerable.Range(0, Spec.Concurrency).Select(worker => Task.Run(async () => + { + long consumed = 0; + for (var i = worker; i < operations; i += Spec.Concurrency) + { + consumed += await Evaluate(i); + lock (checkpointLock) + { + completed++; + if (completed % 32 == 0 || completed == operations) checkpoint?.Invoke(completed); + } + } + Interlocked.Add(ref Checksum, consumed); + }))); + } + else + { + for (var i = 0; i < operations; i++) Checksum += await Evaluate(i); + } + Program.Require(Checksum > 0, "Evaluation results were not consumed"); + } + + async ValueTask Evaluate(int i) + { + var user = i % Fixtures.Users.Length; + if (Spec.Mode == "single") + { + var value = await Client.GetFeatureFlagAsync(_keys[Spec.FlagIndex], Fixtures.Users[user], Fixtures.Options[user]); + Program.Require(value is not null, "Missing local flag"); + return value!.VariantKey?.Length ?? (value.IsEnabled ? 2 : 1); + } + var flags = await Client.GetAllFeatureFlagsAsync(Fixtures.Users[user], Fixtures.Options[user]); + Program.Require(flags.Count == Spec.FlagCount, "Missing local flags"); + return flags.Count; + } + + void VerifyFlag(FeatureFlag? flag, int index, int user) + { + var expected = Fixtures.Expected(index, user, Transport.Generation); + Program.Require(flag is not null && flag.Key == _keys[index] && + (expected is string variant ? flag.IsEnabled && flag.VariantKey == variant : flag.VariantKey is null && flag.IsEnabled == (bool)expected), + $"Incorrect {_keys[index]} for user {user}, generation {Transport.Generation}"); + } + + internal async Task VerifyFlags() + { + for (var user = 0; user < Fixtures.Users.Length; user++) + { + var flags = await Client.GetAllFeatureFlagsAsync(Fixtures.Users[user], Fixtures.Options[user]); + Program.Require(flags.Count == Spec.FlagCount, + $"Wrong all-flags count: {flags.Count}/{Spec.FlagCount}; missing {string.Join(',', _keys.Except(flags.Keys))}; definition requests {Transport.DefinitionRequests}; unexpected requests {Transport.UnexpectedRequests}"); + for (var index = 0; index < Spec.FlagCount; index++) VerifyFlag(flags.GetValueOrDefault(_keys[index]), index, user); + if (Spec.Mode == "single") + VerifyFlag(await Client.GetFeatureFlagAsync(_keys[Spec.FlagIndex], Fixtures.Users[user], Fixtures.Options[user]), Spec.FlagIndex, user); + } + Program.Require(Transport.UnexpectedRequests == 0 && Transport.BatchRequests == 0, "Flags escaped local-only evaluation"); + } + + internal void VerifyDelivery(int operations) => Program.Require( + Transport.Failure is null && Transport.UnexpectedRequests == 0 && Transport.Delivered == operations, + $"Delivery mismatch: expected {operations}, received {Transport.Delivered}; {Transport.Failure}"); + + public async ValueTask DisposeAsync() + { + await Client.DisposeAsync(); + Transport.Dispose(); + } +} diff --git a/benchmarks/global.json b/benchmarks/global.json new file mode 100644 index 00000000..6d04aa60 --- /dev/null +++ b/benchmarks/global.json @@ -0,0 +1,6 @@ +{ + "sdk": { + "version": "9.0.109", + "rollForward": "disable" + } +} diff --git a/benchmarks/run.py b/benchmarks/run.py new file mode 100644 index 00000000..44d4107e --- /dev/null +++ b/benchmarks/run.py @@ -0,0 +1,411 @@ +"""Build the head-side CLR worker against each SDK, then run isolated paired passes.""" + +import argparse +import copy +import hashlib +import json +import math +import os +import platform +import shutil +import subprocess +import tempfile +import uuid +from pathlib import Path + +HERE = Path(__file__).resolve().parent +TIMEOUT = 120 +MAX_TIMEOUTS = 2 + + +def digest(value): + return hashlib.sha256(value).hexdigest() + + +def git(directory, *args): + return subprocess.check_output( + ["git", "-C", str(directory), *args], text=True + ).strip() + + +def environment(): + # Keep machine-specific CLR tuning and diagnostics out of the paired workers. + env = { + k: v + for k, v in os.environ.items() + if not k.startswith(("DOTNET_", "COMPlus_", "CORECLR_", "COR_")) + } + env.update( + DOTNET_TieredCompilation="0", + DOTNET_gcServer="0", + DOTNET_GCHeapHardLimit="20000000", + DOTNET_CLI_TELEMETRY_OPTOUT="1", + ) + return env + + +def cpu_model(): + if platform.system() == "Linux": + for line in Path("/proc/cpuinfo").read_text().splitlines(): + if line.startswith("model name"): + return line.split(":", 1)[1].strip() + if platform.system() == "Darwin": + return subprocess.check_output( + ["sysctl", "-n", "machdep.cpu.brand_string"], text=True + ).strip() + return platform.processor() or platform.machine() + + +def prepare(source, directory): + project = directory / "project" + project.mkdir(parents=True, exist_ok=False) + for path in sorted(HERE.iterdir()): + if path.suffix in (".cs", ".csproj", ".json"): + shutil.copy2(path, project / path.name) + output = directory / "out" + subprocess.run( + [ + "dotnet", + "publish", + str(project / "Benchmark.csproj"), + "--configuration", + "Release", + f"-p:PostHogProject={source / 'src/PostHog/PostHog.csproj'}", + "-p:GeneratePackageOnBuild=false", + "-p:RestoreLockedMode=true", + "--output", + str(output), + ], + check=True, + env=environment(), + cwd=project, + ) + return output / "Benchmark.dll" + + +def describe(worker, smoke): + return json.loads( + subprocess.check_output( + ["dotnet", str(worker), "describe", *(["--smoke"] if smoke else [])], + env=environment(), + text=True, + timeout=TIMEOUT, + ) + ) + + +def run_worker(worker, spec, measurement_pass, timeout=TIMEOUT): + with tempfile.TemporaryDirectory(prefix="posthog-dotnet-worker-") as temporary: + root = Path(temporary) + (root / "spec.json").write_text(json.dumps(spec)) + output = root / "result.json" + # Files keep diagnostic output from growing the coordinator's memory. + with (root / "stdout").open("w+b") as stdout, (root / "stderr").open( + "w+b" + ) as stderr: + with subprocess.Popen( + [ + "dotnet", + str(worker), + "worker", + str(root / "spec.json"), + measurement_pass, + str(output), + ], + cwd=root, + env=environment(), + stdout=stdout, + stderr=stderr, + ) as process: + timed_out = False + try: + process.wait(timeout=timeout) + except subprocess.TimeoutExpired: + timed_out = True + process.kill() + process.wait() + logs = {} + for name, stream in (("stdout", stdout), ("stderr", stderr)): + stream.seek(0, 2) + stream.seek(max(0, stream.tell() - 65536)) + logs[name] = stream.read().decode("utf-8", errors="replace") + if not timed_out and process.returncode == 0 and output.exists(): + if output.stat().st_size > 2 * 1024 * 1024: + raise ValueError("Worker result exceeds size limit") + result = json.loads(output.read_text()) + if ( + result.get("status") != "ok" + or result.get("pass") != measurement_pass + or result.get("operations") != spec["operations"] + ): + raise ValueError("Invalid worker result identity") + return {**result, **logs} + return { + "status": "timeout" if timed_out else "error", + "pass": measurement_pass, + "code": process.returncode, + "error": ( + "Worker exceeded time limit" + if timed_out + else "Worker did not complete; exit alone does not establish OOM" + ), + **logs, + } + + +def append_sample(benchmark, raw, repetition): + benchmark["runs"].append({"repetition": repetition, **raw}) + if raw["status"] != "ok": + benchmark["status"] = raw["status"] + benchmark["reason"] = "One or more isolated workers failed; inspect raw runs." + return + definitions = { + "elapsed_time_per_op": ("ns/op", "Stopwatch.GetTimestamp", "measured"), + "process_cpu_time_per_op": ("ns/op", "Process.TotalProcessorTime", "measured"), + "allocated_bytes_per_op": ( + "bytes/op", + "GC.GetTotalAllocatedBytes(precise:true)", + "measured", + ), + "idle_managed_heap_bytes": ( + "bytes", + "GC.GetTotalMemory(false) after full GC", + "estimated", + ), + "post_settling_heap_used": ( + "bytes", + "GC.GetTotalMemory(false) after full GC", + "estimated", + ), + "idle_process_rss": ("bytes", "Process.WorkingSet64", "measured"), + "post_settling_process_rss": ( + "bytes", + "Process.WorkingSet64 after full GC", + "measured", + ), + "observed_workload_peak_rss": ( + "bytes", + "Process.WorkingSet64 maximum of before-work, checkpoints and after-work", + "measured", + ), + "process_lifetime_peak_rss": ( + "bytes", + "Process.PeakWorkingSet64 including startup and warmup", + "measured", + ), + } + for name, (unit, method, quality) in definitions.items(): + if name not in raw: + continue + value = raw[name] + if ( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or not math.isfinite(value) + or value < 0 + ): + raise ValueError(f"Invalid {name}: {value}") + metric = next((m for m in benchmark["metrics"] if m["id"] == name), None) + if metric is None: + metric = { + "id": name, + "unit": unit, + "method": method, + "quality": quality, + "direction": "lower", + "status": "ok", + "samples": [], + "scope": ( + "CLR managed retained-heap estimate after full GC; excludes native memory, not allocation volume or RSS." + if name in ("idle_managed_heap_bytes", "post_settling_heap_used") + else ( + "Cumulative CLR managed allocations within the boundary, including in-process transport; excludes native allocations." + if name == "allocated_bytes_per_op" + else "Whole worker process; includes managed runtime and in-process offline transport work within the measured boundary." + ) + ), + } + benchmark["metrics"].append(metric) + metric["samples"].append( + { + "repetition": repetition, + "value": value, + "operations": raw["operations"], + **({"elapsed_ns": raw["elapsed_ns"]} if "elapsed_ns" in raw else {}), + } + ) + if raw["pass"] == "memory" and "process_lifetime_peak_rss" not in raw: + unavailable( + benchmark, + "process_lifetime_peak_rss", + "bytes", + "Runtime/OS did not expose a positive lifetime RSS peak.", + ) + + +def unavailable(benchmark, name, unit, reason): + if not any(m["id"] == name for m in benchmark["metrics"]): + benchmark["metrics"].append( + {"id": name, "unit": unit, "status": "unsupported", "reason": reason} + ) + + +def run_pairs(documents, workers, workloads, repetitions, output, execute=run_worker): + timeouts = 0 + failed = False + for item in workloads: + spec = item["spec"] + for document in documents.values(): + benchmark = { + "id": spec["id"], + "fixture_digest": item["fixture_digest"], + "parameters": spec, + "status": "ok", + "metrics": [], + "runs": [], + } + if "allocation" in spec["passes"]: + unavailable( + benchmark, + "allocations_per_op", + "count/op", + "CLR byte counters do not expose exact object allocation counts; native allocations are not included.", + ) + document["benchmarks"].append(benchmark) + for measurement_pass in spec["passes"]: + for repetition in range(repetitions): + for role in ( + ["base", "head"] if repetition % 2 == 0 else ["head", "base"] + ): + try: + raw = execute(workers[role], spec, measurement_pass) + if ( + raw["status"] == "ok" + and raw.get("build_digest") + != documents[role]["execution"]["build_digest"] + ): + raise ValueError("Worker loaded a different SDK build") + benchmark = copy.deepcopy(documents[role]["benchmarks"][-1]) + append_sample(benchmark, raw, repetition) + documents[role]["benchmarks"][-1] = benchmark + except (ValueError, OSError) as error: + raw = { + "status": "error", + "pass": measurement_pass, + "error": str(error)[:4096], + } + append_sample( + documents[role]["benchmarks"][-1], raw, repetition + ) + failed |= raw["status"] != "ok" + timeouts += raw["status"] == "timeout" + (output / f"{role}.json").write_text( + json.dumps(documents[role], indent=2) + "\n" + ) + print( + f"{spec['id']} {measurement_pass} {repetition} {role}: {raw['status']}", + flush=True, + ) + # Finish the pair, then stop; successful partial samples cannot hide failure. + if timeouts >= MAX_TIMEOUTS: + return 1 + return int(failed) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--base", required=True, type=Path) + parser.add_argument("--head", required=True, type=Path) + parser.add_argument("--output", required=True, type=Path) + parser.add_argument("--build-directory", required=True, type=Path) + parser.add_argument("--repetitions", type=int, default=6) + parser.add_argument("--smoke", action="store_true") + args = parser.parse_args() + if not 1 <= args.repetitions <= 100: + parser.error("repetitions must be between 1 and 100") + args.output = args.output.resolve() + args.output.mkdir(parents=True, exist_ok=True) + pair_id = str(uuid.uuid4()) + suite_sources = b"".join( + path.name.encode() + b"\0" + path.read_bytes() + for path in sorted(HERE.iterdir()) + if path.suffix in (".cs", ".csproj", ".py", ".json") + ) + suite = { + "id": "posthog-dotnet-offline", + "version": 2, + "specification": "sdk-benchmarks/v1", + "source_commit": git(HERE, "rev-parse", "HEAD"), + "source_digest": digest(suite_sources), + "profile": "smoke" if args.smoke else "ci", + "repetitions": args.repetitions, + } + documents, workers, inventories = {}, {}, {} + for role in ("base", "head"): + source = getattr(args, role).resolve() + worker = prepare(source, args.build_directory.resolve() / role) + workers[role] = worker + execution = describe(worker, args.smoke) + inventories[role] = execution.pop("workloads") + execution.update( + { + "pair_id": pair_id, + "runtime_flags": [ + "TieredCompilation=false", + "ServerGarbageCollection=false", + "GCHeapHardLimit=536870912", + ], + "cpu": cpu_model(), + "runner_class": ( + "github-hosted-or-self-hosted" + if os.getenv("RUNNER_NAME") + else "local" + ), + "runner_image": os.getenv("ImageVersion"), + "dependency_lock_digest": digest( + (source / "src/PostHog/packages.lock.json").read_bytes() + ), + "packages": { + "application_dependency_graph_sha256": digest( + worker.with_suffix(".deps.json").read_bytes() + ), + }, + "dotnet_sdk_version": subprocess.check_output( + ["dotnet", "--version"], cwd=HERE, text=True + ).strip(), + "resource_limits": { + "worker_timeout_ms": TIMEOUT * 1000, + "max_worker_timeouts": MAX_TIMEOUTS, + "managed_heap_limit_bytes": 512 * 1024 * 1024, + }, + } + ) + documents[role] = { + "schema_version": 1, + "sdk": "posthog-dotnet", + "revision": { + "role": role, + "commit": git(source, "rev-parse", "HEAD"), + "dirty": bool(git(source, "status", "--porcelain")), + }, + "suite": suite, + "execution": execution, + "benchmarks": [], + } + if inventories["base"] != inventories["head"]: + raise ValueError( + "The two builds produced different fixture/workload identities" + ) + subprocess.run( + ["dotnet", str(workers["head"]), "controls"], + check=True, + env=environment(), + timeout=TIMEOUT, + ) + return run_pairs( + documents, workers, inventories["head"], args.repetitions, args.output + ) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/test_run.py b/benchmarks/test_run.py new file mode 100644 index 00000000..fe08f0ae --- /dev/null +++ b/benchmarks/test_run.py @@ -0,0 +1,269 @@ +import json +import os +import subprocess +import tempfile +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +import run + + +class CoordinatorTests(unittest.TestCase): + def setUp(self): + self.item = { + "spec": {"id": "control", "operations": 10, "passes": ["timing"]}, + "fixture_digest": "a" * 64, + } + self.documents = { + role: {"execution": {"build_digest": role}, "benchmarks": []} + for role in ("base", "head") + } + + def execute(self, role, spec, measurement_pass): + return { + "status": "ok", + "pass": measurement_pass, + "operations": 10, + "build_digest": role, + "elapsed_time_per_op": 123, + "elapsed_ns": 1230, + } + + def test_alternates_pairs_and_persists(self): + calls = [] + + def execute(role, spec, measurement_pass): + calls.append(role) + return self.execute(role, spec, measurement_pass) + + with tempfile.TemporaryDirectory() as directory: + status = run.run_pairs( + self.documents, + {r: r for r in self.documents}, + [self.item], + 3, + Path(directory), + execute, + ) + self.assertEqual(status, 0) + self.assertEqual(calls, ["base", "head", "head", "base", "base", "head"]) + for role in self.documents: + document = json.loads((Path(directory) / f"{role}.json").read_text()) + self.assertEqual( + len(document["benchmarks"][0]["metrics"][0]["samples"]), 3 + ) + + def test_timeout_budget_finishes_pair_and_stops(self): + calls = [] + + def execute(role, spec, measurement_pass): + calls.append(role) + return { + "status": "timeout" if role == "head" else "ok", + "pass": measurement_pass, + "operations": 10, + "build_digest": role, + "elapsed_time_per_op": 123, + } + + with tempfile.TemporaryDirectory() as directory: + status = run.run_pairs( + self.documents, + {r: r for r in self.documents}, + [self.item] * 2, + 6, + Path(directory), + execute, + ) + self.assertEqual(status, 1) + self.assertEqual(calls, ["base", "head", "head", "base"]) + self.assertEqual(len(self.documents["base"]["benchmarks"]), 1) + self.assertEqual(len(self.documents["base"]["benchmarks"][0]["runs"]), 2) + self.assertEqual( + self.documents["head"]["benchmarks"][0]["status"], "timeout" + ) + + def test_wrong_build_is_failure(self): + def execute(role, spec, measurement_pass): + return { + **self.execute(role, spec, measurement_pass), + "build_digest": "wrong", + } + + with tempfile.TemporaryDirectory() as directory: + self.assertEqual( + run.run_pairs( + self.documents, + {r: r for r in self.documents}, + [self.item], + 1, + Path(directory), + execute, + ), + 1, + ) + self.assertEqual(self.documents["head"]["benchmarks"][0]["metrics"], []) + + def test_invalid_measurement_does_not_leave_successful_run(self): + def execute(role, spec, measurement_pass): + return { + **self.execute(role, spec, measurement_pass), + "allocated_bytes_per_op": float("nan"), + } + + with tempfile.TemporaryDirectory() as directory: + self.assertEqual( + run.run_pairs( + self.documents, + {r: r for r in self.documents}, + [self.item], + 1, + Path(directory), + execute, + ), + 1, + ) + benchmark = self.documents["head"]["benchmarks"][0] + self.assertEqual(benchmark["metrics"], []) + self.assertEqual([r["status"] for r in benchmark["runs"]], ["error"]) + + def test_missing_rss_is_unsupported_not_zero(self): + benchmark = {"status": "ok", "metrics": [], "runs": []} + run.append_sample( + benchmark, {"status": "ok", "pass": "memory", "operations": 10}, 0 + ) + self.assertEqual(benchmark["metrics"][0]["status"], "unsupported") + self.assertNotIn("samples", benchmark["metrics"][0]) + + def test_common_memory_metrics_and_managed_heap_scope(self): + benchmark = {"status": "ok", "metrics": [], "runs": []} + names = ( + "idle_process_rss", + "observed_workload_peak_rss", + "process_lifetime_peak_rss", + "post_settling_process_rss", + "post_settling_heap_used", + ) + run.append_sample( + benchmark, + { + "status": "ok", + "pass": "memory", + "operations": 32, + **{name: 4096 for name in names}, + }, + 0, + ) + metrics = {metric["id"]: metric for metric in benchmark["metrics"]} + self.assertEqual(set(metrics), set(names)) + heap = metrics["post_settling_heap_used"] + self.assertEqual(heap["quality"], "estimated") + self.assertIn("excludes native memory", heap["scope"]) + self.assertIn("not allocation volume or RSS", heap["scope"]) + self.assertEqual( + metrics["observed_workload_peak_rss"]["method"], + "Process.WorkingSet64 maximum of before-work, checkpoints and after-work", + ) + + def test_environment_removes_external_runtime_tuning(self): + with patch.dict( + os.environ, + { + "COMPlus_TieredCompilation": "1", + "DOTNET_GCHeapHardLimit": "1", + "CORECLR_ENABLE_PROFILING": "1", + }, + ): + env = run.environment() + self.assertNotIn("COMPlus_TieredCompilation", env) + self.assertNotIn("CORECLR_ENABLE_PROFILING", env) + self.assertEqual(env["DOTNET_GCHeapHardLimit"], "20000000") + + def test_describe_has_deadline_and_propagates_timeout(self): + with patch.object( + run.subprocess, + "check_output", + side_effect=subprocess.TimeoutExpired("dotnet", run.TIMEOUT), + ) as describe: + with self.assertRaises(subprocess.TimeoutExpired): + run.describe(Path("worker.dll"), False) + self.assertEqual(describe.call_args.kwargs["timeout"], run.TIMEOUT) + + def test_describe_returns_metadata(self): + with patch.object( + run.subprocess, "check_output", return_value='{"runtime": ".NET"}' + ): + self.assertEqual( + run.describe(Path("worker.dll"), True), {"runtime": ".NET"} + ) + + def test_timeout_kills_and_reaps_worker(self): + process = MagicMock() + process.__enter__.return_value = process + process.wait.side_effect = [subprocess.TimeoutExpired("dotnet", 0.01), None] + process.returncode = -9 + with patch.object(run.subprocess, "Popen", return_value=process): + raw = run.run_worker( + Path("unused"), self.item["spec"], "timing", timeout=0.01 + ) + process.kill.assert_called_once() + self.assertEqual(process.wait.call_count, 2) + self.assertEqual(raw["status"], "timeout") + + def test_dependency_graph_is_sdk_metadata(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + sdk = root / "src/PostHog" + sdk.mkdir(parents=True) + (sdk / "packages.lock.json").write_text("{}") + worker = root / "Benchmark.dll" + worker.with_suffix(".deps.json").write_text('{"PostHog": "2.17.1"}') + with ( + patch( + "sys.argv", + [ + "run.py", + "--base", + directory, + "--head", + directory, + "--output", + str(root / "results"), + "--build-directory", + str(root / "build"), + ], + ), + patch.object(run, "prepare", return_value=worker), + patch.object(run, "describe", side_effect=lambda *_: {"workloads": []}), + patch.object(run, "git", return_value="a" * 40), + patch.object(run, "cpu_model", return_value="control CPU"), + patch.object(run.subprocess, "check_output", return_value="9.0.109"), + patch.object(run.subprocess, "run"), + patch.object(run, "run_pairs", return_value=0) as pairs, + ): + self.assertEqual(run.main(), 0) + suite = pairs.call_args.args[0]["head"]["suite"] + self.assertEqual(suite["version"], 2) + self.assertEqual(suite["specification"], "sdk-benchmarks/v1") + execution = pairs.call_args.args[0]["head"]["execution"] + self.assertEqual( + execution["packages"]["application_dependency_graph_sha256"], + run.digest(worker.with_suffix(".deps.json").read_bytes()), + ) + # The reporter excludes packages from environment compatibility, but + # treats unknown top-level execution fields as environment identity. + self.assertNotIn("application_dependency_digest", execution) + + def test_signal_exit_is_not_proof_of_oom(self): + process = MagicMock() + process.__enter__.return_value = process + process.returncode = -9 + with patch.object(run.subprocess, "Popen", return_value=process): + raw = run.run_worker(Path("unused"), self.item["spec"], "timing") + self.assertEqual(raw["status"], "error") + process.kill.assert_not_called() + + +if __name__ == "__main__": + unittest.main()