deadlock

Tests/Distribution/NodeFanoutSweep/Client/ControlResource.cs

@@ -0,0 +1,15 @@
+using Esiur.Resource;
+
+/// <summary>
+/// Server-side telemetry resource the sweep orchestrator attaches to (sys/control). The server
+/// updates these once per second; the updates propagate to the orchestrator so it can attribute
+/// fan-out saturation to the server (CPU across all cores, can exceed 100%) and confirm the
+/// connected-subscriber count. Exported as fields so the runtime generates change-notifying
+/// properties (CpuPercent, ConnectedClients).
+/// </summary>
+[Resource]
+public partial class ControlResource : Resource
+{
+    [Export] public double cpuPercent;
+    [Export] public int connectedClients;
+}

Tests/Distribution/NodeFanoutSweep/Client/Esiur.Tests.NodeFanoutSweep.Client.csproj

@@ -8,7 +8,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <ProjectReference Include="..\..\..\..\Libraries\Esiur\Esiur.csproj" />
+    <ProjectReference Include="..\..\..\..\Libraries\Esiur\Esiur.csproj" OutputItemType="Analyzer" />
   </ItemGroup>
 
 </Project>

Tests/Distribution/NodeFanoutSweep/Client/Program.cs

@@ -1 +1,77 @@
-Console.WriteLine("Hello, World!");
+// ============================================================
+// Scalability Extension: Fan-Out — SERVER NODE
+// Hosts M sensor resources and emits Value updates at a fixed interval (the fan-out source). Also
+// hosts sys/control, updated once per second with the server process CPU (% across all cores) and
+// the live subscriber count, which the sweep orchestrator reads to characterise saturation.
+// Anonymous (None-mode) access so subscribers connect without credentials.
+//
+// Usage: dotnet run -- --port 10900 --resources 100 --emit-interval-ms 50
+// (Run the orchestrator from the sibling "Server" project against this host:port.)
+// ============================================================
+
+using Esiur.Protocol;
+using Esiur.Resource;
+using Esiur.Stores;
+using System.Diagnostics;
+
+var port           = int.Parse(GetArg(args, "--port", "10900"));
+var resources      = int.Parse(GetArg(args, "--resources", "100"));
+var emitIntervalMs = int.Parse(GetArg(args, "--emit-interval-ms", "50"));
+
+Console.WriteLine($"[Server] resources={resources} emit-interval={emitIntervalMs}ms port={port} cores={Environment.ProcessorCount}");
+
+var wh = new Warehouse();
+await wh.Put("sys", new MemoryStore());
+var server = await wh.Put("sys/server", new EpServer { Port = (ushort)port, AllowUnauthorizedAccess = true });
+
+var sensors = new SensorResource[resources];
+for (var i = 0; i < resources; i++) { sensors[i] = new SensorResource { SensorId = i }; await wh.Put($"sys/sensor_{i}", sensors[i]); }
+
+var control = new ControlResource();
+await wh.Put("sys/control", control);
+
+await wh.Open();
+Console.WriteLine($"[Server] Listening on port {port} with {resources} sensors + sys/control. Press Ctrl+C to stop.");
+
+// Emit loop: drives property-change notifications to every attached subscriber.
+var sw = Stopwatch.StartNew();
+_ = Task.Run(async () =>
+{
+    while (true)
+    {
+        await Task.Delay(emitIntervalMs);
+        var value = sw.Elapsed.TotalSeconds;
+        foreach (var s in sensors) s.Value = value;
+    }
+});
+
+// Telemetry loop: publish server CPU (% across all cores) and subscriber count once per second.
+_ = Task.Run(async () =>
+{
+    var proc = Process.GetCurrentProcess();
+    var prevCpu = proc.TotalProcessorTime;
+    var prevWall = DateTime.UtcNow;
+    while (true)
+    {
+        await Task.Delay(1000);
+        proc.Refresh();
+        var nowCpu = proc.TotalProcessorTime;
+        var nowWall = DateTime.UtcNow;
+        var wallMs = (nowWall - prevWall).TotalMilliseconds;
+        control.CpuPercent = wallMs > 0 ? (nowCpu - prevCpu).TotalMilliseconds / wallMs * 100.0 : 0;
+        control.ConnectedClients = server.Connections.Count;
+        prevCpu = nowCpu;
+        prevWall = nowWall;
+    }
+});
+
+var stop = new TaskCompletionSource();
+Console.CancelKeyPress += (_, e) => { e.Cancel = true; stop.TrySetResult(); };
+await stop.Task;
+await wh.Close();
+
+static string GetArg(string[] args, string key, string def)
+{
+    var i = Array.IndexOf(args, key);
+    return (i >= 0 && i + 1 < args.Length) ? args[i + 1] : def;
+}

