Misc. Benchmark Updates

.gitignore

@@ -56,6 +56,8 @@ dlldata.c
 
 # Benchmark Results
 BenchmarkDotNet.Artifacts/
+benchmarks/results/
+benchmarks/results-railway/
 
 # .NET Core
 project.lock.json

PostgreSignalR.slnx

@@ -10,6 +10,7 @@
     <File Path="benchmarks/docker-compose.yml" />
     <Project Path="benchmarks/PostgreSignalR.Benchmarks.Abstractions/PostgreSignalR.Benchmarks.Abstractions.csproj" />
     <Project Path="benchmarks/PostgreSignalR.Benchmarks.Server/PostgreSignalR.Benchmarks.Server.csproj" />
+    <Project Path="benchmarks/PostgreSignalR.Benchmarks.SharedLoad/PostgreSignalR.Benchmarks.SharedLoad.csproj" />
     <Project Path="benchmarks/PostgreSignalR.Benchmarks/PostgreSignalR.Benchmarks.csproj" />
   </Folder>
 

benchmarks/PostgreSignalR.Benchmarks.Abstractions/ConnectionStringHelper.cs

@@ -0,0 +1,94 @@
+using Npgsql;
+using StackExchange.Redis;
+
+namespace PostgreSignalR.Benchmarks.Abstractions;
+
+public static class ConnectionStringHelper
+{
+    private static bool HasUriScheme(string value, params string[] schemes) =>
+        schemes.Any(scheme => value.StartsWith(scheme + "://", StringComparison.OrdinalIgnoreCase));
+
+    public static string NormalizePostgres(string value)
+    {
+        if (!HasUriScheme(value, "postgres", "postgresql"))
+        {
+            return value;
+        }
+
+        var uri = new Uri(value);
+
+        var builder = new NpgsqlConnectionStringBuilder
+        {
+            Host = uri.Host,
+            Port = uri.IsDefaultPort ? 5432 : uri.Port,
+            Database = uri.AbsolutePath.TrimStart('/'),
+        };
+
+        var userInfo = uri.UserInfo.Split(':', 2);
+
+        if (userInfo.Length > 0 && userInfo[0].Length > 0)
+        {
+            builder.Username = Uri.UnescapeDataString(userInfo[0]);
+        }
+
+        if (userInfo.Length > 1 && userInfo[1].Length > 0)
+        {
+            builder.Password = Uri.UnescapeDataString(userInfo[1]);
+        }
+
+        foreach (var pair in uri.Query.TrimStart('?').Split('&', StringSplitOptions.RemoveEmptyEntries))
+        {
+            var parts = pair.Split('=', 2);
+
+            if (parts.Length != 2)
+            {
+                continue;
+            }
+
+            var key = Uri.UnescapeDataString(parts[0]);
+            var val = Uri.UnescapeDataString(parts[1]);
+
+            if (key.Equals("sslmode", StringComparison.OrdinalIgnoreCase))
+            {
+                builder.SslMode = Enum.Parse<SslMode>(val, ignoreCase: true);
+            }
+            else
+            {
+                builder[key] = val;
+            }
+        }
+
+        return builder.ConnectionString;
+    }
+
+    public static string NormalizeRedis(string value)
+    {
+        if (!HasUriScheme(value, "redis", "rediss"))
+        {
+            return value;
+        }
+
+        var uri = new Uri(value);
+
+        var options = new ConfigurationOptions
+        {
+            Ssl = uri.Scheme == "rediss",
+        };
+
+        options.EndPoints.Add(uri.Host, uri.IsDefaultPort ? 6379 : uri.Port);
+
+        var userInfo = uri.UserInfo.Split(':', 2);
+
+        if (userInfo.Length > 0 && userInfo[0].Length > 0)
+        {
+            options.User = Uri.UnescapeDataString(userInfo[0]);
+        }
+
+        if (userInfo.Length > 1 && userInfo[1].Length > 0)
+        {
+            options.Password = Uri.UnescapeDataString(userInfo[1]);
+        }
+
+        return options.ToString();
+    }
+}

benchmarks/PostgreSignalR.Benchmarks.Abstractions/Message.cs

@@ -3,5 +3,7 @@
 public record Message(
     string MessageId,
     long SentUnixTimeMs,
-    int PayloadBytes
+    int PayloadBytes,
+    string Payload,
+    long Generation
 );

benchmarks/PostgreSignalR.Benchmarks.Abstractions/PostgreSignalR.Benchmarks.Abstractions.csproj

@@ -6,4 +6,9 @@
     <Nullable>enable</Nullable>
   </PropertyGroup>
 
+  <ItemGroup>
+    <PackageReference Include="Npgsql" Version="10.0.3" />
+    <PackageReference Include="StackExchange.Redis" Version="2.7.27" />
+  </ItemGroup>
+
 </Project>

