diff --git a/.gitignore b/.gitignore index dc4496d..63bd6c9 100644 --- a/.gitignore +++ b/.gitignore @@ -56,6 +56,8 @@ dlldata.c # Benchmark Results BenchmarkDotNet.Artifacts/ +benchmarks/results/ +benchmarks/results-railway/ # .NET Core project.lock.json diff --git a/PostgreSignalR.slnx b/PostgreSignalR.slnx index 2e4fb53..b22c1d8 100644 --- a/PostgreSignalR.slnx +++ b/PostgreSignalR.slnx @@ -10,6 +10,7 @@ + diff --git a/benchmarks/PostgreSignalR.Benchmarks.Abstractions/ConnectionStringHelper.cs b/benchmarks/PostgreSignalR.Benchmarks.Abstractions/ConnectionStringHelper.cs new file mode 100644 index 0000000..ca34ccd --- /dev/null +++ b/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(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(); + } +} diff --git a/benchmarks/PostgreSignalR.Benchmarks.Abstractions/Message.cs b/benchmarks/PostgreSignalR.Benchmarks.Abstractions/Message.cs index 896eee6..7e03e0e 100644 --- a/benchmarks/PostgreSignalR.Benchmarks.Abstractions/Message.cs +++ b/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 ); diff --git a/benchmarks/PostgreSignalR.Benchmarks.Abstractions/PostgreSignalR.Benchmarks.Abstractions.csproj b/benchmarks/PostgreSignalR.Benchmarks.Abstractions/PostgreSignalR.Benchmarks.Abstractions.csproj index b760144..30d6988 100644 --- a/benchmarks/PostgreSignalR.Benchmarks.Abstractions/PostgreSignalR.Benchmarks.Abstractions.csproj +++ b/benchmarks/PostgreSignalR.Benchmarks.Abstractions/PostgreSignalR.Benchmarks.Abstractions.csproj @@ -6,4 +6,9 @@ enable + + + + + diff --git a/benchmarks/PostgreSignalR.Benchmarks.Collector/Dockerfile b/benchmarks/PostgreSignalR.Benchmarks.Collector/Dockerfile new file mode 100644 index 0000000..2be488e --- /dev/null +++ b/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"] diff --git a/benchmarks/PostgreSignalR.Benchmarks.Collector/PostgreSignalR.Benchmarks.Collector.csproj b/benchmarks/PostgreSignalR.Benchmarks.Collector/PostgreSignalR.Benchmarks.Collector.csproj new file mode 100644 index 0000000..a3a34b6 --- /dev/null +++ b/benchmarks/PostgreSignalR.Benchmarks.Collector/PostgreSignalR.Benchmarks.Collector.csproj @@ -0,0 +1,9 @@ + + + + net10.0 + enable + enable + + + diff --git a/benchmarks/PostgreSignalR.Benchmarks.Collector/Program.cs b/benchmarks/PostgreSignalR.Benchmarks.Collector/Program.cs new file mode 100644 index 0000000..d2ef8fb --- /dev/null +++ b/benchmarks/PostgreSignalR.Benchmarks.Collector/Program.cs @@ -0,0 +1,39 @@ +using System.Collections.Concurrent; + +var results = new ConcurrentDictionary(); + +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(); diff --git a/benchmarks/PostgreSignalR.Benchmarks.Server/Program.cs b/benchmarks/PostgreSignalR.Benchmarks.Server/Program.cs index 2527872..3833ddd 100644 --- a/benchmarks/PostgreSignalR.Benchmarks.Server/Program.cs +++ b/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("/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 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(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( diff --git a/benchmarks/PostgreSignalR.Benchmarks.Server/PublishRequest.cs b/benchmarks/PostgreSignalR.Benchmarks.Server/PublishRequest.cs index 9b5d0ba..4ca48da 100644 --- a/benchmarks/PostgreSignalR.Benchmarks.Server/PublishRequest.cs +++ b/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 ); diff --git a/benchmarks/PostgreSignalR.Benchmarks.SharedLoad/Dockerfile b/benchmarks/PostgreSignalR.Benchmarks.SharedLoad/Dockerfile new file mode 100644 index 0000000..2273acd --- /dev/null +++ b/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"] diff --git a/benchmarks/PostgreSignalR.Benchmarks.SharedLoad/PostgreSignalR.Benchmarks.SharedLoad.csproj b/benchmarks/PostgreSignalR.Benchmarks.SharedLoad/PostgreSignalR.Benchmarks.SharedLoad.csproj new file mode 100644 index 0000000..1623452 --- /dev/null +++ b/benchmarks/PostgreSignalR.Benchmarks.SharedLoad/PostgreSignalR.Benchmarks.SharedLoad.csproj @@ -0,0 +1,19 @@ + + + + Exe + net10.0 + enable + enable + + + + + + + + + + + + diff --git a/benchmarks/PostgreSignalR.Benchmarks.SharedLoad/Program.cs b/benchmarks/PostgreSignalR.Benchmarks.SharedLoad/Program.cs new file mode 100644 index 0000000..fdb2d22 --- /dev/null +++ b/benchmarks/PostgreSignalR.Benchmarks.SharedLoad/Program.cs @@ -0,0 +1,156 @@ +using Npgsql; +using StackExchange.Redis; +using PostgreSignalR.Benchmarks.Abstractions; + +var backplane = (Environment.GetEnvironmentVariable("BACKPLANE") ?? "none").ToLowerInvariant(); +var enabled = string.Equals(Environment.GetEnvironmentVariable("SIMULATE_SHARED_LOAD"), "true", StringComparison.OrdinalIgnoreCase); + +if (!enabled) +{ + Console.WriteLine("SIMULATE_SHARED_LOAD is not enabled; shared load generator idling."); + await Task.Delay(Timeout.Infinite); + + return; +} + +var concurrency = int.Parse(Environment.GetEnvironmentVariable("SHARED_LOAD_CONCURRENCY") ?? "16"); +var opsPerSec = int.Parse(Environment.GetEnvironmentVariable("SHARED_LOAD_OPS_PER_SEC") ?? "200"); +var perWorkerInterval = TimeSpan.FromMilliseconds(1000.0 * concurrency / opsPerSec); + +Console.WriteLine($"Shared load starting: backplane={backplane}, concurrency={concurrency}, opsPerSec={opsPerSec}"); + +long completed = 0; +long failed = 0; + +_ = Task.Run(async () => +{ + while (true) + { + await Task.Delay(TimeSpan.FromSeconds(10)); + Console.WriteLine($"Shared load: {Interlocked.Read(ref completed)} ops completed, {Interlocked.Read(ref failed)} failed"); + } +}); + +if (backplane is "postgres") +{ + var connectionString = ConnectionStringHelper.NormalizePostgres(Environment.GetEnvironmentVariable("ConnectionStrings__Postgres") ?? throw new Exception("Postgres connection string required.")); + + await using var dataSource = NpgsqlDataSource.Create(connectionString); + await using (var setup = dataSource.CreateCommand("CREATE TABLE IF NOT EXISTS shared_load_rows (id BIGSERIAL PRIMARY KEY, payload TEXT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now())")) + { + await setup.ExecuteNonQueryAsync(); + } + + var workers = Enumerable.Range(0, concurrency).Select(_ => RunPostgresWorkerAsync(dataSource)); + + await Task.WhenAll(workers); +} +else if (backplane is "redis") +{ + var connectionString = ConnectionStringHelper.NormalizeRedis(Environment.GetEnvironmentVariable("ConnectionStrings__Redis") ?? throw new Exception("Redis connection string required.")); + var redis = await ConnectionMultiplexer.ConnectAsync(connectionString); + var db = redis.GetDatabase(); + var workers = Enumerable.Range(0, concurrency).Select(_ => RunRedisWorkerAsync(db)); + + await Task.WhenAll(workers); +} +else +{ + throw new Exception($"Unsupported BACKPLANE '{backplane}' for shared load. Expected 'postgres' or 'redis'."); +} + +async Task RunPostgresWorkerAsync(NpgsqlDataSource dataSource) +{ + var random = new Random(); + + while (true) + { + try + { + var roll = random.NextDouble(); + + if (roll < 0.55) + { + await using var cmd = dataSource.CreateCommand("INSERT INTO shared_load_rows (payload) VALUES (@payload)"); + cmd.Parameters.AddWithValue("payload", RandomPayload(random, 200)); + await cmd.ExecuteNonQueryAsync(); + } + else if (roll < 0.85) + { + await using var cmd = dataSource.CreateCommand("SELECT id, payload FROM shared_load_rows ORDER BY id DESC LIMIT 20"); + await using var reader = await cmd.ExecuteReaderAsync(); + while (await reader.ReadAsync()) { } + } + else if (roll < 0.97) + { + await using var cmd = dataSource.CreateCommand("UPDATE shared_load_rows SET payload = @payload WHERE id = (SELECT id FROM shared_load_rows ORDER BY random() LIMIT 1)"); + cmd.Parameters.AddWithValue("payload", RandomPayload(random, 200)); + await cmd.ExecuteNonQueryAsync(); + } + else + { + await using var cmd = dataSource.CreateCommand("DELETE FROM shared_load_rows WHERE id IN (SELECT id FROM shared_load_rows ORDER BY id ASC LIMIT 50)"); + await cmd.ExecuteNonQueryAsync(); + } + + Interlocked.Increment(ref completed); + } + catch + { + Interlocked.Increment(ref failed); + } + + await Task.Delay(perWorkerInterval); + } +} + +async Task RunRedisWorkerAsync(IDatabase database) +{ + var random = new Random(); + + while (true) + { + try + { + var key = $"shared_load:{random.Next(0, 5000)}"; + var roll = random.NextDouble(); + + if (roll < 0.45) + { + await database.StringSetAsync(key, RandomPayload(random, 200), TimeSpan.FromMinutes(5)); + } + else if (roll < 0.90) + { + await database.StringGetAsync(key); + } + else if (roll < 0.98) + { + await database.StringIncrementAsync("shared_load:counter"); + } + else + { + await database.KeyDeleteAsync(key); + } + + Interlocked.Increment(ref completed); + } + catch + { + Interlocked.Increment(ref failed); + } + + await Task.Delay(perWorkerInterval); + } +} + +static string RandomPayload(Random random, int bytes) +{ + const string chars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"; + return string.Create(bytes, random, (span, rnd) => + { + for (int i = 0; i < span.Length; i++) + { + span[i] = chars[rnd.Next(chars.Length)]; + } + }); +} diff --git a/benchmarks/PostgreSignalR.Benchmarks/BufferedTextWriter.cs b/benchmarks/PostgreSignalR.Benchmarks/BufferedTextWriter.cs new file mode 100644 index 0000000..898b300 --- /dev/null +++ b/benchmarks/PostgreSignalR.Benchmarks/BufferedTextWriter.cs @@ -0,0 +1,26 @@ +using System.Text; + +namespace PostgreSignalR.Benchmarks; + +sealed class BufferedTextWriter(TextWriter inner, StringBuilder buffer) : TextWriter +{ + public override Encoding Encoding => inner.Encoding; + + public override void Write(char value) + { + inner.Write(value); + buffer.Append(value); + } + + public override void Write(string? value) + { + inner.Write(value); + buffer.Append(value); + } + + public override void WriteLine(string? value) + { + inner.WriteLine(value); + buffer.Append(value).Append(Environment.NewLine); + } +} diff --git a/benchmarks/PostgreSignalR.Benchmarks/PostgreSignalR.Benchmarks.csproj b/benchmarks/PostgreSignalR.Benchmarks/PostgreSignalR.Benchmarks.csproj index 11a78a3..d8a2934 100644 --- a/benchmarks/PostgreSignalR.Benchmarks/PostgreSignalR.Benchmarks.csproj +++ b/benchmarks/PostgreSignalR.Benchmarks/PostgreSignalR.Benchmarks.csproj @@ -10,6 +10,7 @@ + diff --git a/benchmarks/PostgreSignalR.Benchmarks/Program.cs b/benchmarks/PostgreSignalR.Benchmarks/Program.cs index 71444b1..ac3f6c8 100644 --- a/benchmarks/PostgreSignalR.Benchmarks/Program.cs +++ b/benchmarks/PostgreSignalR.Benchmarks/Program.cs @@ -1,18 +1,57 @@ using System.Collections.Concurrent; using System.Diagnostics; using System.Net.Http.Json; +using System.Text; using PostgreSignalR.Benchmarks.Abstractions; using PostgreSignalR.Benchmarks; using HdrHistogram; using Microsoft.AspNetCore.SignalR.Client; +using Npgsql; + +var consoleBuffer = new StringBuilder(); +Console.SetOut(new BufferedTextWriter(Console.Out, consoleBuffer)); +Console.SetError(new BufferedTextWriter(Console.Error, consoleBuffer)); + +var resultsCollectorUrl = Environment.GetEnvironmentVariable("RESULTS_COLLECTOR_URL"); + +AppDomain.CurrentDomain.UnhandledException += (_, e) => +{ + consoleBuffer.AppendLine((e.ExceptionObject as Exception)?.ToString() ?? e.ExceptionObject?.ToString()); + PostResultsToCollectorAsync(resultsCollectorUrl, consoleBuffer.ToString()).GetAwaiter().GetResult(); +}; static string Env(string key, string fallback) => Environment.GetEnvironmentVariable(key) ?? fallback; -var serverA = Env("SERVER_A", "http://servera:8080"); -var serverB = Env("SERVER_B", "http://serverb:8080"); +var serverUrlsEnv = Environment.GetEnvironmentVariable("SERVER_URLS"); +List serverUrls; + +if (!string.IsNullOrWhiteSpace(serverUrlsEnv)) +{ + serverUrls = serverUrlsEnv.Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries).ToList(); + + if (serverUrls.Count < 2) + { + throw new Exception($"SERVER_URLS must list at least 2 servers (one publisher, one subscriber), got {serverUrls.Count}."); + } +} +else +{ + var numServers = int.Parse(Env("NUM_SERVERS", "2")); + + if (numServers is < 2 or > 10) + { + throw new Exception($"NUM_SERVERS must be between 2 and 10 (the number of server slots defined in docker-compose.yml), got {numServers}."); + } + + serverUrls = Enumerable.Range(1, numServers).Select(i => $"http://server{i}:8080").ToList(); +} + +var publisherUrl = serverUrls[0]; +var subscriberUrls = serverUrls.Skip(1).ToList(); -var clients = int.Parse(Env("CLIENTS", "500")); +var clientsPerServer = int.Parse(Env("CLIENTS_PER_SERVER", "500")); +var clients = clientsPerServer * subscriberUrls.Count; var publishCount = int.Parse(Env("PUBLISH_COUNT", "20000")); var concurrency = int.Parse(Env("CONCURRENCY", "128")); var payloadBytes = int.Parse(Env("PAYLOAD_BYTES", "128")); @@ -30,36 +69,87 @@ static string Env(string key, string fallback) => var sweepTrialSeconds = int.Parse(Env("SWEEP_TRIAL_SECONDS", "15")); var batchSize = int.Parse(Env("BATCH_SIZE", "25")); +var repeatsPerRate = int.Parse(Env("REPEATS_PER_RATE", "1")); + +var healthCheckTimeoutSeconds = int.Parse(Env("HEALTH_CHECK_TIMEOUT_SECONDS", "60")); + +var drainQuietSeconds = double.Parse(Env("DRAIN_QUIET_SECONDS", "1")); +var drainMaxWaitSeconds = double.Parse(Env("DRAIN_MAX_WAIT_SECONDS", "60")); + +var backplane = Env("BACKPLANE", "redis").ToLowerInvariant(); +var payloadStrategy = Env("PAYLOAD_STRATEGY", "event").ToLowerInvariant(); Console.WriteLine(); Console.WriteLine("Benchmark Starting..."); -Console.WriteLine($"ServerA: {serverA}"); -Console.WriteLine($"ServerB: {serverB}"); -Console.WriteLine($"Clients: {clients}"); +Console.WriteLine($"Servers ({serverUrls.Count}): {string.Join(", ", serverUrls)}"); +Console.WriteLine($"Publisher: {publisherUrl}"); +Console.WriteLine($"Subscribers: {string.Join(", ", subscriberUrls)}"); +Console.WriteLine($"Clients: {clients} ({clientsPerServer} per subscriber)"); Console.WriteLine($"PublishCount: {publishCount}, Concurrency: {concurrency}, PayloadBytes: {payloadBytes}"); Console.WriteLine($"WarmupSeconds: {warmupSeconds}, MeasureSeconds: {measureSeconds}"); +Console.WriteLine($"DrainQuietSeconds: {drainQuietSeconds}, DrainMaxWaitSeconds: {drainMaxWaitSeconds}"); +Console.WriteLine($"RepeatsPerRate: {repeatsPerRate}"); +Console.WriteLine($"HealthCheckTimeoutSeconds: {healthCheckTimeoutSeconds}"); using var http = new HttpClient { Timeout = TimeSpan.FromMinutes(5) }; -await WaitHealthy(http, serverA); -await WaitHealthy(http, serverB); +await Task.WhenAll(serverUrls.Select(url => WaitHealthy(http, url, healthCheckTimeoutSeconds))); + +var (clockOffsetMs, clockOffsetBestRoundTripMs) = await EstimateClockOffsetMsAsync(http, publisherUrl); +Console.WriteLine($"Estimated clock offset vs publisher ({publisherUrl}): {clockOffsetMs}ms ({(clockOffsetMs >= 0 ? "publisher ahead" : "publisher behind")}, best round-trip {clockOffsetBestRoundTripMs}ms)"); + +if ((backplane, payloadStrategy) is ("postgres", "table")) +{ + var postgresConnectionString = ConnectionStringHelper.NormalizePostgres(Environment.GetEnvironmentVariable("ConnectionStrings__Postgres") ?? ""); + + if (string.IsNullOrWhiteSpace(postgresConnectionString)) + { + Console.WriteLine("Unable to clear backplane_payloads table before running benchmark with table payload strategy; driver was not given a connection string."); + } + else + { + Console.WriteLine("Clearing backplane_payloads table before running benchmark with table payload strategy..."); -var hubUrl = $"{serverB.TrimEnd('/')}/hub"; + await using var connection = new NpgsqlConnection(postgresConnectionString); + await connection.OpenAsync(); + + await using var command = new NpgsqlCommand("TRUNCATE TABLE \"backplane_payloads\";", connection); + await command.ExecuteNonQueryAsync(); + } +} var connections = new List(clients); var seen = new ConcurrentDictionary(Environment.ProcessorCount, publishCount); var fanoutCopies = new Counter(); -var histogram = new LongHistogram(60000000, 3); +var negativeLatency = new Counter(); +var generation = new Counter(); +var staleGeneration = new Counter(); +var lastStaleTicks = new Counter(); + +var histogram = HistogramFactory.With64BitBucketSize() + .WithValuesUpTo(60000000) + .WithPrecisionOf(3) + .WithThreadSafeWrites() + .WithThreadSafeReads() + .Create(); var measuring = false; for (int i = 0; i < clients; i++) { + var hubUrl = $"{subscriberUrls[i % subscriberUrls.Count].TrimEnd('/')}/hub"; var connection = new HubConnectionBuilder().WithUrl(hubUrl).WithAutomaticReconnect().Build(); connection.On("bench", message => { + if (message.Generation != Interlocked.Read(ref generation.Value)) + { + Interlocked.Increment(ref staleGeneration.Value); + Interlocked.Exchange(ref lastStaleTicks.Value, Stopwatch.GetTimestamp()); + return; + } + if (!measuring) { return; @@ -71,16 +161,20 @@ static string Env(string key, string fallback) => return; } - lock (histogram) + var rawLatencyUs = (DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() - message.SentUnixTimeMs + clockOffsetMs) * 1000; + + if (rawLatencyUs < 0) { - histogram.RecordValue(Math.Min(Math.Max((DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() - message.SentUnixTimeMs) * 1000, 0), 60000000)); + Interlocked.Increment(ref negativeLatency.Value); } + + histogram.RecordValue(Math.Min(Math.Max(rawLatencyUs, 0), 60000000)); }); connections.Add(connection); } -Console.WriteLine($"Connecting {clients} clients to {hubUrl}..."); +Console.WriteLine($"Connecting {clients} clients ({clientsPerServer} each) across {subscriberUrls.Count} subscriber(s)..."); await Task.WhenAll(connections.Select(c => c.StartAsync())); Console.WriteLine("Clients connected."); @@ -88,34 +182,95 @@ static string Env(string key, string fallback) => Console.WriteLine(); Console.WriteLine($"Warmup: {warmupSeconds}s"); -await Publish(http, serverA, publishCount: Math.Min(2000, publishCount / 10), concurrency: Math.Max(8, concurrency / 4), payloadBytes); -await Task.Delay(TimeSpan.FromSeconds(warmupSeconds)); + +var warmupResult = await RunTrialAsync( + http, + publisherUrl, + targetRate, + warmupSeconds, + 1, + concurrency, + payloadBytes, + batchSize, + seen, + fanoutCopies, + negativeLatency, + generation, + staleGeneration, + lastStaleTicks, + drainQuietSeconds, + drainMaxWaitSeconds, + histogram, + v => measuring = v +); + +async Task FinishAsync(int exitCode) +{ + Console.WriteLine(); + Console.WriteLine("Disconnecting clients..."); + await Task.WhenAll(connections.Select(c => c.DisposeAsync().AsTask())); + Console.WriteLine("Done."); + await PostResultsToCollectorAsync(resultsCollectorUrl, consoleBuffer.ToString()); + return exitCode; +} + +if (warmupResult.Sent > 0 && warmupResult.UniqueReceived == 0) +{ + Console.WriteLine(); + Console.WriteLine($"Stopping; warmup sent {warmupResult.Sent} messages but received 0."); + return await FinishAsync(1); +} if (mode.Equals("sweep", StringComparison.OrdinalIgnoreCase)) { async Task sweep(int rate) { var result = await RunTrialAsync( - http, serverA, - rate, sweepTrialSeconds, - concurrency, payloadBytes, + http, + publisherUrl, + rate, + sweepTrialSeconds, + repeatsPerRate, + concurrency, + payloadBytes, batchSize, - seen, fanoutCopies, + seen, + fanoutCopies, + negativeLatency, + generation, + staleGeneration, + lastStaleTicks, + drainQuietSeconds, + drainMaxWaitSeconds, histogram, v => measuring = v ); Console.WriteLine( $"| {$"{result.TargetRateMsgsPerSec,12}"} |" + + $" {$"{result.AchievedRateMsgsPerSec,16:F0}"} |" + $" {result.P50Us,8} |" + $" {result.P95Us,8} |" + $" {result.P99Us,8} |" + $" {result.MaxUs,8} |" + $" {$"{result.Missing,7}"} |" + $" {$"{result.FanoutCopies,13}"} |" + - $" {$"{result.Sent,13}"} |" + $" {$"{result.Sent,13}"} |" + + $" {$"{result.HistogramCount,10}"} |" + + $" {$"{result.NegativeLatencyCount,11}"} |" ); + if (result.AchievedRateMsgsPerSec < rate * 0.95) + { + Console.WriteLine($" Warning: achieved {result.AchievedRateMsgsPerSec:F0} msg/s, below the {rate} msg/s target - driver/server could not keep pace, latency at this row reflects a lower effective rate"); + } + + if (result.Sent > 0 && result.HistogramCount == 0) + { + Console.WriteLine($"Stopped: received 0 of {result.Sent} messages sent this trial."); + return false; + } + if (result.P99Us > sloP99Ms * 1000L) { Console.WriteLine($"Stopped: p99 {result.P99Us/1000.0}ms exceeded SLO {sloP99Ms}ms"); @@ -133,8 +288,8 @@ async Task sweep(int rate) Console.WriteLine(); Console.WriteLine("Sweep up:"); - Console.WriteLine("| Rate (msg/s) | p50 (Us) | p95 (Us) | p99 (Us) | Max (Us) | Missing | Fanout Copies | Messages Sent |"); - Console.WriteLine("|--------------|----------|----------|----------|----------|---------|---------------|---------------|"); + Console.WriteLine("| Rate (msg/s) | Achieved (msg/s) | p50 (Us) | p95 (Us) | p99 (Us) | Max (Us) | Missing | Fanout Copies | Messages Sent | Hist Count | Neg Latency |"); + Console.WriteLine("|--------------|------------------|----------|----------|----------|----------|---------|---------------|---------------|------------|-------------|"); for (int rate = sweepStartRate; rate <= sweepMaxRate; rate += sweepStepRate) { @@ -146,8 +301,8 @@ async Task sweep(int rate) Console.WriteLine(); Console.WriteLine("Sweep down:"); - Console.WriteLine("| Rate (msg/s) | p50 (Us) | p95 (Us) | p99 (Us) | Max (Us) | Missing | Fanout Copies | Messages Sent |"); - Console.WriteLine("|--------------|----------|----------|----------|----------|---------|---------------|---------------|"); + Console.WriteLine("| Rate (msg/s) | Achieved (msg/s) | p50 (Us) | p95 (Us) | p99 (Us) | Max (Us) | Missing | Fanout Copies | Messages Sent | Hist Count | Neg Latency |"); + Console.WriteLine("|--------------|------------------|----------|----------|----------|----------|---------|---------------|---------------|------------|-------------|"); for (int rate = sweepMaxRate; rate >= sweepStartRate; rate -= sweepStepRate) { @@ -163,11 +318,22 @@ async Task sweep(int rate) Console.WriteLine($"Single run: targetRate={targetRate} msg/s for {measureSeconds}s"); var result = await RunTrialAsync( - http, serverA, - targetRate, measureSeconds, - concurrency, payloadBytes, + http, + publisherUrl, + targetRate, + measureSeconds, + repeatsPerRate, + concurrency, + payloadBytes, batchSize, - seen, fanoutCopies, + seen, + fanoutCopies, + negativeLatency, + generation, + staleGeneration, + lastStaleTicks, + drainQuietSeconds, + drainMaxWaitSeconds, histogram, v => measuring = v ); @@ -175,33 +341,44 @@ async Task sweep(int rate) Console.WriteLine(); Console.WriteLine("Benchmark results:"); Console.WriteLine($"Target rate: {result.TargetRateMsgsPerSec} msg/s for {measureSeconds}s"); - Console.WriteLine($"Sent: {result.Sent} in {result.SendElapsedSec:F2}s"); + Console.WriteLine($"Sent: {result.Sent} in {result.SendElapsedSec:F2}s ({result.AchievedRateMsgsPerSec:F0} msg/s achieved)"); + + if (result.AchievedRateMsgsPerSec < result.TargetRateMsgsPerSec * 0.95) + { + Console.WriteLine($"Warning: achieved rate fell short of the {result.TargetRateMsgsPerSec} msg/s target - driver/server could not keep pace, latency below reflects a lower effective rate"); + } + Console.WriteLine($"Unique received: {result.UniqueReceived}"); Console.WriteLine($"Missing: {result.Missing}"); Console.WriteLine($"Fanout copies (expected): {result.FanoutCopies}"); Console.WriteLine($"Latency p50: {result.P50Us}us, p95: {result.P95Us}us, p99: {result.P99Us}us, max: {result.MaxUs}us"); + Console.WriteLine($"Histogram count: {result.HistogramCount} (compare to Unique received above - should match)"); + Console.WriteLine($"Negative computed latency (clamped to 0): {result.NegativeLatencyCount}"); + + if (result.Sent > 0 && result.HistogramCount == 0) + { + Console.WriteLine("Warning: received 0 messages (initialization error?)."); + } } -Console.WriteLine(); -Console.WriteLine("Disconnecting clients..."); -await Task.WhenAll(connections.Select(c => c.DisposeAsync().AsTask())); -Console.WriteLine("Done."); +return await FinishAsync(0); -static async Task Publish(HttpClient http, string serverA, int publishCount, int concurrency, int payloadBytes) +static async Task Publish(HttpClient http, string publisherUrl, int publishCount, int concurrency, int payloadBytes, long generation) { - var response = await http.PostAsJsonAsync($"{serverA.TrimEnd('/')}/publish", new + var response = await http.PostAsJsonAsync($"{publisherUrl.TrimEnd('/')}/publish", new { PublishCount = publishCount, Concurrency = concurrency, - PayloadBytes = payloadBytes + PayloadBytes = payloadBytes, + Generation = generation }); response.EnsureSuccessStatusCode(); } -static async Task WaitHealthy(HttpClient http, string baseUrl) +static async Task WaitHealthy(HttpClient http, string baseUrl, int timeoutSeconds) { - for (int i = 0; i < 60; i++) + for (int i = 0; i < timeoutSeconds; i++) { try { @@ -216,93 +393,221 @@ static async Task WaitHealthy(HttpClient http, string baseUrl) await Task.Delay(1000); } - throw new Exception($"Health check failed for {baseUrl}"); + throw new Exception($"Health check failed for {baseUrl} after {timeoutSeconds}s"); } -static (long p50, long p95, long p99, long max) GetPercentiles(LongHistogram hist) +static async Task PostResultsToCollectorAsync(string? collectorUrl, string output) +{ + if (string.IsNullOrWhiteSpace(collectorUrl)) + { + return; + } + + try + { + using var client = new HttpClient(); + await client.PostAsync(collectorUrl, new StringContent(output)); + } + catch (Exception ex) + { + Console.Error.WriteLine($"Failed to post results to collector: {ex}"); + } +} + +static async Task<(long OffsetMs, long BestRoundTripMs)> EstimateClockOffsetMsAsync(HttpClient http, string serverUrl) +{ + var bestOffsetMs = 0D; + var bestRoundTripMs = double.MaxValue; + + for (int i = 0; i < 15; i++) + { + var t0 = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); + TimeResult? response; + + try + { + response = await http.GetFromJsonAsync($"{serverUrl.TrimEnd('/')}/time"); + } + catch + { + continue; + } + + var t2 = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); + + if (response is null) + { + continue; + } + + var roundTripMs = t2 - t0; + + if (roundTripMs < bestRoundTripMs) + { + bestRoundTripMs = roundTripMs; + bestOffsetMs = response.UnixTimeMs - (t0 + t2) / 2.0; + } + } + + var boundedRoundTripMs = bestRoundTripMs == double.MaxValue ? 0 : (long)Math.Round(bestRoundTripMs); + return ((long)Math.Round(bestOffsetMs), boundedRoundTripMs); +} + +static async Task DrainAsync(Counter generation, Counter staleGeneration, Counter lastStaleTicks, double quietSeconds, double maxWaitSeconds) { - lock (hist) + var staleAtStart = Interlocked.Read(ref staleGeneration.Value); + + Interlocked.Increment(ref generation.Value); + Interlocked.Exchange(ref lastStaleTicks.Value, Stopwatch.GetTimestamp()); + + var tickFrequency = (double)Stopwatch.Frequency; + var start = Stopwatch.GetTimestamp(); + + while (true) + { + var now = Stopwatch.GetTimestamp(); + var sinceLastStale = (now - Interlocked.Read(ref lastStaleTicks.Value)) / tickFrequency; + + if (sinceLastStale >= quietSeconds) + { + break; + } + + if ((now - start) / tickFrequency >= maxWaitSeconds) + { + Console.WriteLine($" Warning: drain wait hit the {maxWaitSeconds:F0}s cap with stragglers still trickling in - proceeding anyway"); + break; + } + + await Task.Delay(250); + } + + var strayCount = Interlocked.Read(ref staleGeneration.Value) - staleAtStart; + if (strayCount > 0) { - var p50 = hist.GetValueAtPercentile(50); - var p95 = hist.GetValueAtPercentile(95); - var p99 = hist.GetValueAtPercentile(99); - var max = hist.GetMaxValue(); + var waitedSeconds = (Stopwatch.GetTimestamp() - start) / tickFrequency; + Console.WriteLine($" Drained {strayCount} stale-generation stragglers over {waitedSeconds:F1}s before starting the next window"); + } +} - return (p50, p95, p99, max); +static (long p50, long p95, long p99, long max) GetPercentiles(LongHistogram hist) +{ + if (hist.TotalCount == 0) + { + return (0, 0, 0, 0); } + + var p50 = hist.GetValueAtPercentile(50); + var p95 = hist.GetValueAtPercentile(95); + var p99 = hist.GetValueAtPercentile(99); + var max = hist.GetMaxValue(); + + return (p50, p95, p99, max); } static async Task RunTrialAsync( HttpClient http, - string serverA, + string publisherUrl, int targetRateMsgsPerSec, int trialSeconds, + int repeats, int concurrency, int payloadBytes, int batchSize, ConcurrentDictionary seen, Counter fanoutCopies, - LongHistogram hist, - Action setMeasuring) + Counter negativeLatency, + Counter generation, + Counter staleGeneration, + Counter lastStaleTicks, + double drainQuietSeconds, + double drainMaxWaitSeconds, + Recorder hist, + Action setMeasuring +) { - seen.Clear(); - Interlocked.Exchange(ref fanoutCopies.Value, 0); - lock (hist) hist.Reset(); + var pooledHist = new LongHistogram(60000000, 3); - setMeasuring(true); + long totalSent = 0; + long totalUnique = 0; + long totalFanoutCopies = 0; + long totalNegativeLatency = 0; + double totalElapsedSec = 0; - var delay = TimeSpan.FromMilliseconds((int)Math.Max(0, Math.Round(1000.0 * batchSize / targetRateMsgsPerSec))); - - var totalToSend = Math.Max(1, targetRateMsgsPerSec * trialSeconds); + for (int repeat = 0; repeat < repeats; repeat++) + { + seen.Clear(); + Interlocked.Exchange(ref fanoutCopies.Value, 0); + Interlocked.Exchange(ref negativeLatency.Value, 0); - var batchPeriod = TimeSpan.FromSeconds(batchSize / (double)targetRateMsgsPerSec); - var next = Stopwatch.GetTimestamp(); - var tickFreq = (double)Stopwatch.Frequency; + var thisGeneration = Interlocked.Increment(ref generation.Value); - var sw = Stopwatch.StartNew(); + setMeasuring(true); - for (int sent = 0; sent < totalToSend; sent += batchSize) - { - var thisBatch = Math.Min(batchSize, totalToSend - sent); + var totalToSend = Math.Max(1, targetRateMsgsPerSec * trialSeconds); + var batchPeriod = TimeSpan.FromSeconds(batchSize / (double)targetRateMsgsPerSec); + var next = Stopwatch.GetTimestamp(); + var tickFreq = (double)Stopwatch.Frequency; - await Publish(http, serverA, thisBatch, concurrency, payloadBytes); + var sw = Stopwatch.StartNew(); - next += (long)(batchPeriod.TotalSeconds * tickFreq); - var now = Stopwatch.GetTimestamp(); - var remainingTicks = next - now; + var inFlight = new List(); - if (remainingTicks > 0) + for (int sent = 0; sent < totalToSend; sent += batchSize) { - var remainingMs = (int)(remainingTicks * 1000.0 / tickFreq); + var thisBatch = Math.Min(batchSize, totalToSend - sent); - if (remainingMs > 0) - { - await Task.Delay(remainingMs); - } - else + inFlight.Add(Publish(http, publisherUrl, thisBatch, concurrency, payloadBytes, thisGeneration)); + + next += (long)(batchPeriod.TotalSeconds * tickFreq); + var now = Stopwatch.GetTimestamp(); + var remainingTicks = next - now; + + if (remainingTicks > 0) { - await Task.Yield(); + var remainingMs = (int)(remainingTicks * 1000.0 / tickFreq); + + if (remainingMs > 0) + { + await Task.Delay(remainingMs); + } + else + { + await Task.Yield(); + } } } - } - sw.Stop(); + await Task.WhenAll(inFlight); + sw.Stop(); + + await Task.Delay(1000); + setMeasuring(false); + + totalSent += totalToSend; + totalUnique += seen.Count; + totalFanoutCopies += Interlocked.Read(ref fanoutCopies.Value); + totalNegativeLatency += Interlocked.Read(ref negativeLatency.Value); + totalElapsedSec += sw.Elapsed.TotalSeconds; - await Task.Delay(1000); - setMeasuring(false); + pooledHist.Add(hist.GetIntervalHistogram()); + + await DrainAsync(generation, staleGeneration, lastStaleTicks, drainQuietSeconds, drainMaxWaitSeconds); + } - var unique = seen.Count; - var missing = Math.Max(0, totalToSend - unique); + var missing = Math.Max(0, totalSent - totalUnique); - (long p50, long p95, long p99, long max) = GetPercentiles(hist); + (long p50, long p95, long p99, long max) = GetPercentiles(pooledHist); return new TrialResult( targetRateMsgsPerSec, - totalToSend, - sw.Elapsed.TotalSeconds, - unique, - missing, - Interlocked.Read(ref fanoutCopies.Value), - p50, p95, p99, max + (int)totalSent, + totalElapsedSec, + (int)totalUnique, + (int)missing, + totalFanoutCopies, + p50, p95, p99, max, + pooledHist.TotalCount, + totalNegativeLatency ); } diff --git a/benchmarks/PostgreSignalR.Benchmarks/TimeResult.cs b/benchmarks/PostgreSignalR.Benchmarks/TimeResult.cs new file mode 100644 index 0000000..5a22127 --- /dev/null +++ b/benchmarks/PostgreSignalR.Benchmarks/TimeResult.cs @@ -0,0 +1,5 @@ +namespace PostgreSignalR.Benchmarks; + +sealed record TimeResult( + long UnixTimeMs +); diff --git a/benchmarks/PostgreSignalR.Benchmarks/TrialResult.cs b/benchmarks/PostgreSignalR.Benchmarks/TrialResult.cs index 040291b..bbcdddb 100644 --- a/benchmarks/PostgreSignalR.Benchmarks/TrialResult.cs +++ b/benchmarks/PostgreSignalR.Benchmarks/TrialResult.cs @@ -7,5 +7,11 @@ sealed record TrialResult( int UniqueReceived, int Missing, long FanoutCopies, - long P50Us, long P95Us, long P99Us, long MaxUs -); \ No newline at end of file + long P50Us, long P95Us, long P99Us, long MaxUs, + long HistogramCount, + long NegativeLatencyCount +) +{ + public double AchievedRateMsgsPerSec => + SendElapsedSec > 0 ? Sent / SendElapsedSec : 0; +} \ No newline at end of file diff --git a/benchmarks/RAILWAY_SETUP.md b/benchmarks/RAILWAY_SETUP.md new file mode 100644 index 0000000..5b9e310 --- /dev/null +++ b/benchmarks/RAILWAY_SETUP.md @@ -0,0 +1,102 @@ +# Running the benchmarks on Railway (one-time setup) + +`run-comparisons-railway.sh` only handles the *repeatable* part of a Railway run - switching each scenario's config, redeploying, running the driver, and collecting logs. It assumes the project/services below already exist, because creating them is a one-time task best done through Railway's dashboard, where you get live validation, rather than scripted blind CLI calls. + +**Heads up**: I wrote this without access to a live Railway account or CLI, so I could not test any of this end-to-end. The shape of it (private networking hostnames, `railway.json` restart policy key, the general CLI command set) is cross-checked against docs.railway.com, but exact CLI flag syntax for some commands wasn't fully confirmed by what I could verify - see "What's unverified" at the bottom. Expect to adjust a few commands against what `railway --help` actually shows you. + +## 1. Project and environment + +Create a Railway project from this repo, then create a dedicated environment for benchmarking (keeps this isolated from - and easy to wholesale delete separately from - any other use of the same project): + +``` +railway login +railway init # or `railway link` if the project already exists +railway environment new benchmarks +railway environment benchmarks +``` + +## 2. Databases + +Add both a Postgres and a Redis plugin to the environment - both always exist regardless of which backplane a given scenario uses, same as the local docker-compose setup: + +``` +railway add --database postgres +railway add --database redis +``` + +Note the variable names Railway exposes for each (typically `DATABASE_URL` and `REDIS_URL`, referenced from other services as `${{Postgres.DATABASE_URL}}` / `${{Redis.REDIS_URL}}` - confirm the exact plugin/variable names in the dashboard, they're what you'll reference in step 4). + +## 3. Server services (server1 .. server10) + +Create 10 services, each deployed from this repo with: + +- **Root directory**: repo root (so the Dockerfile's `COPY benchmarks/ benchmarks/` resolves correctly, same as `context: .` in docker-compose.yml) +- **Dockerfile path**: `benchmarks/PostgreSignalR.Benchmarks.Server/Dockerfile` +- **Service name**: `server1`, `server2`, ... `server10` exactly - the run script builds `SERVER_URLS` from `http://server{N}.railway.internal:8080`, which only resolves if the service names match. +- **Private networking**: enabled by default for services in the same project/environment (no extra config per Railway's docs), reachable at `.railway.internal` on whatever port the container listens on - our Dockerfile already sets `ASPNETCORE_URLS=http://0.0.0.0:8080`, so no `$PORT` handling is needed. +- **Variables** (same for all 10): + ``` + ASPNETCORE_ENVIRONMENT=Production + Logging__LogLevel__Default=Warning + ConnectionStrings__Postgres=${{Postgres.DATABASE_URL}} + ConnectionStrings__Redis=${{Redis.REDIS_URL}} + ``` + `BACKPLANE` and `PAYLOAD_STRATEGY` are set per-scenario by the run script - don't set them here. + +If you'd rather not create these one at a time by hand, `railway add --repo ` (run 10 times, renaming/reconfiguring the Dockerfile path each time) is the CLI equivalent - just confirm the actual flags against `railway add --help`, since I couldn't verify the exact non-interactive syntax for repeated same-repo services. + +## 4. shared-load service + +Same repo/root directory, with: + +- **Dockerfile path**: `benchmarks/PostgreSignalR.Benchmarks.SharedLoad/Dockerfile` +- **Variables**: + ``` + ConnectionStrings__Postgres=${{Postgres.DATABASE_URL}} + ConnectionStrings__Redis=${{Redis.REDIS_URL}} + ``` + `BACKPLANE` and `SIMULATE_SHARED_LOAD` are set per-scenario by the run script. + +## 5. driver service + +Same repo/root directory, with: + +- **Dockerfile path**: `benchmarks/PostgreSignalR.Benchmarks/Dockerfile` +- **Restart policy: `NEVER`.** The driver is a one-shot job - it runs a scenario and exits - not a long-lived server. Without this, Railway will likely treat the exit as a crash and keep restarting it. Set this via the dashboard's deploy settings, or a `railway.json` alongside the driver's Dockerfile: + ```json + { + "$schema": "https://railway.app/railway.schema.json", + "deploy": { + "restartPolicyType": "NEVER" + } + } + ``` +- **Variables**: + ``` + ConnectionStrings__Postgres=${{Postgres.DATABASE_URL}} + ``` + The driver connects directly to Postgres to `TRUNCATE` the `backplane_payloads` table before each `postgres-*-table` scenario starts (the library's own TTL cleanup is disabled for these benchmarks, and unlike local docker-compose this Postgres instance is never torn down between scenarios - see README's payload-table note). All of `SERVER_URLS`, `MODE`, `CLIENTS_PER_SERVER`, `BACKPLANE`, `PAYLOAD_STRATEGY`, etc. are set per-scenario by the run script - you don't need to set anything else here up front. + +## 6. Run it + +``` +./benchmarks/run-comparisons-railway.sh --list # sanity check the matrix +./benchmarks/run-comparisons-railway.sh redis-dedicated # dry-run: prints the commands only +./benchmarks/run-comparisons-railway.sh --apply redis-dedicated # runs it for real +./benchmarks/run-comparisons-railway.sh --apply # runs the full 10-scenario matrix +./benchmarks/run-comparisons-railway.sh --apply --teardown # deletes the whole environment when done +``` + +Unlike the local `run-comparisons.sh`, this does **not** tear down and recreate infrastructure between scenarios - redeploying 11 services per scenario is already slow enough on real cloud infrastructure without also destroying and rebuilding them each time. Services are left running between scenarios and only deleted by an explicit `--teardown` - remember to run that when you're done, since this bills for real compute the whole time it's up. + +## What's unverified + +I don't have a Railway CLI or account to test against, so treat these as the first things to check if something doesn't work: + +- `railway variables --service NAME --set KEY=value` - confirmed working, including multiple `--set` flags in one call. +- `railway redeploy --service NAME --yes` - confirmed the flags are accepted, but **setting a variable already triggers a redeploy on its own** - an explicit `redeploy` called right after routinely fails with "cannot be redeployed... currently building, deploying" because one is already in flight. The script now treats `redeploy` as a best-effort nudge (failure is logged, not fatal) rather than relying on it. +- `railway logs --service NAME` - confirmed there's no `--follow`/streaming mode (`railway logs --service [DEPLOYMENT_ID]` is the actual usage); the script now polls with repeated one-shot fetches instead. Still unconfirmed: whether a fetch is scoped to just the latest deployment or returns history across deployments - the script waits for the count of `"Done."` lines to *increase* rather than matching the first occurrence, which is correct either way. +- `railway environment delete NAME --yes` - the exact non-interactive teardown syntax. +- Whether `${{Postgres.DATABASE_URL}}`/`${{Redis.REDIS_URL}}` are the actual variable names Railway's plugins expose - check your dashboard, they may differ. + +Run each of these by hand once against your actual Railway CLI before trusting the full `--apply` run, and let me know what's different so I can fix the script against ground truth instead of docs. diff --git a/benchmarks/README.md b/benchmarks/README.md index 8649d6d..6cff84c 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -1,12 +1,13 @@ # Benchmarks -The benchmarks use three projects: +The benchmarks use four projects: * [PostgreSignalR.Benchmarks](https://github.com/IanWold/PostgreSignalR/tree/main/benchmarks/PostgreSignalR.Benchmarks) is the executable that performs the benchmarks. * [PostgreSignalR.Benchmarks.Server](https://github.com/IanWold/PostgreSignalR/tree/main/benchmarks/PostgreSignalR.Benchmarks.Server) is a server implementation the benchmarks use to test the backplanes. +* [PostgreSignalR.Benchmarks.SharedLoad](https://github.com/IanWold/PostgreSignalR/tree/main/benchmarks/PostgreSignalR.Benchmarks.SharedLoad) optionally simulates other traffic on the backplane's Postgres/Redis instance, unrelated to SignalR. * [PostgreSignalR.Benchmarks.Abstractions](https://github.com/IanWold/PostgreSignalR/tree/main/benchmarks/PostgreSignalR.Benchmarks.Abstractions) is a shared class. -The benchmarks are run through docker compose. The docker-compose yml will create containers for postgres and redis, two server containers, and one driver container which will run the benchmarks. The benchmarks can run either the Redis or Postgres backplanes. +The benchmarks are run through docker compose. The docker-compose yml will create containers for postgres and redis, 10 fixed server slots (`server1`-`server10`), the shared-load generator, and one driver container which will run the benchmarks. The benchmarks can run either the Redis or Postgres backplanes. ``` BACKPLANE=postgres MODE=sweep docker compose up --build --abort-on-container-exit --exit-code-from driver @@ -18,11 +19,83 @@ There are two modes it can run in: * `single` runs a single round of tests * `sweep` will run many rounds of tests, incrementing the number of clients. It will sweep up and down. -The other variables you acn specify: +The other variables you can specify: -* `CLIENTS`: the numbre of clients to connect. +* `NUM_SERVERS`: the number of server nodes to spread the backplane fanout across, from 2 to 10 (docker-compose.yml defines 10 fixed slots, `server1`-`server10`; raise the ceiling there if you need more). `server1` is always the sole publish target; the rest are subscribers, each getting `CLIENTS_PER_SERVER` connections. This lets you test whether fanout latency degrades as the number of subscribing nodes grows, separately from load on any one node. Default 2 (one publisher, one subscriber). Ignored if `SERVER_URLS` is set. +* `SERVER_URLS`: a comma-separated list of server base URLs to use instead of the `server1..serverN` docker-compose naming (e.g. `https://bench-server-1.example.com,https://bench-server-2.example.com`). Use this to point the driver at servers deployed somewhere other than this docker-compose setup (i.e. cloud provider). The first URL is always the publish target; the rest are subscribers, same as `NUM_SERVERS`. At least 2 URLs are required. +* `CLIENTS_PER_SERVER`: the number of clients to connect to *each* subscriber node. Total clients connected = `CLIENTS_PER_SERVER * (NUM_SERVERS - 1)`, so every subscriber always carries equal load - raising `NUM_SERVERS` raises total client count too. Default 500. * `PUBLISH_COUNT`: The number of messages to publish. Default 20000. -* `CONCURRENCY`: The maximum number of concurrent requests (from server). Default 128. -* `PAYLOAD_BYTES`: The number of bytes in the payload. Default 128. +* `CONCURRENCY`: The maximum number of concurrent `SendAsync` calls in flight on the server at any time, for the lifetime of the run. Default 128. +* `PAYLOAD_BYTES`: The number of bytes of filler content included in each message's payload. Default 128. * `WARMUP_SECONDS`: The number of seconds to warm up. Default 10. -* `MEASURE_SECONDS`: For `single` runs, the number of seconds to measure. Messages/second will be `PUBLISH_COUNT / MEASURE_SECONDS`. \ No newline at end of file +* `MEASURE_SECONDS`: For `single` runs, the number of seconds to measure. Messages/second will be `PUBLISH_COUNT / MEASURE_SECONDS`. +* `REPEATS_PER_RATE`: The number of independent trials to run at each rate (each rate in a `sweep`, or the single trial in `single` mode). Latency percentiles are computed over the pooled samples from all repeats; `Sent`/`Missing`/`Fanout Copies` are summed. Default 1. +* `HEALTH_CHECK_TIMEOUT_SECONDS`: How long the driver waits (polling once per second) for each server's `/health` endpoint before giving up and failing the run. Default 60. +* `DRAIN_QUIET_SECONDS`: After each repeat (including warmup's), every message carries the generation number of the window it was sent for - a straggler that arrives late always keeps its original generation, so it's rejected by comparing against the currently active one rather than trusting arrival timing. Before starting the next window, the driver waits until no such stragglers have arrived for this many seconds, rather than assuming a fixed delay is enough (which doesn't scale to slower/more congested runs, e.g. many clients or degraded infrastructure). Default 2. +* `DRAIN_MAX_WAIT_SECONDS`: A cap on the above, in case stragglers never fully stop arriving - the run proceeds anyway with a warning rather than hanging forever. Default 60. +* `PAYLOAD_STRATEGY`: Only applies when `BACKPLANE=postgres` + * `event` (default) sends payloads inline in the notification event. + * `table` uses PostgreSignalR's payload table strategy instead (`AddBackplaneTablePayloadStrategy` with `StorageMode=Always`). The server disables the library's own TTL-based row cleanup for this benchmark, so before each run the driver connects to `ConnectionStrings__Postgres` directly and truncates the `backplane_payloads` table itself - this matters most on Railway, where the same Postgres instance is reused across every scenario instead of being torn down like local docker-compose's is. + +The `sweep` output table's `Rate (msg/s)` column is the offered rate, i.e. what the driver was asked to send - it is not necessarily what was achieved. The `Achieved (msg/s)` column is the rate actually measured (messages sent / actual dispatch time), which falls below the target once the driver or server can't keep up. When achieved rate drops more than 5% below target, a warning is printed, since the latency figures on that row reflect the achieved rate, not the labeled one. The same applies to `single` mode's `Sent ... achieved` line. + +## Connection strings + +`ConnectionStrings__Postgres` and `ConnectionStrings__Redis` accept either the native keyword=value formats Npgsql/StackExchange.Redis expect, or a `postgres://`/`postgresql://` and `redis://`/`rediss://` URI - the format most cloud providers hand out as `DATABASE_URL`/`REDIS_URL`. URIs are converted automatically (`rediss://` and a Postgres URI's `sslmode` query param both map through correctly); values already in native format are passed through unchanged, so the local docker-compose setup is unaffected. This lets `server`, `shared-load`, and the driver's backplane connections point at a real managed database instead of the containers the compose file provisions. + +Any other query param on a `postgres://` URI is passed straight through as an Npgsql connection string keyword (e.g. `?MaxPoolSize=10&MinPoolSize=10`), so pool tuning can be done entirely via the Railway variable value without code changes - e.g. appending `?MaxPoolSize=10&MinPoolSize=10` to the `server1`/`server2` services' `ConnectionStrings__Postgres` value (`${{Postgres.DATABASE_URL}}?MaxPoolSize=10&MinPoolSize=10`) keeps a small, pre-warmed pool of persistent connections instead of growing on demand up to Npgsql's default of 100. + +## Dedicated vs. Shared Backplane + +By default the benchmarks give Postgres/Redis to the backplane exclusively - nothing else is talking to them. That's a best case, and not how these are typically deployed in production: Redis is frequently shared with other caching/session traffic, and the whole point of a Postgres backplane is usually to reuse a database you already run for your application, not stand up a dedicated instance. + +`PostgreSignalR.Benchmarks.SharedLoad` simulates that other traffic. It runs a simple, continuous CRUD-ish workload (mostly writes/reads, occasional updates and cleanup deletes for Postgres; mostly sets/gets, occasional counters and deletes for Redis) against the same Postgres database or Redis instance used as the backplane, in a separate table/keyspace so it doesn't interact with SignalR's own messages - it just simulates realistic CPU/IO/connection/lock contention. + +* `SIMULATE_SHARED_LOAD`: `true` to enable the generator, `false` (default) to leave it idle. +* `SHARED_LOAD_CONCURRENCY`: number of parallel workers generating load. Default 16. +* `SHARED_LOAD_OPS_PER_SEC`: approximate total operations/second across all workers. Default 200. + +To compare all four scenarios: + +``` +# Dedicated Redis backplane +BACKPLANE=redis MODE=sweep docker compose up --build --abort-on-container-exit --exit-code-from driver + +# Dedicated Postgres backplane +BACKPLANE=postgres MODE=sweep docker compose up --build --abort-on-container-exit --exit-code-from driver + +# Shared Redis backplane +BACKPLANE=redis SIMULATE_SHARED_LOAD=true MODE=sweep docker compose up --build --abort-on-container-exit --exit-code-from driver + +# Shared Postgres backplane +BACKPLANE=postgres SIMULATE_SHARED_LOAD=true MODE=sweep docker compose up --build --abort-on-container-exit --exit-code-from driver +``` + +## Recreating all my Benchmarks + +I generated `run-comparisons.sh` to run a set of 16 predefined scenarios that I think give a good comparison across several different use cases, grouped into four questions: + +1. `redis`/`postgres`-`dedicated`/`shared`: the core backplane comparison, with a rate sweep up to 2000 msg/s - this doubles as "what's the max sustainable publish rate" discovery. +2. `postgres-*-table`: same, but the payload-table strategy instead of the default event one. +3. `*-clients-N`: fixed low rate (10-100 msg/s, a realistic occasional-broadcast pattern rather than a firehose), with the client count per server pushed up in steps (500/2000/4000) - "what's the max sustainable client count" discovery. +4. `*-dedicated-Nservers`: fixed client count/rate, but spreading load across more subscriber nodes - fan-out width, a different axis from the above. + +Each scenario's rate sweep, client count, and other parameters are baked in per-scenario (see `run-comparisons.sh --list` for the exact values) rather than configurable via environment variables, since the whole point is a fixed, repeatable matrix - edit the `scenarios` array directly if you want different values. Scenarios in groups 1-2 and 4 take roughly 20-30 minutes each; group 3's scenarios are quicker given their smaller rate sweep. The full suite takes several hours. + +If you're just interested in running certain scenarios, you can execute `run-comparisons.sh --list` to see all of them and list scenarios out to run, like `run-comparisons.sh redis-dedicated postgres-shared`. + +Logs are saved to `results//.log` + +## Running on Railway + +Local docker-compose puts everything on one Docker host, which doesn't approximate a real deployment (no real network hops, all services sharing the same CPU/memory). `run-comparisons-railway.sh` runs the same scenario matrix against a real Railway project instead, with `SERVER_URLS` and the URI-style connection string support (see "Connection strings" above) doing the work of pointing the same driver/server/shared-load images at cloud infrastructure instead of docker-compose's containers. + +This needs a one-time setup - see [RAILWAY_SETUP.md](RAILWAY_SETUP.md), which also documents the parts of the script that are best-effort and not yet confirmed against a live Railway account. Once set up: + +``` +./benchmarks/run-comparisons-railway.sh --list # same matrix as run-comparisons.sh +./benchmarks/run-comparisons-railway.sh redis-dedicated # dry-run: prints the commands only +./benchmarks/run-comparisons-railway.sh --apply redis-dedicated # runs it for real +``` + +It defaults to dry-run (printing the `railway` commands rather than running them) since it hasn't been tested against a live account - pass `--apply` once you've sanity-checked the commands. diff --git a/benchmarks/run-comparisons-railway.sh b/benchmarks/run-comparisons-railway.sh new file mode 100755 index 0000000..604ce30 --- /dev/null +++ b/benchmarks/run-comparisons-railway.sh @@ -0,0 +1,330 @@ +#!/usr/bin/env bash +set -euo pipefail + +# ============================================================================ +# BEST-EFFORT DRAFT - commands not verified against a live Railway account. +# +# I have no Railway CLI or credentials available in the environment I wrote +# this in, so none of the `railway` invocations below have actually been run +# against real infrastructure. They're my best understanding of the current +# CLI cross-checked against docs.railway.com, but Railway's CLI syntax has +# changed across major versions and a few specifics weren't confirmed by what +# I could verify (see README section "Railway script: what's unverified"). +# +# This defaults to dry-run (it only prints the `railway` commands it would +# run). Pass --apply once you've compared the printed commands against +# `railway --help` / `railway --help` on your machine. +# ============================================================================ +# +# Runs the same scenario matrix as run-comparisons.sh, but against a Railway +# project instead of local docker-compose, so the backplanes are exercised +# over a real network between isolated hosts rather than one shared Docker +# engine. It assumes the one-time setup in benchmarks/RAILWAY_SETUP.md has +# already been done (project, environment, 11 services + Postgres/Redis +# plugins already exist) - this script only handles the repeatable part: +# switching each scenario's config, redeploying, running the driver to +# completion, and collecting its logs. +# +# Usage: +# ./benchmarks/run-comparisons-railway.sh --list # print the matrix, exit +# ./benchmarks/run-comparisons-railway.sh --apply # run every scenario for real +# ./benchmarks/run-comparisons-railway.sh --apply redis-dedicated # run one scenario for real +# ./benchmarks/run-comparisons-railway.sh redis-dedicated # dry-run: print the commands only +# +# Environment: +# RAILWAY_ENVIRONMENT Railway environment to operate in. Default "benchmarks". +# MAX_SERVERS Must match however many server1..serverN services you +# provisioned in RAILWAY_SETUP.md. Default 10. + +cd "$(dirname "${BASH_SOURCE[0]}")/.." + +: "${RAILWAY_ENVIRONMENT:=production}" +: "${MAX_SERVERS:=10}" + +: "${MODE:=sweep}" +: "${PUBLISH_COUNT:=20000}" +: "${CONCURRENCY:=128}" +: "${PAYLOAD_BYTES:=128}" +: "${WARMUP_SECONDS:=10}" +: "${TARGET_RATE:=100}" +: "${SLO_P99_MS:=250}" +: "${SWEEP_TRIAL_SECONDS:=15}" +: "${BATCH_SIZE:=25}" +: "${REPEATS_PER_RATE:=3}" +# Cloud builds/rollouts are slower than local docker-compose; give the driver's +# own internal health-check loop more room before it gives up on a server. +: "${HEALTH_CHECK_TIMEOUT_SECONDS:=180}" +: "${DRAIN_QUIET_SECONDS:=1}" +: "${DRAIN_MAX_WAIT_SECONDS:=60}" + +# name backplane shared num_servers payload_strategy clients_per_server sweep_start sweep_step sweep_max +scenarios=( + "redis-dedicated redis false 2 event 50 100 100 2000" + "postgres-dedicated postgres false 2 event 50 100 100 2000" + "postgres-dedicated-table postgres false 2 table 50 100 100 2000" + "redis-shared redis true 2 event 50 100 100 2000" + "postgres-shared postgres true 2 event 50 100 100 2000" + "postgres-shared-table postgres true 2 table 50 100 100 2000" + "redis-clients-500 redis false 2 event 500 10 10 100" + "redis-clients-2000 redis false 2 event 2000 10 10 100" + "redis-clients-4000 redis false 2 event 4000 10 10 100" + "postgres-clients-500 postgres false 2 event 500 10 10 100" + "postgres-clients-2000 postgres false 2 event 2000 10 10 100" + "postgres-clients-4000 postgres false 2 event 4000 10 10 100" + "postgres-clients-500-table postgres false 2 table 500 10 10 100" + "postgres-clients-2000-table postgres false 2 table 2000 10 10 100" + "postgres-clients-4000-table postgres false 2 table 4000 10 10 100" + "redis-dedicated-5servers redis false 5 event 50 100 100 2000" + "postgres-dedicated-5servers postgres false 5 event 50 100 100 2000" + "redis-dedicated-10servers redis false 10 event 50 100 100 2000" + "postgres-dedicated-10servers postgres false 10 event 50 100 100 2000" +) + +print_matrix() { + printf '%-30s %-9s %-7s %-11s %-9s %-19s %-12s %-11s %s\n' \ + "NAME" "BACKPLANE" "SHARED" "NUM_SERVERS" "STRATEGY" "CLIENTS_PER_SERVER" "SWEEP_START" "SWEEP_STEP" "SWEEP_MAX" + for entry in "${scenarios[@]}"; do + read -r name backplane shared num_servers strategy clients_per_server sweep_start sweep_step sweep_max <<< "$entry" + printf '%-30s %-9s %-7s %-11s %-9s %-19s %-12s %-11s %s\n' \ + "$name" "$backplane" "$shared" "$num_servers" "$strategy" "$clients_per_server" "$sweep_start" "$sweep_step" "$sweep_max" + done +} + +apply=false +teardown_only=false +requested=() + +for arg in "$@"; do + case "$arg" in + --list|-l) print_matrix; exit 0 ;; + --apply) apply=true ;; + --teardown) teardown_only=true ;; + *) requested+=("$arg") ;; + esac +done + +# Every railway invocation goes through this so --apply is the one switch +# between "print what would happen" and "actually spend money." +run() { + if $apply; then + echo "+ $*" + "$@" + else + echo "[dry-run] $*" + fi +} + +# `railway variables --set` appears to trigger a redeploy on its own - confirmed by Railway +# refusing an explicit redeploy called right after with "currently building, deploying". +# This is only a best-effort nudge in case that's not always the case; a failure here is +# expected and fine (it almost always means the variable change already triggered one), so +# it must not be allowed to abort the whole run the way `run` (and set -e) would. +try_redeploy() { + local name=$1 + + if $apply; then + echo "+ railway redeploy --service $name --yes (best-effort - a failure here just means the variable change above already triggered a redeploy)" + railway redeploy --service "$name" --yes || echo " (redeploy call failed/skipped for $name - assuming the variable change already triggered one)" + else + echo "[dry-run] railway redeploy --service $name --yes (best-effort, ignored if it fails)" + fi +} + +if $teardown_only; then + echo "Tearing down Railway environment '$RAILWAY_ENVIRONMENT'..." + run railway environment delete "$RAILWAY_ENVIRONMENT" --yes + exit 0 +fi + +selected=() + +if [ ${#requested[@]} -eq 0 ]; then + selected=("${scenarios[@]}") +else + for name in "${requested[@]}"; do + found=false + + for entry in "${scenarios[@]}"; do + if [ "${entry%% *}" == "$name" ]; then + selected+=("$entry") + found=true + break + fi + done + + if ! $found; then + echo "Unknown scenario: $name" >&2 + echo "Run with --list to see available scenarios." >&2 + exit 1 + fi + done +fi + +run railway environment "$RAILWAY_ENVIRONMENT" + +timestamp=$(date +%Y%m%d-%H%M%S) +results_dir="benchmarks/results-railway/$timestamp" +mkdir -p "$results_dir" + +echo "Environment: $RAILWAY_ENVIRONMENT" +echo "Mode: $([ "$apply" = true ] && echo APPLY || echo DRY-RUN)" +echo "Results directory: $results_dir" +echo "Scenarios: $(for e in "${selected[@]}"; do printf '%s ' "${e%% *}"; done)" +echo + +# Sets BACKPLANE/PAYLOAD_STRATEGY on all MAX_SERVERS server services and +# BACKPLANE/SIMULATE_SHARED_LOAD on shared-load, then redeploys both so the +# new config actually takes effect. All 10 slots are updated regardless of +# this scenario's num_servers, same as the docker-compose version - the ones +# beyond num_servers just sit unused. +configure_scenario() { + local backplane=$1 shared=$2 strategy=$3 + + for i in $(seq 1 "$MAX_SERVERS"); do + local name="server$i" + run railway variables --service "$name" --set "BACKPLANE=$backplane" --set "PAYLOAD_STRATEGY=$strategy" + try_redeploy "$name" + done + + run railway variables --service shared-load --set "BACKPLANE=$backplane" --set "SIMULATE_SHARED_LOAD=$shared" + try_redeploy shared-load +} + +# Builds SERVER_URLS from the first num_servers of the MAX_SERVERS private +# hostnames (server1.railway.internal, ... - see RAILWAY_SETUP.md), sets the +# rest of the driver's env to match run-comparisons.sh's baseline, deploys it, +# and polls its logs until a new "Done." line (the driver's own final line) +# appears. +run_driver() { + local num_servers=$1 clients_per_server=$2 sweep_start=$3 sweep_step=$4 sweep_max=$5 log_file=$6 backplane=$7 strategy=$8 + + local server_urls="" + for i in $(seq 1 "$num_servers"); do + server_urls+="http://server$i.railway.internal:8080," + done + server_urls="${server_urls%,}" + + # BACKPLANE/PAYLOAD_STRATEGY tell the driver whether to truncate backplane_payloads before this + # scenario starts (see PostgreSignalR.Benchmarks/Program.cs) - needed here because, unlike local + # docker-compose, this Postgres instance is never torn down between scenarios. + run railway variables --service driver \ + --set "SERVER_URLS=$server_urls" \ + --set "BACKPLANE=$backplane" \ + --set "PAYLOAD_STRATEGY=$strategy" \ + --set "MODE=$MODE" \ + --set "CLIENTS_PER_SERVER=$clients_per_server" \ + --set "PUBLISH_COUNT=$PUBLISH_COUNT" \ + --set "CONCURRENCY=$CONCURRENCY" \ + --set "PAYLOAD_BYTES=$PAYLOAD_BYTES" \ + --set "WARMUP_SECONDS=$WARMUP_SECONDS" \ + --set "TARGET_RATE=$TARGET_RATE" \ + --set "SLO_P99_MS=$SLO_P99_MS" \ + --set "SWEEP_START_RATE=$sweep_start" \ + --set "SWEEP_STEP_RATE=$sweep_step" \ + --set "SWEEP_MAX_RATE=$sweep_max" \ + --set "SWEEP_TRIAL_SECONDS=$SWEEP_TRIAL_SECONDS" \ + --set "BATCH_SIZE=$BATCH_SIZE" \ + --set "REPEATS_PER_RATE=$REPEATS_PER_RATE" \ + --set "HEALTH_CHECK_TIMEOUT_SECONDS=$HEALTH_CHECK_TIMEOUT_SECONDS" \ + --set "DRAIN_QUIET_SECONDS=$DRAIN_QUIET_SECONDS" \ + --set "DRAIN_MAX_WAIT_SECONDS=$DRAIN_MAX_WAIT_SECONDS" + + try_redeploy driver + + if $apply; then + # This CLI's `railway logs` has no --follow/streaming mode (confirmed: it rejects the + # flag outright), so this polls with repeated one-shot fetches instead of tailing a + # single background process. Each fetch may or may not be scoped to just the latest + # deployment - unconfirmed - so rather than trust the first "Done." we see (which could + # be left over from a previous scenario if fetches aren't scoped that way), we record how + # many completions are present before waiting, and wait for that count to increase. + # The match is intentionally NOT anchored to a whole line (^Done\.$) - fetched output has + # been observed out of logical print order and occasionally with lines run together, and + # "Done." only ever appears in the driver's own final line, so a plain substring match is + # both safe and more robust against that than requiring it to stand alone on its own line. + echo "+ railway logs --service driver (establishing baseline)" + railway logs --service driver > "$log_file" 2>/dev/null || true + + local baseline + baseline=$(grep -c "Done\." "$log_file" 2>/dev/null || true) + baseline=${baseline:-0} + + local waited=0 + local max_wait=3600 + local success=false + + while (( waited < max_wait )); do + sleep 10 + waited=$((waited + 10)) + + railway logs --service driver > "$log_file" 2>/dev/null || true + + local current + current=$(grep -c "Done\." "$log_file" 2>/dev/null || true) + current=${current:-0} + + if (( current > baseline )); then + success=true + break + fi + done + + if ! $success; then + echo " Warning: did not see a new \"Done.\" line within ${max_wait}s - check $log_file" + return 1 + fi + else + echo "[dry-run] railway logs --service driver (poll for a new \"Done.\" beyond whatever's already there)" + fi + + return 0 +} + +summary=() + +for entry in "${selected[@]}"; do + read -r name backplane shared num_servers strategy clients_per_server sweep_start sweep_step sweep_max <<< "$entry" + log_file="$results_dir/$name.log" + + echo "==============================================" + echo "Scenario: $name" + echo " BACKPLANE=$backplane SIMULATE_SHARED_LOAD=$shared NUM_SERVERS=$num_servers PAYLOAD_STRATEGY=$strategy" + echo " CLIENTS_PER_SERVER=$clients_per_server SWEEP=$sweep_start-$sweep_max step $sweep_step" + echo " Log: $log_file" + echo "==============================================" + + configure_scenario "$backplane" "$shared" "$strategy" + + status=0 + run_driver "$num_servers" "$clients_per_server" "$sweep_start" "$sweep_step" "$sweep_max" "$log_file" "$backplane" "$strategy" || status=$? + + summary+=("$name:$status") + + echo +done + +echo +echo "================= Summary =================" + +failures=0 + +for entry in "${summary[@]}"; do + name="${entry%%:*}" + status="${entry##*:}" + + if [ "$status" -eq 0 ]; then + echo " OK $name" + else + echo " FAIL $name (exit $status)" + failures=$((failures + 1)) + fi +done + +echo "Logs saved to: $results_dir" +echo "Services are left running (redeploy is much slower here than local docker-compose down/up)." +echo "Run './benchmarks/run-comparisons-railway.sh --apply --teardown' when you're done, to stop billing." +echo "==============================================" + +if [ "$failures" -gt 0 ]; then + exit 1 +fi diff --git a/benchmarks/run-comparisons.sh b/benchmarks/run-comparisons.sh new file mode 100644 index 0000000..2578a51 --- /dev/null +++ b/benchmarks/run-comparisons.sh @@ -0,0 +1,186 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Runs the backplane comparison matrix described in benchmarks/README.md. +# +# Each scenario is torn down (including volumes) before it starts, so a scenario +# never inherits leftover Postgres/Redis state (payload tables, shared-load rows, +# stale NOTIFY listeners) from the previous run. +# +# WARNING: this is slow - each scenario's rate sweep and CLIENTS_PER_SERVER are tuned per +# scenario now (see --list), and a full sweep is commonly 20-30 minutes, so the full matrix +# (16 scenarios) is several hours. Pass scenario names to run a subset, or override +# REPEATS_PER_RATE / SWEEP_TRIAL_SECONDS in your environment for a quicker smoke test +# (CLIENTS_PER_SERVER and the sweep rate bounds themselves are per-scenario, not overridable +# this way - edit the scenarios array directly if you need different values there). +# +# Usage: +# ./benchmarks/run-comparisons.sh # run every scenario +# ./benchmarks/run-comparisons.sh --list # print the matrix and exit +# ./benchmarks/run-comparisons.sh redis-dedicated postgres-shared # run only the named scenarios +# REPEATS_PER_RATE=1 SWEEP_TRIAL_SECONDS=5 ./benchmarks/run-comparisons.sh redis-dedicated +# # quick smoke test of one scenario + +cd "$(dirname "${BASH_SOURCE[0]}")/.." + +: "${MODE:=sweep}" +: "${PUBLISH_COUNT:=20000}" +: "${CONCURRENCY:=128}" +: "${PAYLOAD_BYTES:=128}" +: "${REPEATS_PER_RATE:=3}" + +export MODE PUBLISH_COUNT CONCURRENCY PAYLOAD_BYTES REPEATS_PER_RATE + +# Four groups, each answering a different question - see run-comparisons-railway.sh for the +# full rationale (this mirrors the same matrix): +# 1. redis/postgres-dedicated/shared: core backplane comparison + max sustainable rate. +# 2. postgres-*-table: same, with the payload-table strategy instead of the default event one. +# 3. *-clients-N: fixed low rate, CLIENTS_PER_SERVER pushed up in steps - max sustainable +# client count. Client count isn't swept within one run like rate is, so each checkpoint +# is its own scenario. +# 4. *-dedicated-Nservers: fixed client count/rate, more subscriber nodes - fan-out width. +# +# name backplane shared num_servers payload_strategy clients_per_server sweep_start sweep_step sweep_max +scenarios=( + "redis-dedicated redis false 2 event 50 100 100 2000" + "postgres-dedicated postgres false 2 event 50 100 100 2000" + "postgres-dedicated-table postgres false 2 table 50 100 100 2000" + "redis-shared redis true 2 event 50 100 100 2000" + "postgres-shared postgres true 2 event 50 100 100 2000" + "postgres-shared-table postgres true 2 table 50 100 100 2000" + "redis-clients-500 redis false 2 event 500 10 10 100" + "redis-clients-2000 redis false 2 event 2000 10 10 100" + "redis-clients-4000 redis false 2 event 4000 10 10 100" + "postgres-clients-500 postgres false 2 event 500 10 10 100" + "postgres-clients-2000 postgres false 2 event 2000 10 10 100" + "postgres-clients-4000 postgres false 2 event 4000 10 10 100" + "postgres-clients-500-table postgres false 2 table 500 10 10 100" + "postgres-clients-2000-table postgres false 2 table 2000 10 10 100" + "postgres-clients-4000-table postgres false 2 table 4000 10 10 100" + "redis-dedicated-5servers redis false 5 event 50 100 100 2000" + "postgres-dedicated-5servers postgres false 5 event 50 100 100 2000" + "redis-dedicated-10servers redis false 10 event 50 100 100 2000" + "postgres-dedicated-10servers postgres false 10 event 50 100 100 2000" +) + +print_matrix() { + printf '%-30s %-9s %-7s %-11s %-9s %-19s %-12s %-11s %s\n' \ + "NAME" "BACKPLANE" "SHARED" "NUM_SERVERS" "STRATEGY" "CLIENTS_PER_SERVER" "SWEEP_START" "SWEEP_STEP" "SWEEP_MAX" + for entry in "${scenarios[@]}"; do + read -r name backplane shared num_servers strategy clients_per_server sweep_start sweep_step sweep_max <<< "$entry" + printf '%-30s %-9s %-7s %-11s %-9s %-19s %-12s %-11s %s\n' \ + "$name" "$backplane" "$shared" "$num_servers" "$strategy" "$clients_per_server" "$sweep_start" "$sweep_step" "$sweep_max" + done +} + +requested=() +list_only=false + +for arg in "$@"; do + case "$arg" in + --list|-l) list_only=true ;; + *) requested+=("$arg") ;; + esac +done + +if $list_only; then + print_matrix + exit 0 +fi + +selected=() + +if [ ${#requested[@]} -eq 0 ]; then + selected=("${scenarios[@]}") +else + for name in "${requested[@]}"; do + found=false + + for entry in "${scenarios[@]}"; do + if [ "${entry%% *}" == "$name" ]; then + selected+=("$entry") + found=true + break + fi + done + + if ! $found; then + echo "Unknown scenario: $name" >&2 + echo "Run with --list to see available scenarios." >&2 + exit 1 + fi + done +fi + +timestamp=$(date +%Y%m%d-%H%M%S) +results_dir="benchmarks/results/$timestamp" +mkdir -p "$results_dir" + +echo "Baseline: MODE=$MODE PUBLISH_COUNT=$PUBLISH_COUNT CONCURRENCY=$CONCURRENCY PAYLOAD_BYTES=$PAYLOAD_BYTES REPEATS_PER_RATE=$REPEATS_PER_RATE (CLIENTS_PER_SERVER and sweep bounds are per-scenario now, see --list)" +echo "Results directory: $results_dir" +echo "Scenarios: $(for e in "${selected[@]}"; do printf '%s ' "${e%% *}"; done)" +echo + +summary=() + +cleanup() { + echo "Tearing down compose stack..." + docker compose down --volumes --remove-orphans >/dev/null 2>&1 || true +} +trap cleanup EXIT + +for entry in "${selected[@]}"; do + read -r name backplane shared num_servers strategy clients_per_server sweep_start sweep_step sweep_max <<< "$entry" + log_file="$results_dir/$name.log" + + echo "==============================================" + echo "Scenario: $name" + echo " BACKPLANE=$backplane SIMULATE_SHARED_LOAD=$shared NUM_SERVERS=$num_servers PAYLOAD_STRATEGY=$strategy" + echo " CLIENTS_PER_SERVER=$clients_per_server SWEEP=$sweep_start-$sweep_max step $sweep_step" + echo " Log: $log_file" + echo "==============================================" + + docker compose down --volumes --remove-orphans >/dev/null 2>&1 || true + + status=0 + BACKPLANE="$backplane" \ + SIMULATE_SHARED_LOAD="$shared" \ + NUM_SERVERS="$num_servers" \ + PAYLOAD_STRATEGY="$strategy" \ + CLIENTS_PER_SERVER="$clients_per_server" \ + SWEEP_START_RATE="$sweep_start" \ + SWEEP_STEP_RATE="$sweep_step" \ + SWEEP_MAX_RATE="$sweep_max" \ + docker compose up --build --abort-on-container-exit --exit-code-from driver 2>&1 | tee "$log_file" || status=$? + + summary+=("$name:$status") + + echo +done + +docker compose down --volumes --remove-orphans >/dev/null 2>&1 || true +trap - EXIT + +echo +echo "================= Summary =================" + +failures=0 + +for entry in "${summary[@]}"; do + name="${entry%%:*}" + status="${entry##*:}" + + if [ "$status" -eq 0 ]; then + echo " OK $name" + else + echo " FAIL $name (exit $status)" + failures=$((failures + 1)) + fi +done + +echo "Logs saved to: $results_dir" +echo "==============================================" + +if [ "$failures" -gt 0 ]; then + exit 1 +fi diff --git a/docker-compose.yml b/docker-compose.yml index 1be9027..69b34e2 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,3 +1,18 @@ +x-server: &server + build: + context: . + dockerfile: benchmarks/PostgreSignalR.Benchmarks.Server/Dockerfile + environment: + ASPNETCORE_ENVIRONMENT: Production + Logging__LogLevel__Default: Warning + BACKPLANE: ${BACKPLANE:-none} + ConnectionStrings__Redis: redis:6379 + ConnectionStrings__Postgres: Host=postgres;Database=bench;Username=postgres;Password=postgres + PAYLOAD_STRATEGY: ${PAYLOAD_STRATEGY:-event} + depends_on: + - postgres + - redis + services: postgres: image: postgres:16 @@ -8,58 +23,109 @@ services: ports: - "5432:5432" - # redis: - # image: redis:7 - # ports: - # - "6379:6379" + redis: + image: redis:7 + ports: + - "6379:6379" - servera: - build: - context: . - dockerfile: benchmarks/PostgreSignalR.Benchmarks.Server/Dockerfile - environment: - ASPNETCORE_ENVIRONMENT: Production - Logging__LogLevel__Default: Warning - BACKPLANE: ${BACKPLANE:-none} - # ConnectionStrings__Redis: redis:6379 - ConnectionStrings__Postgres: Host=postgres;Database=bench;Username=postgres;Password=postgres - MAKETABLE: true - depends_on: - - postgres - # - redis + server1: + <<: *server ports: - "8081:8080" - serverb: + server2: + <<: *server + ports: + - "8082:8080" + + server3: + <<: *server + ports: + - "8083:8080" + + server4: + <<: *server + ports: + - "8084:8080" + + server5: + <<: *server + ports: + - "8085:8080" + + server6: + <<: *server + ports: + - "8086:8080" + + server7: + <<: *server + ports: + - "8087:8080" + + server8: + <<: *server + ports: + - "8088:8080" + + server9: + <<: *server + ports: + - "8089:8080" + + server10: + <<: *server + ports: + - "8090:8080" + + shared-load: build: context: . - dockerfile: benchmarks/PostgreSignalR.Benchmarks.Server/Dockerfile + dockerfile: benchmarks/PostgreSignalR.Benchmarks.SharedLoad/Dockerfile environment: - ASPNETCORE_ENVIRONMENT: Production - Logging__LogLevel__Default: Warning BACKPLANE: ${BACKPLANE:-none} - # ConnectionStrings__Redis: redis:6379 + SIMULATE_SHARED_LOAD: ${SIMULATE_SHARED_LOAD:-false} + SHARED_LOAD_CONCURRENCY: ${SHARED_LOAD_CONCURRENCY:-16} + SHARED_LOAD_OPS_PER_SEC: ${SHARED_LOAD_OPS_PER_SEC:-200} + ConnectionStrings__Redis: redis:6379 ConnectionStrings__Postgres: Host=postgres;Database=bench;Username=postgres;Password=postgres depends_on: - postgres - # - redis - ports: - - "8082:8080" + - redis driver: build: context: . dockerfile: benchmarks/PostgreSignalR.Benchmarks/Dockerfile environment: - SERVER_A: http://servera:8080 - SERVER_B: http://serverb:8080 + NUM_SERVERS: ${NUM_SERVERS:-2} MODE: ${MODE:-single} - CLIENTS: ${CLIENTS:-500} + CLIENTS_PER_SERVER: ${CLIENTS_PER_SERVER:-500} PUBLISH_COUNT: ${PUBLISH_COUNT:-20000} CONCURRENCY: ${CONCURRENCY:-128} PAYLOAD_BYTES: ${PAYLOAD_BYTES:-128} WARMUP_SECONDS: ${WARMUP_SECONDS:-10} MEASURE_SECONDS: ${MEASURE_SECONDS:-30} + TARGET_RATE: ${TARGET_RATE:-100} + SLO_P99_MS: ${SLO_P99_MS:-250} + SWEEP_START_RATE: ${SWEEP_START_RATE:-100} + SWEEP_STEP_RATE: ${SWEEP_STEP_RATE:-100} + SWEEP_MAX_RATE: ${SWEEP_MAX_RATE:-2000} + SWEEP_TRIAL_SECONDS: ${SWEEP_TRIAL_SECONDS:-15} + BATCH_SIZE: ${BATCH_SIZE:-25} + REPEATS_PER_RATE: ${REPEATS_PER_RATE:-1} + HEALTH_CHECK_TIMEOUT_SECONDS: ${HEALTH_CHECK_TIMEOUT_SECONDS:-60} + DRAIN_QUIET_SECONDS: ${DRAIN_QUIET_SECONDS:-1} + DRAIN_MAX_WAIT_SECONDS: ${DRAIN_MAX_WAIT_SECONDS:-60} depends_on: - - servera - - serverb + - server1 + - server2 + - server3 + - server4 + - server5 + - server6 + - server7 + - server8 + - server9 + - server10 + - shared-load diff --git a/src/PostgresHubLifetimeManager.cs b/src/PostgresHubLifetimeManager.cs index bc3d249..b9a6176 100644 --- a/src/PostgresHubLifetimeManager.cs +++ b/src/PostgresHubLifetimeManager.cs @@ -39,7 +39,6 @@ public sealed class PostgresHubLifetimeManager : HubLifetimeManager, private readonly PostgresBackplaneOptions _options; private readonly string _serverName = GenerateServerName(); private readonly PostgresProtocol _protocol; - private readonly SemaphoreSlim _commandLock = new(1); private readonly SemaphoreSlim _initializationLock = new(1, 1); private readonly ConcurrentDictionary> _notificationHandlers = new(StringComparer.Ordinal); private readonly ClientResultsManager _clientResultsManager = new(); @@ -377,7 +376,6 @@ private async Task PublishAsync(string channel, byte[] message) { _logger.BackplanePublishing(channel); - await _commandLock.WaitAsync(); try { await _payloadStrategy.NotifyAsync(channel, message); @@ -387,10 +385,6 @@ private async Task PublishAsync(string channel, byte[] message) _logger.BackplaneUnableToConnect(ex); throw; } - finally - { - _commandLock.Release(); - } } private Task AddGroupAsyncCore(HubConnectionContext connection, string groupName)