Tests/Distribution/NodeFanoutSweep/Client/SensorResource.cs

@@ -0,0 +1,14 @@
+using Esiur.Resource;
+
+/// <summary>
+/// Observable sensor resource. Setting <c>value</c> raises a property-change notification that the
+/// Esiur runtime propagates to every attached subscriber — the fan-out path under measurement.
+/// (The generated property is named <c>Value</c>; subscribers filter on that name.)
+/// </summary>
+[Resource]
+public partial class SensorResource : Resource
+{
+    public int SensorId { get; set; }
+
+    [Export] public double value;
+}

Tests/Distribution/NodeFanoutSweep/Server/Program.cs

@@ -67,11 +67,17 @@ Console.WriteLine($"[Orchestrator] N values: {string.Join(",", nValues)}");
 // Attach to the server's control resource once.
 // ----------------------------------------------------------------
 var controlWh = new Warehouse();
-dynamic? control = null;
+EpResource? control = null;
+byte cpuIdx = 255, clientsIdx = 255;
 try
 {
     var controlConn = await controlWh.Get<EpConnection>($"ep://{host}:{port}");
-    control = await controlConn.Get("sys/control");
+    control = (EpResource)await controlConn.Get("sys/control");
+    // Resolve property indices by name (EpResource exposes values by index, not dynamic member).
+    var props = control.Instance.Definition.Properties;
+    cpuIdx = (byte)Array.FindIndex(props, p => p.Name == "CpuPercent");
+    clientsIdx = (byte)Array.FindIndex(props, p => p.Name == "ConnectedClients");
+    Console.WriteLine($"[Orchestrator] sys/control attached (CpuPercent=idx {cpuIdx}, ConnectedClients=idx {clientsIdx}).");
 }
 catch (Exception ex)
 {
@@ -141,19 +147,38 @@ foreach (int n in nValues)
         Console.WriteLine($"[Orchestrator] Measurement window {windowSec}s...");
         var cpuSamples = new List<double>();
         var connSamples = new List<int>();
+        var clientCpuSamples = new List<double>();
+        var clientProc = Process.GetCurrentProcess();
+        var prevClientCpu = clientProc.TotalProcessorTime;
+        var prevClientWall = DateTime.UtcNow;
         var winSw = Stopwatch.StartNew();
         while (winSw.Elapsed.TotalSeconds < windowSec)
         {
             await Task.Delay(1000);
-            if (control != null)
+
+            // Server CPU + subscriber count via the control resource (read by property index;
+            // values arrive as variable-width numerics, hence Convert.*).
+            if (control != null && cpuIdx != 255)
             {
                 try
                 {
-                    cpuSamples.Add((double)control.CpuPercent);
-                    connSamples.Add((int)control.ConnectedClients);
+                    if (control.TryGetPropertyValue(cpuIdx, out var cpuVal) && cpuVal != null)
+                        cpuSamples.Add(Convert.ToDouble(cpuVal));
+                    if (control.TryGetPropertyValue(clientsIdx, out var cliVal) && cliVal != null)
+                        connSamples.Add(Convert.ToInt32(cliVal));
                 }
-                catch { /* control resource may not have current value yet */ }
+                catch { /* control resource may not have a current value yet */ }
             }
+
+            // This harness's own CPU (% across all cores). Recorded so saturation can be attributed
+            // to the server rather than to the single subscriber process driving N connections.
+            clientProc.Refresh();
+            var nowClientCpu = clientProc.TotalProcessorTime;
+            var nowClientWall = DateTime.UtcNow;
+            var wallMs = (nowClientWall - prevClientWall).TotalMilliseconds;
+            if (wallMs > 0) clientCpuSamples.Add((nowClientCpu - prevClientCpu).TotalMilliseconds / wallMs * 100.0);
+            prevClientCpu = nowClientCpu;
+            prevClientWall = nowClientWall;
         }
 
         double elapsedSec = winSw.Elapsed.TotalSeconds;
@@ -176,12 +201,15 @@ foreach (int n in nValues)
         double aggregate = perSubRates.Sum();
         double avgServerCpu = cpuSamples.Count > 0 ? cpuSamples.Average() : double.NaN;
         double peakServerCpu = cpuSamples.Count > 0 ? cpuSamples.Max() : double.NaN;
+        double avgClientCpu = clientCpuSamples.Count > 0 ? clientCpuSamples.Average() : double.NaN;
+        double peakClientCpu = clientCpuSamples.Count > 0 ? clientCpuSamples.Max() : double.NaN;
 
         Console.WriteLine($"[Orchestrator] N={n} rep={rep + 1}: "
                         + $"mean_per_sub={meanPerSub:F1}/s "
                         + $"aggregate={aggregate:F0}/s "
                         + $"late={totalLate} "
-                        + $"server_cpu_avg={avgServerCpu:F1}% peak={peakServerCpu:F1}%");
+                        + $"server_cpu_avg={avgServerCpu:F1}%/peak={peakServerCpu:F1}% "
+                        + $"client_cpu_avg={avgClientCpu:F1}%/peak={peakClientCpu:F1}%");
 
         perRepResults.Add(new RepResult
         {
@@ -195,6 +223,8 @@ foreach (int n in nValues)
             LateDeliveries = totalLate,
             ServerCpuAvg = avgServerCpu,
             ServerCpuPeak = peakServerCpu,
+            ClientCpuAvg = avgClientCpu,
+            ClientCpuPeak = peakClientCpu,
         });
 
         // ---------- teardown ----------