benchmarks/PostgreSignalR.Benchmarks.Collector/Dockerfile

@@ -0,0 +1,19 @@
+FROM mcr.microsoft.com/dotnet/sdk:10.0 AS build
+WORKDIR /src
+
+# Copy the rest of the source
+COPY benchmarks/ benchmarks/
+
+# Restore just the collector project
+RUN dotnet restore benchmarks/PostgreSignalR.Benchmarks.Collector/PostgreSignalR.Benchmarks.Collector.csproj
+
+# Publish
+RUN dotnet publish benchmarks/PostgreSignalR.Benchmarks.Collector/PostgreSignalR.Benchmarks.Collector.csproj -c Release -o /out --no-restore
+
+FROM mcr.microsoft.com/dotnet/aspnet:10.0
+WORKDIR /app
+
+COPY --from=build /out .
+ENV ASPNETCORE_URLS=http://0.0.0.0:8080
+EXPOSE 8080
+ENTRYPOINT ["dotnet", "PostgreSignalR.Benchmarks.Collector.dll"]

benchmarks/PostgreSignalR.Benchmarks.Collector/PostgreSignalR.Benchmarks.Collector.csproj

@@ -0,0 +1,9 @@
+<Project Sdk="Microsoft.NET.Sdk.Web">
+
+  <PropertyGroup>
+    <TargetFramework>net10.0</TargetFramework>
+    <Nullable>enable</Nullable>
+    <ImplicitUsings>enable</ImplicitUsings>
+  </PropertyGroup>
+
+</Project>

benchmarks/PostgreSignalR.Benchmarks.Collector/Program.cs

@@ -0,0 +1,39 @@
+using System.Collections.Concurrent;
+
+var results = new ConcurrentDictionary<string, string>();
+
+var app = WebApplication.CreateBuilder(args).Build();
+
+app.MapPost("/results/{key}", async (string key, HttpRequest request) =>
+{
+    using var reader = new StreamReader(request.Body);
+    results[key] = await reader.ReadToEndAsync();
+    return Results.Ok();
+});
+
+app.MapGet("/results/{key}", async (string key, HttpContext context) =>
+{
+    var deadline = DateTime.UtcNow.AddSeconds(
+        context.Request.Query.TryGetValue("waitSeconds", out var waitSecondsRaw)
+        && int.TryParse(waitSecondsRaw, out var parsedWaitSeconds)
+            ? parsedWaitSeconds
+            : 0
+    );
+
+    while (true)
+    {
+        if (results.TryGetValue(key, out var body))
+        {
+            return Results.Text(body);
+        }
+
+        if (DateTime.UtcNow >= deadline)
+        {
+            return Results.NotFound();
+        }
+
+        await Task.Delay(250, context.RequestAborted);
+    }
+});
+
+app.Run();

benchmarks/PostgreSignalR.Benchmarks.Server/Program.cs

@@ -1,50 +1,60 @@
 using PostgreSignalR.Benchmarks.Abstractions;
 using PostgreSignalR.Benchmarks.Server;
 using Microsoft.AspNetCore.SignalR;
+using PostgreSignalR;
 
 var builder = WebApplication.CreateBuilder(args);
 
 builder.Services.AddRouting();
 builder.Services.AddSignalR();
 
-var backplane = (Environment.GetEnvironmentVariable("BACKPLANE") ?? throw new Exception()).ToLowerInvariant();
+var backplane = (Environment.GetEnvironmentVariable("BACKPLANE") ?? "redis").ToLowerInvariant();
 
-// Uncomment if running benchmarks with payload table
-//var instantiatePayloadTable = bool.Parse(Environment.GetEnvironmentVariable("MAKETABLE") ?? "false");
+var usePayloadTable = (Environment.GetEnvironmentVariable("PAYLOAD_STRATEGY") ?? "event").Equals("table", StringComparison.OrdinalIgnoreCase);
 
 if (backplane is "redis")
 {
-    var connectionString = Environment.GetEnvironmentVariable("ConnectionStrings__Redis") ?? throw new Exception();
+    var connectionString = ConnectionStringHelper.NormalizeRedis(Environment.GetEnvironmentVariable("ConnectionStrings__Redis") ?? throw new Exception("ConnectionStrings__Redis is required when BACKPLANE=redis but was not set."));
     builder.Services.AddSignalR().AddStackExchangeRedis(connectionString);
 }
 else if (backplane is "postgres")
 {
-    var connectionString = Environment.GetEnvironmentVariable("ConnectionStrings__Postgres") ?? throw new Exception();
-    builder.Services.AddSignalR().AddPostgresBackplane(connectionString);
-
-    // Uncomment if running benchmarks with payload table
-    // .AddBackplaneTablePayloadStrategy(o =>
-    // {
-    //     o.AutomaticCleanup = false;
-    //     o.StorageMode = PostgreSignalR.PostgresBackplanePayloadTableStorage.Always;
-    // });
+    var connectionString = ConnectionStringHelper.NormalizePostgres(Environment.GetEnvironmentVariable("ConnectionStrings__Postgres") ?? throw new Exception("ConnectionStrings__Postgres is required when BACKPLANE=postgres but was not set."));
+    var signalrBuilder = builder.Services.AddSignalR().AddPostgresBackplane(connectionString);
+
+    if (usePayloadTable)
+    {
+        signalrBuilder.AddBackplaneTablePayloadStrategy(o =>
+        {
+            o.AutomaticCleanup = false;
+            o.StorageMode = PostgresBackplanePayloadTableStorage.Always;
+        });
+    }
 }
 
 var app = builder.Build();
 
