TS-38029 Rework Commons

Author: snowbagoly

Commander.Server/ProfilerTestController.cs

@@ -10,11 +10,13 @@ namespace Cqse.Teamscale.Profiler.Commander.Server
     public class ProfilerTestController : ControllerBase
     {
         private readonly ProfilerIpc profilerIpc;
+        private readonly ProfilerTestControllerState state;
         private readonly ILogger logger;
 
-        public ProfilerTestController(ProfilerIpc profilerIpc, ILogger<ProfilerTestController> logger)
+        public ProfilerTestController(ProfilerIpc profilerIpc, ProfilerTestControllerState state, ILogger<ProfilerTestController> logger)
         {
             this.profilerIpc = profilerIpc;
+            this.state = state;
             this.logger = logger;
         }
 
@@ -26,7 +28,7 @@ public string GetCurrent()
 
         public long GetStart()
         {
-            return profilerIpc.TestStartMS;
+            return state.TestStartMS;
         }
 
         [HttpPost("start/{testName}")]
@@ -38,7 +40,7 @@ public HttpStatusCode StartTest(string testName)
             }
 
             logger.LogInformation("Starting test: {}", testName);
-            profilerIpc.TestStartMS = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
+            state.TestStartMS = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
             profilerIpc.StartTest(HttpUtility.UrlDecode(testName));
             return HttpStatusCode.NoContent;
         }

Commander.Server/ProfilerTestControllerState.cs

@@ -0,0 +1,10 @@
+namespace Cqse.Teamscale.Profiler.Commander.Server
+{
+    /// <summary>
+    /// As the controller is stateless, we store infos about the current test run in this state class.
+    /// </summary>
+    public class ProfilerTestControllerState
+    {
+        public long TestStartMS { get; set; } = 0;
+    }
+}

Commander.Server/Program.cs

@@ -1,8 +1,10 @@
 using Cqse.Teamscale.Profiler.Commons.Ipc;
+using Cqse.Teamscale.Profiler.Commander.Server;
 
 var builder = WebApplication.CreateBuilder(new WebApplicationOptions());
 // explicitly initialize the Ipc, otherwise it might be created upon 1st request and we miss the start event in the profiler
-builder.Services.AddSingleton<ProfilerIpc>(new ProfilerIpc(new IpcConfig()));
+builder.Services.AddSingleton(new ProfilerIpc(new IpcConfig()));
+builder.Services.AddSingleton(new ProfilerTestControllerState());
 builder.Services.AddControllers();
 
 var app = builder.Build();

Commons/Ipc/ProfilerClient.cs

@@ -0,0 +1,16 @@
+using NetMQ.Sockets;
+
+namespace Cqse.Teamscale.Profiler.Commons.Ipc
+{
+    internal class ProfilerClient
+    {
+        ProfilerClient(string clientAddress, RequestSocket socket)
+        {
+            ClientAddress = clientAddress;
+            Socket = socket;
+        }
+
+        public string ClientAddress { get; }
+        public RequestSocket Socket { get; }
+    }
+}

Commons/Ipc/ProfilerIpc.cs

@@ -15,8 +15,6 @@ public class ProfilerIpc : IDisposable
         /// </summary>
         public string TestName { get; private set; } = String.Empty;
 
-        public long TestStartMS { get; set; } = 0;
-
         public IpcConfig Config { get; }
 
         public ProfilerIpc(IpcConfig config)

Commons/Ipc/ZmqIpcServer.cs

@@ -16,7 +16,7 @@ public class ZmqIpcServer : IDisposable
         private NetMQPoller? poller;
         private ResponseSocket? responseSocket;
 
-        private Dictionary<int, Tuple<string, RequestSocket>> clients = new Dictionary<int, Tuple<string, RequestSocket>>();
+        private Dictionary<int, ProfilerClient> pidToClient = new Dictionary<int, ProfilerClient>();
 
         private static readonly Logger logger = LogManager.GetCurrentClassLogger();
 
@@ -58,24 +58,24 @@ protected void StartRequestHandler()
             poller.RunAsync("Profiler IPC", true);
         }
 
-        private  void RegisterClient(string message)
+        private void RegisterClient(string message)
         {
             int pid = Int32.Parse(message.Split(':')[1]);
-            lock (clients)
+            lock (pidToClient)
             {
                 string clientAddress;
-                if (clients.ContainsKey(pid))
+                if (pidToClient.ContainsKey(pid))
                 {
-                    clientAddress = clients[pid].Item1;
+                    clientAddress = pidToClient[pid].ClientAddress;
                     responseSocket.SendFrame(clientAddress);
                     return;
                 }
-                portOffset++;
                 RequestSocket clientRequestSocket = new RequestSocket();
                 clientAddress = config.RequestSocket + ":" + ((config.StartPortNumber + portOffset) % 65535);
+                portOffset++;
                 clientRequestSocket.Connect(clientAddress);
 
-                clients.Add(pid, Tuple.Create(clientAddress, clientRequestSocket));
+                pidToClient.Add(pid, new ProfilerClient(clientAddress, clientRequestSocket));
                 responseSocket.SendFrame(clientAddress);
                 logger.Info($"Registered profiler on address {clientAddress}");
             }
@@ -84,34 +84,33 @@ private  void RegisterClient(string message)
         /// <summary>
         /// Sends the given test event to all connected profiler instances.
         /// </summary>
-        /// <param name="testEvent"></param>
         public void SendTestEvent(string testEvent)
         {
             HashSet<int> clientsToRemove = new HashSet<int>();
-            System.Threading.Tasks.Parallel.ForEach(clients, entry =>
+            System.Threading.Tasks.Parallel.ForEach(pidToClient, entry =>
             {
-                entry.Value.Item2.SendFrame(Encoding.UTF8.GetBytes(testEvent));
-                if (entry.Value.Item2.TryReceiveFrameString(TimeSpan.FromSeconds(3.0), out string? response))
+                entry.Value.Socket.SendFrame(Encoding.UTF8.GetBytes(testEvent));
+                if (entry.Value.Socket.TryReceiveFrameString(TimeSpan.FromSeconds(3.0), out string? response))
                 {
-                    logger.Info($"Got Response from {entry.Value.Item1}: {response}");
+                    logger.Info($"Got Response from {entry.Value.ClientAddress}: {response}");
                 } else
                 {
                     lock(clientsToRemove)
                     {
                         clientsToRemove.Add(entry.Key);
                     }
-                    logger.Error($"Got no response from Profiler with Socket {entry.Key}");
+                    logger.Error($"Got no response from Profiler with PID {entry.Key} with address {entry.Value.ClientAddress}, removing from clients");
                 }
             });
-            lock (clients)
+            lock (pidToClient)
             {
                 foreach (var client in clientsToRemove)
                 {
-                    if (!clients.ContainsKey(client)) {
+                    if (!pidToClient.ContainsKey(client)) {
                         continue;
                     }
-                    clients[client].Item2.Close();
-                    clients.Remove(client);
+                    pidToClient[client].Socket.Close();
+                    pidToClient.Remove(client);
                 }
             }
         }
@@ -120,9 +119,9 @@ public void Dispose()
         {
             this.poller?.Dispose();
             this.responseSocket?.Dispose();
-            foreach (var client in clients)
+            foreach (var client in pidToClient)
             {
-                client.Value.Item2.Dispose();
+                client.Value.Socket.Dispose();
             }
             NetMQConfig.Cleanup(false);
         }