test(perf): measure phase-8 WP-4 residual criteria xc-3/4/5/8 (#49)
New ResidualBench tool (eventlog, flatten, audit, audit-contention) and docker/measure-ui-responsiveness.sh. All four residuals pass; results in docs/plans/2026-11-10-wp4-residual-scope-results.md. Findings filed as #122, #123 and #124 (fixed in the following commit).
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,221 @@
|
||||
using System.Diagnostics;
|
||||
using Microsoft.Data.SqlClient;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using ZB.MOM.WW.Audit;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase;
|
||||
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.ResidualBench;
|
||||
|
||||
/// <summary>
|
||||
/// [xc-4] Audit logging does not degrade central performance. Creates a throwaway, fully
|
||||
/// migrated configuration database (the production schema: partitioned <c>AuditLog</c>,
|
||||
/// clustered <c>(OccurredAtUtc, EventId)</c>, every NC index), seeds it through the real
|
||||
/// <see cref="AuditLogRepository.InsertManyIfNotExistsAsync"/> ingest path, then runs a fixed
|
||||
/// central read mix while site-shaped ingest packets arrive at increasing offered rates.
|
||||
/// </summary>
|
||||
internal static class AuditBench
|
||||
{
|
||||
internal const string DefaultAdmin =
|
||||
"Server=localhost,1433;User Id=sa;Password=ScadaBridge_Dev1#;TrustServerCertificate=true;Encrypt=false";
|
||||
|
||||
private const int PacketSize = 256; // rows per site ingest packet
|
||||
|
||||
public static async Task<int> RunAsync(BenchArgs args)
|
||||
{
|
||||
var admin = args.Get("connection-string", DefaultAdmin);
|
||||
var seedRows = args.GetInt("seed-rows", 1_000_000);
|
||||
var phaseSeconds = args.GetInt("phase-seconds", 60);
|
||||
var keep = args.Get("keep", "false") == "true";
|
||||
args.RejectUnknown();
|
||||
|
||||
var dbName = $"ScadaBridgeResidualBench_{Guid.NewGuid():N}"[..40];
|
||||
var cs = new SqlConnectionStringBuilder(admin) { InitialCatalog = dbName }.ConnectionString;
|
||||
await ExecAsync(admin, $"CREATE DATABASE [{dbName}]; ALTER DATABASE [{dbName}] SET RECOVERY SIMPLE;");
|
||||
Console.WriteLine($"# database {dbName}");
|
||||
try
|
||||
{
|
||||
var options = new DbContextOptionsBuilder<ScadaBridgeDbContext>()
|
||||
.UseSqlServer(cs, sql => sql.CommandTimeout(1800)).Options;
|
||||
var sw = Stopwatch.StartNew();
|
||||
await using (var ctx = new ScadaBridgeDbContext(options)) await ctx.Database.MigrateAsync();
|
||||
Console.WriteLine($"migrated in {sw.Elapsed.TotalSeconds:F0} s");
|
||||
|
||||
// ---- seed: also the unthrottled ingest ceiling on a growing table ----------------
|
||||
var now = DateTime.UtcNow;
|
||||
sw.Restart();
|
||||
var seeded = 0L;
|
||||
// ONE writer: production ingest is the AuditLogIngestActor cluster singleton, which
|
||||
// awaits each packet inside ReceiveAsync, so central never runs two site-ingest
|
||||
// inserts at once. (Four concurrent writers here deadlock — error 1205 — see results doc.)
|
||||
var packets = (seedRows + PacketSize - 1) / PacketSize;
|
||||
for (var packet = 0; packet < packets; packet++)
|
||||
{
|
||||
var rows = Enumerable.Range(0, PacketSize)
|
||||
.Select(i => Row((long)packet * PacketSize + i, now.AddDays(-45).AddSeconds(45.0 * 86_400 * ((long)packet * PacketSize + i) / seedRows)))
|
||||
.ToList();
|
||||
seeded += await InsertAsync(options, rows);
|
||||
}
|
||||
Console.WriteLine($"seeded {seeded:N0} rows (45 days) in {sw.Elapsed.TotalSeconds:F0} s = {seeded / sw.Elapsed.TotalSeconds:N0} rows/s (1 serial writer, {PacketSize}-row packets)");
|
||||
Console.WriteLine($"AuditLog size {await ScalarAsync(cs, "SELECT SUM(used_page_count) * 8 / 1024 FROM sys.dm_db_partition_stats WHERE object_id = OBJECT_ID('dbo.AuditLog')")} MB");
|
||||
|
||||
// ---- phases ------------------------------------------------------------------
|
||||
Console.WriteLine();
|
||||
var phases = new (string Name, int RowsPerSec)[] { ("no ingest", 0), ("ingest 100 rows/s", 100), ("ingest 500 rows/s", 500), ("ingest 2,000 rows/s", 2_000), ("ingest unthrottled (1 serial writer)", -1) };
|
||||
foreach (var (name, rate) in phases)
|
||||
await RunPhaseAsync(options, name, rate, TimeSpan.FromSeconds(phaseSeconds));
|
||||
}
|
||||
finally
|
||||
{
|
||||
SqlConnection.ClearAllPools();
|
||||
if (!keep)
|
||||
await ExecAsync(admin, $"ALTER DATABASE [{dbName}] SET SINGLE_USER WITH ROLLBACK IMMEDIATE; DROP DATABASE [{dbName}];");
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
private static readonly (string Name, Func<ScadaBridgeDbContext, Task> Run)[] Reads =
|
||||
[
|
||||
("Audit Log page 1 (newest 50, no filter)", ctx => new AuditLogRepository(ctx).QueryAsync(new AuditLogQueryFilter(), new AuditLogPaging(50))),
|
||||
("channel=ApiOutbound, last 24 h, 50", ctx => new AuditLogRepository(ctx).QueryAsync(
|
||||
new AuditLogQueryFilter(Channels: [AuditChannel.ApiOutbound], FromUtc: DateTime.UtcNow.AddDays(-1)), new AuditLogPaging(50))),
|
||||
("site=site-07, last 7 d, 100", ctx => new AuditLogRepository(ctx).QueryAsync(
|
||||
new AuditLogQueryFilter(SourceSiteIds: ["site-07"], FromUtc: DateTime.UtcNow.AddDays(-7)), new AuditLogPaging(100))),
|
||||
("status=Failed, last 24 h, 50", ctx => new AuditLogRepository(ctx).QueryAsync(
|
||||
new AuditLogQueryFilter(Statuses: [AuditStatus.Failed], FromUtc: DateTime.UtcNow.AddDays(-1)), new AuditLogPaging(50))),
|
||||
("KPI snapshot, 1 h window", ctx => new AuditLogRepository(ctx).GetKpiSnapshotAsync(TimeSpan.FromHours(1))),
|
||||
("non-audit: templates + sites list", async ctx =>
|
||||
{
|
||||
await ctx.Templates.AsNoTracking().Take(100).ToListAsync();
|
||||
await ctx.Sites.AsNoTracking().ToListAsync();
|
||||
}),
|
||||
];
|
||||
|
||||
private static async Task RunPhaseAsync(DbContextOptions<ScadaBridgeDbContext> options, string name, int rowsPerSec, TimeSpan duration)
|
||||
{
|
||||
using var cts = new CancellationTokenSource(duration);
|
||||
var ingested = 0L;
|
||||
var packetLatency = new Samples();
|
||||
var next = 10_000_000L + Random.Shared.Next(1_000_000) * 100L;
|
||||
|
||||
async Task Writer(int writers, int index)
|
||||
{
|
||||
var sw = Stopwatch.StartNew();
|
||||
var sent = 0L;
|
||||
while (!cts.IsCancellationRequested)
|
||||
{
|
||||
if (rowsPerSec > 0)
|
||||
{
|
||||
// Paced: one packet whenever this writer falls behind its share of the offered rate.
|
||||
var due = (long)(sw.Elapsed.TotalSeconds * rowsPerSec / writers);
|
||||
if (sent + PacketSize > due) { await Task.Delay(2); continue; }
|
||||
}
|
||||
var baseId = Interlocked.Add(ref next, PacketSize);
|
||||
var rows = Enumerable.Range(0, PacketSize).Select(i => Row(baseId + i, DateTime.UtcNow.AddSeconds(-Random.Shared.Next(0, 30)))).ToList();
|
||||
var t = Stopwatch.GetTimestamp();
|
||||
Interlocked.Add(ref ingested, await InsertAsync(options, rows));
|
||||
packetLatency.Add(Stopwatch.GetElapsedTime(t).TotalMilliseconds);
|
||||
sent += PacketSize;
|
||||
}
|
||||
}
|
||||
|
||||
var writerCount = rowsPerSec == 0 ? 0 : 1;
|
||||
var writers = Enumerable.Range(0, writerCount).Select(i => Task.Run(() => Writer(writerCount, i))).ToList();
|
||||
|
||||
// One reader looping the read mix back to back — a busy operator's UI plus the health poll.
|
||||
var latencies = Reads.ToDictionary(r => r.Name, _ => new Samples());
|
||||
var started = Stopwatch.StartNew();
|
||||
while (!cts.IsCancellationRequested)
|
||||
{
|
||||
foreach (var (readName, run) in Reads)
|
||||
{
|
||||
var t = Stopwatch.GetTimestamp();
|
||||
await using var ctx = new ScadaBridgeDbContext(options);
|
||||
await run(ctx);
|
||||
latencies[readName].Add(Stopwatch.GetElapsedTime(t).TotalMilliseconds);
|
||||
}
|
||||
}
|
||||
await Task.WhenAll(writers);
|
||||
var seconds = started.Elapsed.TotalSeconds;
|
||||
|
||||
Console.WriteLine($"### {name}: achieved ingest {ingested / seconds:N0} rows/s, deadlock retries so far {Interlocked.Read(ref _deadlocks)}" +
|
||||
(packetLatency.Count > 0 ? $"; {PacketSize}-row packet {packetLatency.Summary()}" : ""));
|
||||
Console.WriteLine("| read | P50 ms | P95 ms | P99 ms | max ms | n |");
|
||||
Console.WriteLine("|---|---|---|---|---|---|");
|
||||
foreach (var (readName, s) in latencies)
|
||||
Console.WriteLine($"| {readName} | {s.Percentile(50):F1} | {s.Percentile(95):F1} | {s.Percentile(99):F1} | {s.Max:F1} | {s.Count} |");
|
||||
Console.WriteLine();
|
||||
}
|
||||
|
||||
private static readonly string[] Targets = ["MES.MoveIn", "MES.MoveOut", "Historian.Write", "ERP.Order", "Delmia.Recipe"];
|
||||
|
||||
/// <summary>Site-shaped row: ~1.5 KB of request/response summary + extra, 10 sites × 500 instances.</summary>
|
||||
internal static AuditEvent Row(long i, DateTime occurredAtUtc)
|
||||
{
|
||||
var channel = (i % 10) switch { < 5 => AuditChannel.ApiOutbound, < 8 => AuditChannel.DbOutbound, < 9 => AuditChannel.Notification, _ => AuditChannel.ApiInbound };
|
||||
var kind = channel switch
|
||||
{
|
||||
AuditChannel.ApiOutbound => i % 3 == 0 ? AuditKind.ApiCallCached : AuditKind.ApiCall,
|
||||
AuditChannel.DbOutbound => i % 3 == 0 ? AuditKind.DbWriteCached : AuditKind.DbWrite,
|
||||
AuditChannel.Notification => AuditKind.NotifySend,
|
||||
_ => AuditKind.InboundRequest,
|
||||
};
|
||||
var failed = i % 25 == 0;
|
||||
var site = $"site-{i % 10:D2}";
|
||||
return ScadaBridgeAuditEventFactory.Create(
|
||||
channel, kind, failed ? AuditStatus.Failed : AuditStatus.Delivered,
|
||||
occurredAtUtc: occurredAtUtc,
|
||||
actor: $"script:Machine-{i % 500:D4}",
|
||||
target: Targets[i % Targets.Length],
|
||||
sourceNode: i % 2 == 0 ? "node-a" : "node-b",
|
||||
correlationId: kind is AuditKind.ApiCallCached or AuditKind.DbWriteCached ? Guid.NewGuid() : null,
|
||||
executionId: Guid.NewGuid(),
|
||||
sourceSiteId: site,
|
||||
sourceInstanceId: $"Machine-{i % 500:D4}",
|
||||
sourceScript: $"Recipe{i % 11}",
|
||||
httpStatus: channel is AuditChannel.ApiOutbound or AuditChannel.ApiInbound ? (failed ? 503 : 200) : null,
|
||||
durationMs: (int)(i % 400),
|
||||
errorMessage: failed ? "Service Unavailable: upstream timeout after 30000 ms" : null,
|
||||
requestSummary: "{\"workOrder\":\"WO-" + i + "\",\"machine\":\"Machine-" + (i % 500).ToString("D4") + "\",\"items\":[" +
|
||||
string.Join(",", Enumerable.Range(0, 8).Select(k => "{\"sku\":\"SKU-" + k + "\",\"qty\":" + (k * 3) + "}")) + "]}",
|
||||
responseSummary: failed ? null : "{\"accepted\":true,\"reference\":\"" + Guid.NewGuid() + "\",\"message\":\"Recorded\"}",
|
||||
extra: "{\"retry\":0,\"bufferDepth\":" + (i % 50) + "}");
|
||||
}
|
||||
|
||||
private static long _deadlocks;
|
||||
|
||||
/// <summary>One ingest packet through the production repository; a deadlock victim (1205) is retried, as the site would re-send it.</summary>
|
||||
private static async Task<int> InsertAsync(DbContextOptions<ScadaBridgeDbContext> options, IReadOnlyList<AuditEvent> rows)
|
||||
{
|
||||
for (var attempt = 1; ; attempt++)
|
||||
{
|
||||
try
|
||||
{
|
||||
await using var ctx = new ScadaBridgeDbContext(options);
|
||||
return await new AuditLogRepository(ctx).InsertManyIfNotExistsAsync(rows);
|
||||
}
|
||||
catch (SqlException ex) when (ex.Number == 1205 && attempt < 5)
|
||||
{
|
||||
Interlocked.Increment(ref _deadlocks);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
internal static async Task ExecAsync(string cs, string sql)
|
||||
{
|
||||
await using var c = new SqlConnection(cs);
|
||||
await c.OpenAsync();
|
||||
await using var cmd = new SqlCommand(sql, c) { CommandTimeout = 300 };
|
||||
await cmd.ExecuteNonQueryAsync();
|
||||
}
|
||||
|
||||
internal static async Task<object?> ScalarAsync(string cs, string sql)
|
||||
{
|
||||
await using var c = new SqlConnection(cs);
|
||||
await c.OpenAsync();
|
||||
await using var cmd = new SqlCommand(sql, c) { CommandTimeout = 300 };
|
||||
return await cmd.ExecuteScalarAsync();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,105 @@
|
||||
using System.Diagnostics;
|
||||
using Microsoft.Data.SqlClient;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using ZB.MOM.WW.Audit;
|
||||
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase;
|
||||
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.ResidualBench;
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#124: concurrent callers of <see cref="AuditLogRepository.InsertManyIfNotExistsAsync"/>
|
||||
/// — the shape the ingest singleton, the reconciliation singleton and the cached-telemetry
|
||||
/// dual-write produce when they overlap. Runs N writers against a throwaway migrated database
|
||||
/// for a fixed time and reports throughput and the deadlock victims (error 1205) that reached
|
||||
/// the caller. One writer in N re-sends a packet another writer already inserted, the way a
|
||||
/// reconciliation pull overlaps live telemetry.
|
||||
/// </summary>
|
||||
internal static class AuditContentionBench
|
||||
{
|
||||
private const int PacketSize = 256;
|
||||
|
||||
public static async Task<int> RunAsync(BenchArgs args)
|
||||
{
|
||||
var admin = args.Get("connection-string", AuditBench.DefaultAdmin);
|
||||
var writerCount = args.GetInt("writers", 4);
|
||||
var seconds = args.GetInt("seconds", 30);
|
||||
var seedRows = args.GetInt("seed-rows", 50_000);
|
||||
args.RejectUnknown();
|
||||
|
||||
var dbName = $"ScadaBridgeResidualBench_{Guid.NewGuid():N}"[..40];
|
||||
var cs = new SqlConnectionStringBuilder(admin) { InitialCatalog = dbName }.ConnectionString;
|
||||
await AuditBench.ExecAsync(admin, $"CREATE DATABASE [{dbName}]; ALTER DATABASE [{dbName}] SET RECOVERY SIMPLE;");
|
||||
Console.WriteLine($"# database {dbName}, {writerCount} writers, {seconds} s, {seedRows:N0} seed rows");
|
||||
try
|
||||
{
|
||||
var options = new DbContextOptionsBuilder<ScadaBridgeDbContext>()
|
||||
.UseSqlServer(cs, sql => sql.CommandTimeout(600)).Options;
|
||||
await using (var ctx = new ScadaBridgeDbContext(options)) await ctx.Database.MigrateAsync();
|
||||
|
||||
var now = DateTime.UtcNow;
|
||||
for (var p = 0; p < seedRows / PacketSize; p++)
|
||||
{
|
||||
var rows = Enumerable.Range(0, PacketSize)
|
||||
.Select(i => AuditBench.Row((long)p * PacketSize + i, now.AddHours(-2).AddMilliseconds((p * PacketSize) + i)))
|
||||
.ToList();
|
||||
await using var ctx = new ScadaBridgeDbContext(options);
|
||||
await new AuditLogRepository(ctx).InsertManyIfNotExistsAsync(rows);
|
||||
}
|
||||
|
||||
long inserted = 0, packets = 0, deadlocks = 0, otherFaults = 0, seq = 1_000_000_000;
|
||||
var recent = new System.Collections.Concurrent.ConcurrentQueue<IReadOnlyList<AuditEvent>>();
|
||||
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(seconds));
|
||||
var sw = Stopwatch.StartNew();
|
||||
|
||||
async Task Writer(int index)
|
||||
{
|
||||
while (!cts.IsCancellationRequested)
|
||||
{
|
||||
IReadOnlyList<AuditEvent> rows;
|
||||
// Writer 0 replays an already-sent packet now and then (reconciliation overlap).
|
||||
if (index == 0 && writerCount > 1 && Interlocked.Read(ref packets) % 4 == 3 && recent.TryDequeue(out var replay))
|
||||
{
|
||||
rows = replay;
|
||||
}
|
||||
else
|
||||
{
|
||||
var start = Interlocked.Add(ref seq, PacketSize) - PacketSize;
|
||||
var t = DateTime.UtcNow;
|
||||
rows = Enumerable.Range(0, PacketSize)
|
||||
.Select(i => AuditBench.Row(start + i, t.AddMilliseconds(-i)))
|
||||
.ToList();
|
||||
recent.Enqueue(rows);
|
||||
while (recent.Count > 64) recent.TryDequeue(out _);
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await using var ctx = new ScadaBridgeDbContext(options);
|
||||
Interlocked.Add(ref inserted, await new AuditLogRepository(ctx).InsertManyIfNotExistsAsync(rows));
|
||||
Interlocked.Increment(ref packets);
|
||||
}
|
||||
catch (SqlException ex) when (ex.Number == 1205)
|
||||
{
|
||||
Interlocked.Increment(ref deadlocks);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
if (Interlocked.Increment(ref otherFaults) <= 3) Console.WriteLine($"fault: {ex.GetBaseException().Message}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
await Task.WhenAll(Enumerable.Range(0, writerCount).Select(i => Task.Run(() => Writer(i))));
|
||||
Console.WriteLine(
|
||||
$"{writerCount} writers: {packets:N0} packets, {inserted:N0} rows in {sw.Elapsed.TotalSeconds:F0} s = " +
|
||||
$"{inserted / sw.Elapsed.TotalSeconds:N0} rows/s; deadlock victims reaching the caller {deadlocks:N0}; other faults {otherFaults:N0}");
|
||||
}
|
||||
finally
|
||||
{
|
||||
SqlConnection.ClearAllPools();
|
||||
await AuditBench.ExecAsync(admin, $"ALTER DATABASE [{dbName}] SET SINGLE_USER WITH ROLLBACK IMMEDIATE; DROP DATABASE [{dbName}];");
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
using System.Diagnostics;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.ResidualBench;
|
||||
|
||||
/// <summary><c>--key value</c> options. An unknown key is rejected by the consuming mode, not silently ignored.</summary>
|
||||
internal sealed class BenchArgs
|
||||
{
|
||||
private readonly Dictionary<string, string> _values = new(StringComparer.Ordinal);
|
||||
private readonly HashSet<string> _read = new(StringComparer.Ordinal);
|
||||
|
||||
public static BenchArgs Parse(IEnumerable<string> args)
|
||||
{
|
||||
var result = new BenchArgs();
|
||||
var list = args.ToList();
|
||||
for (var i = 0; i < list.Count; i++)
|
||||
{
|
||||
if (!list[i].StartsWith("--", StringComparison.Ordinal))
|
||||
throw new ArgumentException($"unexpected argument '{list[i]}'");
|
||||
var key = list[i][2..];
|
||||
var value = i + 1 < list.Count && !list[i + 1].StartsWith("--", StringComparison.Ordinal) ? list[++i] : "true";
|
||||
result._values[key] = value;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
public string Get(string key, string fallback)
|
||||
{
|
||||
_read.Add(key);
|
||||
return _values.TryGetValue(key, out var v) ? v : fallback;
|
||||
}
|
||||
|
||||
public int GetInt(string key, int fallback) => int.Parse(Get(key, fallback.ToString()));
|
||||
|
||||
/// <summary>Throws if any supplied key was never read — a typo must not silently change the run.</summary>
|
||||
public void RejectUnknown()
|
||||
{
|
||||
var unknown = _values.Keys.Where(k => !_read.Contains(k)).ToList();
|
||||
if (unknown.Count > 0)
|
||||
throw new ArgumentException($"unknown option(s): {string.Join(", ", unknown.Select(k => "--" + k))}");
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Latency sample set with nearest-rank percentiles.</summary>
|
||||
internal sealed class Samples
|
||||
{
|
||||
private readonly List<double> _ms = new();
|
||||
|
||||
public int Count => _ms.Count;
|
||||
|
||||
public void Add(double ms) { lock (_ms) _ms.Add(ms); }
|
||||
|
||||
public double Percentile(double p)
|
||||
{
|
||||
lock (_ms)
|
||||
{
|
||||
if (_ms.Count == 0) return double.NaN;
|
||||
var sorted = _ms.OrderBy(x => x).ToList();
|
||||
var rank = (int)Math.Ceiling(p / 100.0 * sorted.Count) - 1;
|
||||
return sorted[Math.Clamp(rank, 0, sorted.Count - 1)];
|
||||
}
|
||||
}
|
||||
|
||||
public double Max { get { lock (_ms) return _ms.Count == 0 ? double.NaN : _ms.Max(); } }
|
||||
|
||||
public string Summary() =>
|
||||
$"n={Count} P50 {Percentile(50):F2} ms, P95 {Percentile(95):F2} ms, P99 {Percentile(99):F2} ms, max {Max:F2} ms";
|
||||
|
||||
public static double Time(Action action)
|
||||
{
|
||||
var sw = Stopwatch.StartNew();
|
||||
action();
|
||||
return sw.Elapsed.TotalMilliseconds;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,377 @@
|
||||
using System.Diagnostics;
|
||||
using System.Reflection;
|
||||
using Microsoft.Data.Sqlite;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Microsoft.Extensions.Options;
|
||||
using ZB.MOM.WW.LocalDb;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.RemoteQuery;
|
||||
using ZB.MOM.WW.ScadaBridge.SiteEventLogging;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.ResidualBench;
|
||||
|
||||
/// <summary>
|
||||
/// [xc-3] Site event logging handles high event volume within the 30-day / 1 GB limits.
|
||||
/// Everything runs over a real <c>ZB.MOM.WW.LocalDb</c> file with the production
|
||||
/// <see cref="SiteEventLogger"/>, <see cref="EventLogQueryService"/> and
|
||||
/// <see cref="EventLogPurgeService"/>.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Phases:
|
||||
/// <list type="number">
|
||||
/// <item><b>calibrate</b> — bytes per row on disk for the representative event mix, so
|
||||
/// rows-at-cap and the sustained event rate that fits 30 days inside 1 GB.</item>
|
||||
/// <item><b>write</b> — writer drain rate and enqueue→durable latency, with and without
|
||||
/// the LocalDb CDC capture triggers (<c>RegisterReplicated</c>), plus storm bursts against
|
||||
/// the bounded write queue.</item>
|
||||
/// <item><b>full</b> — a table filled to the 1 GB cap with 30 days of history: query
|
||||
/// shapes through the real query service, and writer latency while queries run.</item>
|
||||
/// <item><b>purge</b> — one real purge tick over a table 10% over the cap.</item>
|
||||
/// </list>
|
||||
/// </remarks>
|
||||
internal static class EventLogBench
|
||||
{
|
||||
private const int Instances = 500;
|
||||
private const long CapBytes = 1024L * 1024 * 1024;
|
||||
|
||||
public static async Task<int> RunAsync(BenchArgs args)
|
||||
{
|
||||
var dataDir = args.Get("data-dir", Path.Combine(Path.GetTempPath(), "sb-residual-eventlog"));
|
||||
var calibrationRows = args.GetInt("calibration-rows", 200_000);
|
||||
var writeEvents = args.GetInt("write-events", 200_000);
|
||||
var skipFull = args.Get("skip-full", "false") == "true";
|
||||
args.RejectUnknown();
|
||||
|
||||
Directory.CreateDirectory(dataDir);
|
||||
Console.WriteLine($"# data dir {dataDir}");
|
||||
|
||||
// ---- 1. calibrate ------------------------------------------------------------
|
||||
double bytesPerRow;
|
||||
{
|
||||
var db = Open(Path.Combine(dataDir, "calibrate.db"), out var path);
|
||||
using var logger = NewLogger(db, new SiteEventLogOptions());
|
||||
var before = UsedBytes(db);
|
||||
Seed(db, calibrationRows, DateTimeOffset.UtcNow.AddDays(-30), DateTimeOffset.UtcNow);
|
||||
var after = UsedBytes(db);
|
||||
bytesPerRow = (after - before) / (double)calibrationRows;
|
||||
var rowsAtCap = (long)(CapBytes / bytesPerRow);
|
||||
var ratePerSec = rowsAtCap / (30.0 * 86_400);
|
||||
Console.WriteLine();
|
||||
Console.WriteLine("## calibrate");
|
||||
Console.WriteLine($"rows seeded {calibrationRows:N0}");
|
||||
Console.WriteLine($"bytes per row (on disk) {bytesPerRow:F1}");
|
||||
Console.WriteLine($"rows at the 1 GB cap {rowsAtCap:N0}");
|
||||
Console.WriteLine($"30 days fit in 1 GB at <= {ratePerSec:F3} events/s sustained per site " +
|
||||
$"({ratePerSec * 86_400:N0}/day, {ratePerSec * 86_400 / Instances:F1}/instance/day at {Instances} instances)");
|
||||
Console.WriteLine($"days retained at 5 ev/s {rowsAtCap / (5.0 * 86_400):F1}");
|
||||
Console.WriteLine($"days retained at 20 ev/s {rowsAtCap / (20.0 * 86_400):F1}");
|
||||
logger.Dispose();
|
||||
Cleanup(db, path);
|
||||
}
|
||||
|
||||
// ---- 2. write throughput -----------------------------------------------------
|
||||
Console.WriteLine();
|
||||
Console.WriteLine("## write");
|
||||
foreach (var replicated in new[] { false, true })
|
||||
{
|
||||
var db = Open(Path.Combine(dataDir, $"write-{replicated}.db"), out var path);
|
||||
using var logger = NewLogger(db, new SiteEventLogOptions());
|
||||
if (replicated) db.RegisterReplicated("site_events");
|
||||
await MeasureDrainAsync(logger, writeEvents, replicated ? "CDC on " : "CDC off");
|
||||
foreach (var burst in new[] { 5_000, 10_000, 25_000 })
|
||||
await MeasureBurstAsync(logger, burst, replicated ? "CDC on " : "CDC off");
|
||||
logger.Dispose();
|
||||
Cleanup(db, path);
|
||||
}
|
||||
|
||||
if (skipFull) return 0;
|
||||
|
||||
// ---- 3. a table at the cap ---------------------------------------------------
|
||||
Console.WriteLine();
|
||||
Console.WriteLine("## full (1 GB, 30 days)");
|
||||
var fullRows = (long)(CapBytes / bytesPerRow);
|
||||
{
|
||||
var db = Open(Path.Combine(dataDir, "full.db"), out var path);
|
||||
var options = new SiteEventLogOptions();
|
||||
using var logger = NewLogger(db, options);
|
||||
var now = DateTimeOffset.UtcNow;
|
||||
var sw = Stopwatch.StartNew();
|
||||
Seed(db, fullRows, now.AddDays(-30), now);
|
||||
Console.WriteLine($"seeded {fullRows:N0} rows in {sw.Elapsed.TotalSeconds:F0} s; used {UsedBytes(db) / 1048576.0:F0} MiB, file {new FileInfo(path).Length / 1048576.0:F0} MiB");
|
||||
|
||||
var query = new EventLogQueryService(logger, Options.Create(options), NullLogger<EventLogQueryService>.Instance);
|
||||
RunQueries(query, now);
|
||||
await MeasureWriterUnderQueriesAsync(logger, query, now);
|
||||
|
||||
// ---- 4. purge tick over the cap --------------------------------------------
|
||||
Console.WriteLine();
|
||||
Console.WriteLine("## purge (cap trim, 10% over)");
|
||||
Seed(db, fullRows / 10, now.AddDays(-29), now);
|
||||
var purge = new EventLogPurgeService(logger, Options.Create(options), NullLogger<EventLogPurgeService>.Instance);
|
||||
var runPurge = typeof(EventLogPurgeService).GetMethod("RunPurge", BindingFlags.Instance | BindingFlags.NonPublic)
|
||||
?? throw new MissingMethodException("EventLogPurgeService.RunPurge");
|
||||
var countBefore = Count(db);
|
||||
sw.Restart();
|
||||
runPurge.Invoke(purge, null);
|
||||
var purgeMs = sw.Elapsed.TotalMilliseconds;
|
||||
Console.WriteLine($"rows {countBefore:N0} -> {Count(db):N0} in {purgeMs / 1000:F1} s (ceiling {options.MaxPurgedRowsPerTick:N0}/tick)");
|
||||
logger.Dispose();
|
||||
Cleanup(db, path);
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
private static async Task MeasureDrainAsync(SiteEventLogger logger, int total, string label)
|
||||
{
|
||||
// Windows below the queue capacity, so nothing is dropped: measures the writer's drain rate.
|
||||
const int window = 5_000;
|
||||
var latency = new Samples();
|
||||
var sw = Stopwatch.StartNew();
|
||||
var failedBefore = logger.FailedWriteCount;
|
||||
for (var offset = 0; offset < total; offset += window)
|
||||
{
|
||||
var tasks = new Task[window];
|
||||
for (var i = 0; i < window; i++)
|
||||
{
|
||||
var start = Stopwatch.GetTimestamp();
|
||||
var e = RowFor(offset + i);
|
||||
tasks[i] = logger.LogEventAsync(e.Type, e.Severity, e.Instance, e.Source, e.Message, e.Details)
|
||||
.ContinueWith(_ => latency.Add(Stopwatch.GetElapsedTime(start).TotalMilliseconds), TaskScheduler.Default);
|
||||
}
|
||||
await Task.WhenAll(tasks);
|
||||
}
|
||||
var seconds = sw.Elapsed.TotalSeconds;
|
||||
Console.WriteLine($"{label} drain: {total:N0} events in {seconds:F2} s = {total / seconds:N0} events/s; " +
|
||||
$"failed {logger.FailedWriteCount - failedBefore}; enqueue->durable {latency.Summary()}");
|
||||
}
|
||||
|
||||
private static async Task MeasureBurstAsync(SiteEventLogger logger, int burst, string label)
|
||||
{
|
||||
// A storm offered all at once (e.g. every alarm on a site flipping on one data-connection drop).
|
||||
var failedBefore = logger.FailedWriteCount;
|
||||
var sw = Stopwatch.StartNew();
|
||||
var tasks = new Task[burst];
|
||||
for (var i = 0; i < burst; i++)
|
||||
{
|
||||
var e = RowFor(i);
|
||||
tasks[i] = logger.LogEventAsync(e.Type, e.Severity, e.Instance, e.Source, e.Message, e.Details);
|
||||
}
|
||||
try { await Task.WhenAll(tasks); } catch { /* dropped events fault their task; counted below */ }
|
||||
Console.WriteLine($"{label} burst {burst,6:N0} at once: all settled in {sw.Elapsed.TotalMilliseconds:F0} ms, " +
|
||||
$"dropped {logger.FailedWriteCount - failedBefore:N0} (queue capacity {new SiteEventLogOptions().WriteQueueCapacity:N0})");
|
||||
}
|
||||
|
||||
private static void RunQueries(EventLogQueryService query, DateTimeOffset now)
|
||||
{
|
||||
var shapes = new (string Name, Func<string?, EventLogQueryRequest> Build)[]
|
||||
{
|
||||
("no filter, page 1 (oldest first)", t => Req(null, null, null, null, null, null, t)),
|
||||
("last 1 h", t => Req(now.AddHours(-1), null, null, null, null, null, t)),
|
||||
("last 24 h", t => Req(now.AddDays(-1), null, null, null, null, null, t)),
|
||||
("severity=Error, 30 d", t => Req(null, null, null, "Error", null, null, t)),
|
||||
("severity=Error, last 24 h", t => Req(now.AddDays(-1), null, null, "Error", null, null, t)),
|
||||
("event_type=alarm, last 24 h", t => Req(now.AddDays(-1), null, "alarm", null, null, null, t)),
|
||||
("instance=Machine-0042, 30 d", t => Req(null, null, null, null, "Machine-0042", null, t)),
|
||||
("keyword 'timeout' (clamped to 7 d)", t => Req(null, null, null, null, null, "timeout", t)),
|
||||
("keyword with no match (full 7 d window scan)", t => Req(null, null, null, null, null, "no-such-text", t)),
|
||||
("keyword 'Machine-0042', last 24 h", t => Req(now.AddDays(-1), null, null, null, null, "Machine-0042", t)),
|
||||
};
|
||||
|
||||
Console.WriteLine();
|
||||
Console.WriteLine("| query (page size 500) | first ms | warm P50 ms | warm max ms | rows | has more |");
|
||||
Console.WriteLine("|---|---|---|---|---|---|");
|
||||
foreach (var (name, build) in shapes)
|
||||
{
|
||||
EventLogQueryResponse? response = null;
|
||||
var first = Samples.Time(() => response = query.ExecuteQuery(build(null)));
|
||||
var warm = new Samples();
|
||||
for (var i = 0; i < 5; i++) warm.Add(Samples.Time(() => query.ExecuteQuery(build(null))));
|
||||
if (response is { Success: false }) throw new InvalidOperationException($"{name}: {response.ErrorMessage}");
|
||||
Console.WriteLine($"| {name} | {first:F1} | {warm.Percentile(50):F1} | {warm.Max:F1} | {response!.Entries.Count} | {response.HasMore} |");
|
||||
}
|
||||
|
||||
// Paging through: 20 consecutive pages of the last 24 h.
|
||||
string? token = null;
|
||||
var pages = new Samples();
|
||||
for (var i = 0; i < 20; i++)
|
||||
{
|
||||
EventLogQueryResponse? r = null;
|
||||
pages.Add(Samples.Time(() => r = query.ExecuteQuery(Req(now.AddDays(-1), null, null, null, null, null, token))));
|
||||
token = r!.ContinuationToken;
|
||||
if (token is null) break;
|
||||
}
|
||||
Console.WriteLine($"| 20 keyset pages, last 24 h | | {pages.Percentile(50):F1} | {pages.Max:F1} | 500/page | |");
|
||||
}
|
||||
|
||||
private static async Task MeasureWriterUnderQueriesAsync(SiteEventLogger logger, EventLogQueryService query, DateTimeOffset now)
|
||||
{
|
||||
// Offered 200 events/s (≈ 190x the rate that fits 30 days in 1 GB) for 20 s, first idle,
|
||||
// then with a back-to-back worst-case keyword query loop on another thread.
|
||||
Console.WriteLine();
|
||||
foreach (var withQueries in new[] { false, true })
|
||||
{
|
||||
using var cts = new CancellationTokenSource();
|
||||
var queriesRun = 0;
|
||||
var queryLoop = withQueries
|
||||
? Task.Run(() =>
|
||||
{
|
||||
while (!cts.IsCancellationRequested)
|
||||
{
|
||||
query.ExecuteQuery(Req(null, null, null, null, null, "no-such-text", null));
|
||||
queriesRun++;
|
||||
}
|
||||
})
|
||||
: Task.CompletedTask;
|
||||
|
||||
var latency = new Samples();
|
||||
var tasks = new List<Task>();
|
||||
var failedBefore = logger.FailedWriteCount;
|
||||
var sw = Stopwatch.StartNew();
|
||||
var sent = 0;
|
||||
while (sw.Elapsed < TimeSpan.FromSeconds(20))
|
||||
{
|
||||
var due = (int)(sw.Elapsed.TotalSeconds * 200);
|
||||
for (; sent < due; sent++)
|
||||
{
|
||||
var start = Stopwatch.GetTimestamp();
|
||||
var e = RowFor(sent);
|
||||
tasks.Add(logger.LogEventAsync(e.Type, e.Severity, e.Instance, e.Source, e.Message, e.Details)
|
||||
.ContinueWith(_ => latency.Add(Stopwatch.GetElapsedTime(start).TotalMilliseconds), TaskScheduler.Default));
|
||||
}
|
||||
await Task.Delay(5);
|
||||
}
|
||||
await Task.WhenAll(tasks);
|
||||
cts.Cancel();
|
||||
await queryLoop;
|
||||
Console.WriteLine($"writer at 200 ev/s, {(withQueries ? $"with a back-to-back no-match keyword scan loop ({queriesRun} scans)" : "idle")}: " +
|
||||
$"enqueue->durable {latency.Summary()}; failed {logger.FailedWriteCount - failedBefore}");
|
||||
}
|
||||
}
|
||||
|
||||
private static EventLogQueryRequest Req(
|
||||
DateTimeOffset? from, DateTimeOffset? to, string? type, string? severity, string? instance, string? keyword, string? token) =>
|
||||
new("bench", "site-bench", from, to, type, severity, instance, keyword, token, 500, DateTimeOffset.UtcNow);
|
||||
|
||||
// ---- representative event mix --------------------------------------------------------
|
||||
|
||||
internal readonly record struct Row(string Type, string Severity, string? Instance, string Source, string Message, string? Details);
|
||||
|
||||
private static readonly string StackDetails =
|
||||
"System.TimeoutException: The operation has timed out.\n" +
|
||||
string.Concat(Enumerable.Range(0, 8).Select(i =>
|
||||
$" at ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts.ScriptRuntimeContext.Step{i}Async(String path, CancellationToken ct)\n"));
|
||||
|
||||
/// <summary>
|
||||
/// Deterministic mix modelled on the post-WP3.2 event stream (per-run script events off, an
|
||||
/// interval summary instead): alarm transitions dominate, then script errors, connection
|
||||
/// changes, S&F and lifecycle. 2-in-3 rows carry an instance; 1-in-10 carry a details payload.
|
||||
/// </summary>
|
||||
internal static Row RowFor(long i)
|
||||
{
|
||||
var instance = i % 3 == 2 ? null : $"Machine-{i % Instances:D4}";
|
||||
var details = i % 10 == 0 ? StackDetails : null;
|
||||
return (i % 20) switch
|
||||
{
|
||||
< 9 => new Row("alarm", i % 9 == 0 ? "Warning" : "Info", instance, $"AlarmActor:HighTemp{i % 7}",
|
||||
$"Alarm 'HighTemp{i % 7}' on '{instance ?? "site"}' changed to {(i % 2 == 0 ? "Active" : "Normal")}", details),
|
||||
< 14 => new Row("script", i % 5 == 0 ? "Error" : "Warning", instance, $"ScriptActor:Recipe{i % 11}",
|
||||
$"Script 'Recipe{i % 11}' failed: timeout waiting for attribute 'Done' after 30000 ms", details),
|
||||
< 17 => new Row("connection", i % 3 == 0 ? "Warning" : "Info", null, $"DataConnectionActor:PLC-{i % 40:D2}",
|
||||
$"Connection 'PLC-{i % 40:D2}' quality changed to {(i % 2 == 0 ? "Bad" : "Good")}", details),
|
||||
< 19 => new Row("store_and_forward", "Warning", instance, "StoreAndForwardService",
|
||||
$"Message parked after 50 retries to target 'MES-{i % 4}'", details),
|
||||
_ => new Row("instance_lifecycle", "Info", instance, "DeploymentManagerActor",
|
||||
$"Instance '{instance ?? "Machine-0000"}' started", null),
|
||||
};
|
||||
}
|
||||
|
||||
// ---- LocalDb plumbing ----------------------------------------------------------------
|
||||
|
||||
private static ILocalDb Open(string path, out string resolvedPath)
|
||||
{
|
||||
foreach (var suffix in new[] { "", "-wal", "-shm" }) File.Delete(path + suffix);
|
||||
var configuration = new ConfigurationBuilder()
|
||||
.AddInMemoryCollection(new Dictionary<string, string?> { ["LocalDb:Path"] = path })
|
||||
.Build();
|
||||
var provider = new ServiceCollection().AddZbLocalDb(configuration).BuildServiceProvider();
|
||||
resolvedPath = path;
|
||||
Providers[path] = provider;
|
||||
return provider.GetRequiredService<ILocalDb>();
|
||||
}
|
||||
|
||||
private static readonly Dictionary<string, ServiceProvider> Providers = new();
|
||||
|
||||
private static void Cleanup(ILocalDb db, string path)
|
||||
{
|
||||
if (Providers.Remove(path, out var provider)) provider.Dispose();
|
||||
foreach (var suffix in new[] { "", "-wal", "-shm" })
|
||||
{
|
||||
try { File.Delete(path + suffix); } catch { /* best effort */ }
|
||||
}
|
||||
}
|
||||
|
||||
private static SiteEventLogger NewLogger(ILocalDb db, SiteEventLogOptions options) =>
|
||||
new(Options.Create(options), NullLogger<SiteEventLogger>.Instance, db);
|
||||
|
||||
private static long UsedBytes(ILocalDb db)
|
||||
{
|
||||
using var c = db.CreateConnection();
|
||||
long Scalar(string sql) { using var cmd = c.CreateCommand(); cmd.CommandText = sql; return (long)cmd.ExecuteScalar()!; }
|
||||
using (var cp = c.CreateCommand()) { cp.CommandText = "PRAGMA wal_checkpoint(TRUNCATE)"; cp.ExecuteNonQuery(); }
|
||||
return (Scalar("PRAGMA page_count") - Scalar("PRAGMA freelist_count")) * Scalar("PRAGMA page_size");
|
||||
}
|
||||
|
||||
private static long Count(ILocalDb db)
|
||||
{
|
||||
using var c = db.CreateConnection();
|
||||
using var cmd = c.CreateCommand();
|
||||
cmd.CommandText = "SELECT COUNT(*) FROM site_events";
|
||||
return (long)cmd.ExecuteScalar()!;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Bulk-seeds rows with timestamps spread evenly over [from, to] — the logger always stamps
|
||||
/// "now", so history has to be written directly. Same columns and formats as the logger.
|
||||
/// </summary>
|
||||
private static void Seed(ILocalDb db, long rows, DateTimeOffset from, DateTimeOffset to)
|
||||
{
|
||||
using var c = db.CreateConnection();
|
||||
var span = (to - from).Ticks;
|
||||
const int batch = 20_000;
|
||||
for (long offset = 0; offset < rows; offset += batch)
|
||||
{
|
||||
using var tx = c.BeginTransaction();
|
||||
using var cmd = c.CreateCommand();
|
||||
cmd.Transaction = tx;
|
||||
cmd.CommandText = """
|
||||
INSERT INTO site_events (id, timestamp, event_type, severity, instance_id, source, message, details)
|
||||
VALUES ($id, $ts, $type, $sev, $inst, $src, $msg, $det)
|
||||
""";
|
||||
var pId = cmd.Parameters.Add("$id", SqliteType.Text);
|
||||
var pTs = cmd.Parameters.Add("$ts", SqliteType.Text);
|
||||
var pType = cmd.Parameters.Add("$type", SqliteType.Text);
|
||||
var pSev = cmd.Parameters.Add("$sev", SqliteType.Text);
|
||||
var pInst = cmd.Parameters.Add("$inst", SqliteType.Text);
|
||||
var pSrc = cmd.Parameters.Add("$src", SqliteType.Text);
|
||||
var pMsg = cmd.Parameters.Add("$msg", SqliteType.Text);
|
||||
var pDet = cmd.Parameters.Add("$det", SqliteType.Text);
|
||||
cmd.Prepare();
|
||||
var end = Math.Min(rows, offset + batch);
|
||||
for (var i = offset; i < end; i++)
|
||||
{
|
||||
var r = RowFor(i);
|
||||
pId.Value = Guid.NewGuid().ToString("N");
|
||||
pTs.Value = from.AddTicks(span / rows * i).ToUniversalTime().ToString("o");
|
||||
pType.Value = r.Type;
|
||||
pSev.Value = r.Severity;
|
||||
pInst.Value = (object?)r.Instance ?? DBNull.Value;
|
||||
pSrc.Value = r.Source;
|
||||
pMsg.Value = r.Message;
|
||||
pDet.Value = (object?)r.Details ?? DBNull.Value;
|
||||
cmd.ExecuteNonQuery();
|
||||
}
|
||||
tx.Commit();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,236 @@
|
||||
using System.Diagnostics;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Instances;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Templates;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
|
||||
using ZB.MOM.WW.ScadaBridge.TemplateEngine.Flattening;
|
||||
using ZB.MOM.WW.ScadaBridge.TemplateEngine.Validation;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.ResidualBench;
|
||||
|
||||
/// <summary>
|
||||
/// [xc-5] Template flattening and validation complete within acceptable time for large
|
||||
/// templates. Runs the production <see cref="FlatteningService"/>, <see cref="ValidationService"/>
|
||||
/// (the deploy gate, including the Roslyn trust check + compile of every script) and
|
||||
/// <see cref="RevisionHashService"/> in-process over synthetic template graphs of three sizes.
|
||||
/// The configuration-database read that <c>FlatteningPipeline</c> does first is not in frame.
|
||||
/// </summary>
|
||||
internal static class FlattenBench
|
||||
{
|
||||
private sealed record Shape(
|
||||
string Name, int Depth, int AttrsPerLevel, int AlarmsPerLevel, int ScriptsPerLevel,
|
||||
int Modules, int ModuleAttrs, int ModuleAlarms, int ModuleScripts, int SubModuleAttrs);
|
||||
|
||||
private static readonly Shape[] Shapes =
|
||||
[
|
||||
new("typical", Depth: 2, AttrsPerLevel: 40, AlarmsPerLevel: 5, ScriptsPerLevel: 3,
|
||||
Modules: 2, ModuleAttrs: 10, ModuleAlarms: 1, ModuleScripts: 1, SubModuleAttrs: 0),
|
||||
new("large", Depth: 5, AttrsPerLevel: 100, AlarmsPerLevel: 10, ScriptsPerLevel: 6,
|
||||
Modules: 10, ModuleAttrs: 50, ModuleAlarms: 5, ModuleScripts: 3, SubModuleAttrs: 10),
|
||||
new("extra-large", Depth: 8, AttrsPerLevel: 250, AlarmsPerLevel: 25, ScriptsPerLevel: 15,
|
||||
Modules: 20, ModuleAttrs: 100, ModuleAlarms: 10, ModuleScripts: 5, SubModuleAttrs: 20),
|
||||
];
|
||||
|
||||
public static int Run(BenchArgs args)
|
||||
{
|
||||
var instances = args.GetInt("instances", 500);
|
||||
var repeats = args.GetInt("repeats", 20);
|
||||
args.RejectUnknown();
|
||||
|
||||
var flattener = new FlatteningService();
|
||||
var validator = new ValidationService();
|
||||
var hasher = new RevisionHashService();
|
||||
|
||||
// Warm the JIT and Roslyn once on a throwaway graph so the first shape doesn't pay it.
|
||||
{
|
||||
var warm = Build(Shapes[0], salt: "jit-warmup");
|
||||
var w = flattener.Flatten(warm.Instance, warm.Chain, warm.Compositions, warm.ComposedChains, warm.Connections).Value;
|
||||
validator.Validate(w);
|
||||
hasher.ComputeHash(w);
|
||||
}
|
||||
|
||||
Console.WriteLine();
|
||||
Console.WriteLine("| shape | chain depth | flattened attrs / alarms / scripts | flatten P50 ms | validate, cold compile ms | validate, warm cache P50 ms | validate, compile off P50 ms | hash P50 ms | errors |");
|
||||
Console.WriteLine("|---|---|---|---|---|---|---|---|---|");
|
||||
foreach (var shape in Shapes)
|
||||
{
|
||||
var g = Build(shape, salt: shape.Name);
|
||||
FlattenedConfiguration config = null!;
|
||||
var flatten = new Samples();
|
||||
for (var i = 0; i < repeats; i++)
|
||||
flatten.Add(Samples.Time(() => config = Flatten(flattener, g)));
|
||||
|
||||
ScriptCompileVerdictCache.Clear();
|
||||
ValidationResult result = null!;
|
||||
var cold = Samples.Time(() => result = validator.Validate(config));
|
||||
|
||||
var warm = new Samples();
|
||||
for (var i = 0; i < repeats; i++) warm.Add(Samples.Time(() => validator.Validate(config)));
|
||||
var noCompile = new Samples();
|
||||
for (var i = 0; i < repeats; i++) noCompile.Add(Samples.Time(() => validator.Validate(config, validateScriptCompilation: false)));
|
||||
var hash = new Samples();
|
||||
for (var i = 0; i < repeats; i++) hash.Add(Samples.Time(() => hasher.ComputeHash(config)));
|
||||
|
||||
Console.WriteLine($"| {shape.Name} | {shape.Depth} | {config.Attributes.Count:N0} / {config.Alarms.Count} / {config.Scripts.Count} | " +
|
||||
$"{flatten.Percentile(50):F2} | {cold:F0} | {warm.Percentile(50):F2} | {noCompile.Percentile(50):F2} | {hash.Percentile(50):F2} | {result.Errors.Count} |");
|
||||
if (result.Errors.Count > 0)
|
||||
Console.WriteLine($" first errors: {result.SummarizeErrors(3)}");
|
||||
}
|
||||
|
||||
// Bulk-deploy CPU: N instances of one template (scripts shared, so after the first
|
||||
// instance every compile is a verdict-cache hit — the production shape).
|
||||
Console.WriteLine();
|
||||
foreach (var shape in Shapes.Skip(1))
|
||||
{
|
||||
var graphs = Enumerable.Range(0, instances).Select(i => Build(shape, salt: "bulk", instanceIndex: i)).ToList();
|
||||
ScriptCompileVerdictCache.Clear();
|
||||
var per = new Samples();
|
||||
var sw = Stopwatch.StartNew();
|
||||
foreach (var g in graphs)
|
||||
{
|
||||
per.Add(Samples.Time(() =>
|
||||
{
|
||||
var c = Flatten(flattener, g);
|
||||
var v = validator.Validate(c);
|
||||
if (!v.IsValid) throw new InvalidOperationException(v.SummarizeErrors(3));
|
||||
hasher.ComputeHash(c);
|
||||
}));
|
||||
}
|
||||
Console.WriteLine($"{shape.Name}: flatten+validate+hash of {instances} instances sharing one template, serial: " +
|
||||
$"{sw.Elapsed.TotalSeconds:F2} s total; per instance {per.Summary()}");
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
private static FlattenedConfiguration Flatten(FlatteningService flattener, Graph g)
|
||||
{
|
||||
var r = flattener.Flatten(g.Instance, g.Chain, g.Compositions, g.ComposedChains, g.Connections);
|
||||
return r.IsSuccess ? r.Value : throw new InvalidOperationException(r.Error);
|
||||
}
|
||||
|
||||
private sealed record Graph(
|
||||
Instance Instance,
|
||||
IReadOnlyList<Template> Chain,
|
||||
IReadOnlyDictionary<int, IReadOnlyList<TemplateComposition>> Compositions,
|
||||
IReadOnlyDictionary<int, IReadOnlyList<Template>> ComposedChains,
|
||||
IReadOnlyDictionary<int, DataConnection> Connections);
|
||||
|
||||
/// <summary>
|
||||
/// A chain of <c>Depth</c> templates (each level adds attributes/alarms/scripts and
|
||||
/// overrides a fifth of its parent's attribute values), <c>Modules</c> feature modules
|
||||
/// composed on the base, each module composing one sub-module, and an instance that
|
||||
/// overrides 10% of the attributes and binds every data-sourced attribute to one connection.
|
||||
/// </summary>
|
||||
private static Graph Build(Shape s, string salt, int instanceIndex = 0)
|
||||
{
|
||||
var nextId = 1;
|
||||
var connection = new DataConnection("PLC-01", "OpcUa", 1) { Id = 1 };
|
||||
|
||||
Template NewTemplate(string name, int? parent) => new(name) { Id = nextId++, ParentTemplateId = parent };
|
||||
|
||||
void Populate(Template t, string prefix, int attrs, int alarms, int scripts)
|
||||
{
|
||||
for (var a = 0; a < attrs; a++)
|
||||
{
|
||||
t.Attributes.Add(new TemplateAttribute($"{prefix}Attr{a}")
|
||||
{
|
||||
DataType = a % 4 == 0 ? DataType.String : DataType.Double,
|
||||
Value = a % 4 == 0 ? "idle" : "0",
|
||||
DataSourceReference = a % 2 == 0 ? $"ns=2;s={prefix}.Attr{a}" : null,
|
||||
});
|
||||
}
|
||||
for (var al = 0; al < alarms; al++)
|
||||
{
|
||||
t.Alarms.Add(new TemplateAlarm($"{prefix}Alarm{al}")
|
||||
{
|
||||
TriggerType = AlarmTriggerType.RangeViolation,
|
||||
TriggerConfiguration = $"{{\"attributeName\":\"{prefix}Attr{al * 4 % Math.Max(attrs, 1) + 1}\",\"high\":100}}",
|
||||
PriorityLevel = 1 + al % 3,
|
||||
});
|
||||
}
|
||||
for (var sc = 0; sc < scripts; sc++)
|
||||
t.Scripts.Add(new TemplateScript($"{prefix}Script{sc}", ScriptBody($"{salt}:{prefix}:{sc}")) { Id = nextId++ });
|
||||
}
|
||||
|
||||
var chain = new List<Template>();
|
||||
Template? parentT = null;
|
||||
for (var level = 0; level < s.Depth; level++)
|
||||
{
|
||||
var t = NewTemplate($"Level{level}", parentT?.Id);
|
||||
Populate(t, $"L{level}", s.AttrsPerLevel, s.AlarmsPerLevel, s.ScriptsPerLevel);
|
||||
if (parentT is not null)
|
||||
{
|
||||
foreach (var inherited in parentT.Attributes.Where((_, i) => i % 5 == 0))
|
||||
t.Attributes.Add(new TemplateAttribute(inherited.Name) { DataType = inherited.DataType, Value = inherited.DataType == DataType.String ? "derived" : "1" });
|
||||
}
|
||||
chain.Insert(0, t); // most-derived first
|
||||
parentT = t;
|
||||
}
|
||||
|
||||
var baseTemplate = chain[^1];
|
||||
var compositions = new Dictionary<int, IReadOnlyList<TemplateComposition>>();
|
||||
var composedChains = new Dictionary<int, IReadOnlyList<Template>>();
|
||||
var baseModules = new List<TemplateComposition>();
|
||||
for (var m = 0; m < s.Modules; m++)
|
||||
{
|
||||
var module = NewTemplate($"Module{m}", null);
|
||||
Populate(module, "M", s.ModuleAttrs, s.ModuleAlarms, s.ModuleScripts);
|
||||
if (s.SubModuleAttrs > 0)
|
||||
{
|
||||
var sub = NewTemplate($"SubModule{m}", null);
|
||||
Populate(sub, "S", s.SubModuleAttrs, 0, 0);
|
||||
composedChains[sub.Id] = [sub];
|
||||
compositions[module.Id] = [new TemplateComposition("Sensor") { ComposedTemplateId = sub.Id, TemplateId = module.Id }];
|
||||
}
|
||||
composedChains[module.Id] = [module];
|
||||
baseModules.Add(new TemplateComposition($"Unit{m}") { ComposedTemplateId = module.Id, TemplateId = baseTemplate.Id });
|
||||
}
|
||||
if (baseModules.Count > 0) compositions[baseTemplate.Id] = baseModules;
|
||||
|
||||
var instance = new Instance($"Machine-{instanceIndex:D4}") { Id = 1 + instanceIndex, TemplateId = chain[0].Id, SiteId = 1 };
|
||||
var allAttrs = chain.SelectMany(t => t.Attributes).Select(a => a.Name).Distinct().ToList();
|
||||
foreach (var name in allAttrs.Where((_, i) => i % 10 == 0))
|
||||
instance.AttributeOverrides.Add(new InstanceAttributeOverride(name) { OverrideValue = "5" });
|
||||
foreach (var name in chain.SelectMany(t => t.Attributes).Where(a => a.DataSourceReference is not null).Select(a => a.Name).Distinct())
|
||||
instance.ConnectionBindings.Add(new InstanceConnectionBinding(name) { DataConnectionId = connection.Id });
|
||||
|
||||
return new Graph(instance, chain, compositions, composedChains, new Dictionary<int, DataConnection> { [connection.Id] = connection });
|
||||
}
|
||||
|
||||
/// <summary>~35 lines of ordinary script C#; the tag makes each script's code (the compile-cache key) unique.</summary>
|
||||
private static string ScriptBody(string tag) => $$"""
|
||||
// {{tag}}
|
||||
var readings = new List<double>();
|
||||
for (var i = 0; i < 24; i++)
|
||||
{
|
||||
readings.Add(Math.Sin(i / 3.0) * 10.0 + i);
|
||||
}
|
||||
var mean = readings.Average();
|
||||
var variance = readings.Select(r => (r - mean) * (r - mean)).Sum() / readings.Count;
|
||||
var stdDev = Math.Sqrt(variance);
|
||||
var outliers = readings.Where(r => Math.Abs(r - mean) > 2 * stdDev).ToList();
|
||||
var summary = new Dictionary<string, object>
|
||||
{
|
||||
["mean"] = mean,
|
||||
["stdDev"] = stdDev,
|
||||
["outliers"] = outliers.Count,
|
||||
["tag"] = "{{tag}}",
|
||||
};
|
||||
string state;
|
||||
if (outliers.Count == 0)
|
||||
{
|
||||
state = "stable";
|
||||
}
|
||||
else if (outliers.Count < 3)
|
||||
{
|
||||
state = "drifting";
|
||||
}
|
||||
else
|
||||
{
|
||||
state = "unstable";
|
||||
}
|
||||
var text = string.Join(",", summary.Select(kv => $"{kv.Key}={kv.Value}"));
|
||||
return state + ";" + text;
|
||||
""";
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
using System.Runtime;
|
||||
using ZB.MOM.WW.ScadaBridge.ResidualBench;
|
||||
|
||||
// Phase-8 WP-4 residual-scope benchmarks (ScadaBridge#49). See the csproj header.
|
||||
//
|
||||
// dotnet run -c Release --project tests/ZB.MOM.WW.ScadaBridge.ResidualBench -- eventlog [--data-dir DIR] [--rows N]
|
||||
// dotnet run -c Release --project tests/ZB.MOM.WW.ScadaBridge.ResidualBench -- flatten
|
||||
// dotnet run -c Release --project tests/ZB.MOM.WW.ScadaBridge.ResidualBench -- audit --connection-string "..."
|
||||
// dotnet run -c Release --project tests/ZB.MOM.WW.ScadaBridge.ResidualBench -- audit-contention [--writers N] [--seconds S]
|
||||
|
||||
if (args.Length == 0)
|
||||
{
|
||||
Console.Error.WriteLine("usage: ResidualBench <eventlog|flatten|audit|audit-contention> [options]");
|
||||
return 2;
|
||||
}
|
||||
|
||||
var options = BenchArgs.Parse(args.Skip(1));
|
||||
Console.WriteLine($"# ResidualBench {args[0]} — {DateTimeOffset.UtcNow:o}");
|
||||
Console.WriteLine($"# runtime {Environment.Version}, server GC {GCSettings.IsServerGC}, {Environment.ProcessorCount} logical cores");
|
||||
|
||||
return args[0] switch
|
||||
{
|
||||
"eventlog" => await EventLogBench.RunAsync(options),
|
||||
"flatten" => FlattenBench.Run(options),
|
||||
"audit" => await AuditBench.RunAsync(options),
|
||||
"audit-contention" => await AuditContentionBench.RunAsync(options),
|
||||
_ => Unknown(args[0]),
|
||||
};
|
||||
|
||||
static int Unknown(string mode)
|
||||
{
|
||||
Console.Error.WriteLine($"unknown mode '{mode}'");
|
||||
return 2;
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<!--
|
||||
Phase-8 WP-4 residual-scope benchmarks (ScadaBridge#49): the four criteria the
|
||||
target-scale LoadHarness deliberately did not cover.
|
||||
|
||||
eventlog [xc-3] site event log volume vs the 30-day / 1 GB limits
|
||||
flatten [xc-5] template flattening + validation time for large templates
|
||||
audit [xc-4] central AuditLog ingest vs concurrent central reads (MS SQL)
|
||||
|
||||
[xc-8] (UI responsiveness) is measured against the docker rig with Playwright,
|
||||
not here. An Exe rather than an xunit project for the same reason as LoadHarness:
|
||||
the runs take minutes and must not execute under `dotnet test`.
|
||||
|
||||
Results: docs/plans/2026-11-10-wp4-residual-scope-results.md
|
||||
-->
|
||||
<PropertyGroup>
|
||||
<TargetFramework>net10.0</TargetFramework>
|
||||
<OutputType>Exe</OutputType>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
<Nullable>enable</Nullable>
|
||||
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
|
||||
<IsPackable>false</IsPackable>
|
||||
<RootNamespace>ZB.MOM.WW.ScadaBridge.ResidualBench</RootNamespace>
|
||||
<ServerGarbageCollection>true</ServerGarbageCollection>
|
||||
<ConcurrentGarbageCollection>true</ConcurrentGarbageCollection>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Microsoft.Extensions.DependencyInjection" />
|
||||
<PackageReference Include="Microsoft.Extensions.Configuration.Json" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="../../src/ZB.MOM.WW.ScadaBridge.Commons/ZB.MOM.WW.ScadaBridge.Commons.csproj" />
|
||||
<ProjectReference Include="../../src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/ZB.MOM.WW.ScadaBridge.SiteEventLogging.csproj" />
|
||||
<ProjectReference Include="../../src/ZB.MOM.WW.ScadaBridge.TemplateEngine/ZB.MOM.WW.ScadaBridge.TemplateEngine.csproj" />
|
||||
<ProjectReference Include="../../src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
Reference in New Issue
Block a user