-// Uncomment if running benchmarks with payload table
-// if (backplane is "postgres")
-// {
-//     await app.InitializePostgresBackplanePayloadTableAsync();
-// }
+if (backplane is "postgres" && usePayloadTable)
+{
+    await app.InitializePostgresBackplanePayloadTableAsync();
+}
+
+SemaphoreSlim? publishSemaphore = null;
 
 app.MapHub<BenchmarkHub>("/hub");
 
-app.MapGet("/health", () => Results.Ok(new { ok = true, backplane }));
+app.MapGet("/health", () => Results.Ok(new { ok = true, backplane, payloadStrategy = usePayloadTable ? "table" : "event" }));
+
+app.MapGet("/time", () => Results.Ok(new { unixTimeMs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() }));
 
 app.MapPost("/publish", async (PublishRequest request, IHubContext<BenchmarkHub> hub, CancellationToken c) =>
 {
-    var semaphore = new SemaphoreSlim(request.Concurrency, request.Concurrency);
+    var semaphore = LazyInitializer.EnsureInitialized(
+        ref publishSemaphore,
+        () => new SemaphoreSlim(request.Concurrency, request.Concurrency)
+    );
+
+    var payload = request.PayloadBytes > 0 ? new string('x', request.PayloadBytes) : string.Empty;
     var sendTasks = new List<Task>(request.PublishCount);
 
     for (int i = 0; i < request.PublishCount; i++)
@@ -54,7 +64,9 @@
         var message = new Message(
             MessageId: Guid.NewGuid().ToString("N"),
             SentUnixTimeMs: DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
-            PayloadBytes: request.PayloadBytes
+            PayloadBytes: request.PayloadBytes,
+            Payload: payload,
+            Generation: request.Generation
         );
 
         sendTasks.Add(hub.Clients.All.SendAsync("bench", message, c).ContinueWith(

benchmarks/PostgreSignalR.Benchmarks.Server/PublishRequest.cs

@@ -1,5 +1,8 @@
+namespace PostgreSignalR.Benchmarks.Server;
+
 public sealed record PublishRequest(
     int PublishCount,
     int Concurrency,
-    int PayloadBytes
+    int PayloadBytes,
+    long Generation
 );

benchmarks/PostgreSignalR.Benchmarks.SharedLoad/Dockerfile

@@ -0,0 +1,17 @@
+FROM mcr.microsoft.com/dotnet/sdk:10.0 AS build
+WORKDIR /src
+
+# Copy the rest of the source
+COPY benchmarks/ benchmarks/
+
+# Restore just the shared-load project
+RUN dotnet restore benchmarks/PostgreSignalR.Benchmarks.SharedLoad/PostgreSignalR.Benchmarks.SharedLoad.csproj
+
+# Publish
+RUN dotnet publish benchmarks/PostgreSignalR.Benchmarks.SharedLoad/PostgreSignalR.Benchmarks.SharedLoad.csproj -c Release -o /out --no-restore
+
+FROM mcr.microsoft.com/dotnet/runtime:10.0
+WORKDIR /app
+
+COPY --from=build /out .
+ENTRYPOINT ["dotnet", "PostgreSignalR.Benchmarks.SharedLoad.dll"]

benchmarks/PostgreSignalR.Benchmarks.SharedLoad/PostgreSignalR.Benchmarks.SharedLoad.csproj

@@ -0,0 +1,19 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net10.0</TargetFramework>
+    <Nullable>enable</Nullable>
+    <ImplicitUsings>enable</ImplicitUsings>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Npgsql" Version="10.0.3" />
+    <PackageReference Include="StackExchange.Redis" Version="2.7.27" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="../PostgreSignalR.Benchmarks.Abstractions/PostgreSignalR.Benchmarks.Abstractions.csproj" />
+  </ItemGroup>
+
+</Project>