@@ -237,6 +267,8 @@ foreach (int n in nValues)
             TotalLate = perRepResults.Sum(r => r.LateDeliveries),
             MeanServerCpuAvg = perRepResults.Average(r => r.ServerCpuAvg),
             MeanServerCpuPeak = perRepResults.Average(r => r.ServerCpuPeak),
+            MeanClientCpuAvg = perRepResults.Average(r => r.ClientCpuAvg),
+            MeanClientCpuPeak = perRepResults.Average(r => r.ClientCpuPeak),
         });
     }
 }
@@ -246,12 +278,13 @@ foreach (int n in nValues)
 // ----------------------------------------------------------------
 var sb = new System.Text.StringBuilder();
 sb.AppendLine("n,replications,mean_per_sub_rate,ci95_halfwidth,mean_aggregate," +
-              "total_late,mean_server_cpu_avg,mean_server_cpu_peak");
+              "total_late,mean_server_cpu_avg,mean_server_cpu_peak,mean_client_cpu_avg,mean_client_cpu_peak");
 foreach (var r in allResults)
 {
     sb.AppendLine(string.Create(CultureInfo.InvariantCulture,
         $"{r.N},{r.Replications},{r.MeanPerSubRate:F2},{r.Ci95HalfWidth:F2}," +
-        $"{r.MeanAggregate:F1},{r.TotalLate},{r.MeanServerCpuAvg:F2},{r.MeanServerCpuPeak:F2}"));
+        $"{r.MeanAggregate:F1},{r.TotalLate},{r.MeanServerCpuAvg:F2},{r.MeanServerCpuPeak:F2}," +
+        $"{r.MeanClientCpuAvg:F2},{r.MeanClientCpuPeak:F2}"));
 }
 await File.WriteAllTextAsync(outputCsv, sb.ToString());
 Console.WriteLine($"\n[Orchestrator] Results written to {outputCsv}");
@@ -375,6 +408,8 @@ record RepResult
     public long LateDeliveries;
     public double ServerCpuAvg;
     public double ServerCpuPeak;
+    public double ClientCpuAvg;
+    public double ClientCpuPeak;
 }
 
 record SweepResult
@@ -387,4 +422,6 @@ record SweepResult
     public long TotalLate;
     public double MeanServerCpuAvg;
     public double MeanServerCpuPeak;
+    public double MeanClientCpuAvg;
+    public double MeanClientCpuPeak;
 }