chore: metaverse benchmark scenario

.github/workflows/benchmark.yaml

@@ -24,6 +24,7 @@ on:
         type: choice
         options:
           - workflow_dispatch_broadcast
+          - workflow_dispatch_metaversebroadcast
           - workflow_dispatch_memorypack_h2
           - workflow_dispatch_memorypack_h2c
           - workflow_dispatch_memorypack_h3

perf/BenchmarkApp/PerformanceTest.Client/BroadcastScenario.cs

@@ -8,10 +8,7 @@ public class BroadcastScenario : IScenario, IPerTestBroadcastHubReceiver
 {
     IPerfTestService client = default!;
     IPerTestBroadcastHub hubClient = default!;
-    readonly TimeProvider timeProvider = TimeProvider.System;
     PerformanceTestRunningContext context = default!;
-    int connectionId;
-    long begin;
 
     protected virtual int TargetFps => 0; // 0 means maximum speed
 
@@ -25,23 +22,20 @@ public async ValueTask PrepareAsync(GrpcChannel channel)
     // So most times MoveNext won't wait at all, and it may wait occasionally.
     public async ValueTask RunAsync(int connectionId, PerformanceTestRunningContext ctx, CancellationToken cancellationToken)
     {
-        var start = TimeProvider.System.GetTimestamp();
         context = ctx;
-        this.connectionId = connectionId;
-        begin = timeProvider.GetTimestamp();
         await hubClient.JoinGroupAsync();
-        
+
         // Only the first client triggers broadcast after warmup completes
         if (connectionId == 0)
         {
             // Wait for warmup to complete
             await ctx.WaitForReadyAsync();
-            
+
             // Use exact duration to match client-side metrics collection period
             var duration = TimeSpan.FromSeconds(ctx.DurationSeconds);
             _ = client.BroadcastAsync(duration, TargetFps);
         }
-        
+
         while (!cancellationToken.IsCancellationRequested)
         {
             await Task.Delay(100, cancellationToken).ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
@@ -52,6 +46,7 @@ public async Task CompleteAsync()
     {
         if (hubClient is not null)
         {
+            await hubClient.LeaveGroupAsync();
             await hubClient.DisposeAsync();
         }
     }
@@ -60,8 +55,6 @@ public void OnMessage(BroadcastPositionMessage message)
     {
         // Collect only Count
         context.Increment();
-        //context.LatencyThrottled(connectionId, timeProvider.GetElapsedTime(begin), 100); // avoid OOM
-        //begin = timeProvider.GetTimestamp();
     }
 }
 

perf/BenchmarkApp/PerformanceTest.Client/MetaverseBroadcastScenario.cs

@@ -0,0 +1,101 @@
+using Grpc.Net.Client;
+using MagicOnion.Client;
+using PerformanceTest.Shared;
+
+namespace PerformanceTest.Client;
+
+public class MetaverseBroadcastScenario : IScenario, IMetaverseBroadcastHubReceiver
+{
+    const int clientFps = 15;
+    static Random random = new(76); // Fixed seed for reproducibility
+
+    IMetaverseBroadcastHub hubClient = default!;
+    PerformanceTestRunningContext context = default!;
+    int connectionId;
+
+    protected virtual int TargetFps => 30; // Default 30 FPS
+
+    public async ValueTask PrepareAsync(GrpcChannel channel)
+    {
+        hubClient = await StreamingHubClient.ConnectAsync<IMetaverseBroadcastHub, IMetaverseBroadcastHubReceiver>(channel, this);
+    }
+
+    public async ValueTask RunAsync(int connectionId, PerformanceTestRunningContext ctx, CancellationToken cancellationToken)
+    {
+        context = ctx;
+        this.connectionId = connectionId;
+        await hubClient.JoinAsync(TargetFps);
+
+        // Wait for warmup to complete
+        await ctx.WaitForReadyAsync();
+
+        // jitter to avoid all clients sending updates at the same time
+        await Task.Delay(random.Next(0, 500), cancellationToken);
+
+        if (connectionId == 0)
+        {
+            await hubClient.StartBroadcast(TargetFps);
+        }
+
+        // Simulate client position updates periodically (fixed to 15 FPS for all scenarios to isolate broadcast performance)
+        var interval = TimeSpan.FromMilliseconds(1000.0 / clientFps);
+        using var updateTimer = new PeriodicTimer(interval);
+
+        try
+        {
+            while (await updateTimer.WaitForNextTickAsync(cancellationToken))
+            {
+                // Send random position update
+                var position = new BroadcastPositionMessage(
+                    connectionId,
+                    new Vector3(
+                        (float)(random.NextDouble() * 1000),
+                        (float)(random.NextDouble() * 100),
+                        (float)(random.NextDouble() * 1000)
+                    )
+                );
+
+                await hubClient.UpdatePositionAsync(position);
+            }
+        }
+        catch (OperationCanceledException)
+        {
+            // Expected on cancellation
+        }
+
+        if (connectionId == 0)
+        {
+            await hubClient.StopBroadcast();
+        }
+    }
+
+    public async Task CompleteAsync()
+    {
+        if (hubClient is not null)
+        {
+            await hubClient.LeaveAsync();
+            await hubClient.DisposeAsync();
+        }
+    }
+
+    public void OnBroadcastAllPositions(AllClientsPositionMessage message)
+    {
+        // Collect metrics for each broadcast received
+        context.Increment();
+    }
+}
+
+public class MetaverseBroadcast15FpsScenario : MetaverseBroadcastScenario
+{
+    protected override int TargetFps => 15;
+}
+
+public class MetaverseBroadcast30FpsScenario : MetaverseBroadcastScenario
+{
+    protected override int TargetFps => 30;
+}
+
+public class MetaverseBroadcast60FpsScenario : MetaverseBroadcastScenario
+{
+    protected override int TargetFps => 60;
+}

perf/BenchmarkApp/PerformanceTest.Client/PerformanceTestRunningContext.cs

@@ -15,15 +15,15 @@ public class PerformanceTestRunningContext
     public TimeSpan Timeout { get; }
     public int DurationSeconds { get; }
 
-    public PerformanceTestRunningContext(int connectionCount, (int WarmupSec, int RunSec) serverTimeout, DatadogMetricsRecorder recoder, ScenarioType scenario)
+    public PerformanceTestRunningContext(int connectionCount, (int WarmupSec, int RunSec) serverTimeout, DatadogMetricsRecorder recoder, ScenarioType scenario, TimeProvider timeProvider)
     {
         Timeout = TimeSpan.FromSeconds(serverTimeout.WarmupSec + serverTimeout.RunSec + 3); // add some sec for safely complete serverstreaming
         DurationSeconds = serverTimeout.RunSec;
         stopwatch = new Stopwatch();
         latencyPerConnection = new(connectionCount);
         errorsPerConnection = 0;
         locks = new(connectionCount);
-        profileService = new ProfileService(TimeProvider.System, recoder, scenario);
+        profileService = new ProfileService(timeProvider, recoder, scenario);
 
         for (var i = 0; i < connectionCount; i++)
         {

perf/BenchmarkApp/PerformanceTest.Client/Program.cs

@@ -90,7 +90,7 @@ async Task Main(
 
         for (var i = 1; i <= rounds; i++)
         {
-            WriteLog($"Round: {i}");
+            WriteLog($"Scenario {scenario}, Round: {i}");
             if (!resultsByScenario.TryGetValue(scenario2, out var results))
             {
                 results = new List<PerformanceResult>(10000);
@@ -99,9 +99,11 @@ async Task Main(
             var result = await RunScenarioAsync(scenario2, config, config.ChannelList, controlServiceClient, datadog);
             results.Add(result);
 
-            WriteLog($"Interval 1s for next scenario...");
+            WriteLog($"Interval 1s for next run/scenario...");
             await Task.Delay(TimeSpan.FromSeconds(1)).ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing); // interval
         }
+
+        await controlServiceClient.NotifyCompleteScenarioAsync();
     }
     WriteLog($"All scenario complete");
 
@@ -200,14 +202,16 @@ async Task<PerformanceResult> RunScenarioAsync(ScenarioType scenario, ScenarioCo
         ScenarioType.StreamingHubLargePayload32K => () => new StreamingHubLargePayload32KScenario(),
         ScenarioType.StreamingHubLargePayload64K => () => new StreamingHubLargePayload64KScenario(),
         ScenarioType.ServerStreaming => () => new ServerStreamingScenario(),
-        ScenarioType.Broadcast => () => new BroadcastScenario(),
         ScenarioType.Broadcast60Fps => () => new Broadcast60FpsScenario(),
         ScenarioType.Broadcast30Fps => () => new Broadcast30FpsScenario(),
         ScenarioType.Broadcast15Fps => () => new Broadcast15FpsScenario(),
+        ScenarioType.MetaverseBroadcast60Fps => () => new MetaverseBroadcast60FpsScenario(),
+        ScenarioType.MetaverseBroadcast30Fps => () => new MetaverseBroadcast30FpsScenario(),
+        ScenarioType.MetaverseBroadcast15Fps => () => new MetaverseBroadcast15FpsScenario(),
         _ => throw new Exception($"Unknown Scenario: {scenario}"),
     };
 
-    var ctx = new PerformanceTestRunningContext(connectionCount: config.Channels, serverTimeout: (config.Warmup, config.Duration), datadog, scenario);
+    var ctx = new PerformanceTestRunningContext(connectionCount: config.Channels, serverTimeout: (config.Warmup, config.Duration), datadog, scenario, config.TimeProvider);
     using var cts = new CancellationTokenSource();
     var cleanIndex = 0;
     var threadBefore = ThreadPool.ThreadCount;
@@ -488,12 +492,12 @@ private static IReadOnlyList<PerformanceResult> RemoveOutlinerByIQR(IReadOnlyLis
         var mean = data.Select(x => x.Latency.Mean).OrderBy(x => x).ToArray();
 
         // get outliner for rps
-        var lowerBoundRps = OutlinerHelper.GetLowerBound(rps);
-        var upperBoundRps = OutlinerHelper.GetUpperBound(rps);
+        var lowerBoundRps = OutlierIqr.GetLowerBound(rps);
+        var upperBoundRps = OutlierIqr.GetUpperBound(rps);
 
         // get outliner for mean
-        var lowerBoundMean = OutlinerHelper.GetLowerBound(mean);
-        var upperBoundMean = OutlinerHelper.GetUpperBound(mean);
+        var lowerBoundMean = OutlierIqr.GetLowerBound(mean);
+        var upperBoundMean = OutlierIqr.GetUpperBound(mean);
 
         // compute tuple in range
         var filteredData = data
@@ -514,6 +518,7 @@ public class ScenarioConfiguration
     public int Streams { get; }
     public int Channels { get; }
     public bool Verbose { get; }
+    public TimeProvider TimeProvider { get; } = TimeProvider.System;
 
     private readonly bool clientAuth;
     private bool useHttp3;