diff --git a/RemoteExec.Client/LoadBalancingStrategy.cs b/RemoteExec.Client/LoadBalancingStrategy.cs index a744a07..f0392ac 100644 --- a/RemoteExec.Client/LoadBalancingStrategy.cs +++ b/RemoteExec.Client/LoadBalancingStrategy.cs @@ -2,9 +2,6 @@ public enum LoadBalancingStrategy { - RoundRobin, - Random, - LeastConnections, - LeastActiveTasks, - ResourceAware + ResourceAware, + LeastBacklog } \ No newline at end of file diff --git a/RemoteExec.Client/RemoteExecutor.cs b/RemoteExec.Client/RemoteExecutor.cs index be44163..e799a64 100644 --- a/RemoteExec.Client/RemoteExecutor.cs +++ b/RemoteExec.Client/RemoteExecutor.cs @@ -1,92 +1,135 @@ using Microsoft.AspNetCore.SignalR.Client; +using Microsoft.Extensions.Logging; using RemoteExec.Shared; using System.Collections.Concurrent; -using System.Diagnostics.CodeAnalysis; +using System.Diagnostics; using System.Reflection; using System.Text.Json; using System.Threading.Channels; namespace RemoteExec.Client; -public class RemoteExecutor +public class RemoteExecutor : IDisposable { - private readonly List connections = []; - private readonly ConcurrentDictionary serverMetrics = new(); - private int _currentConnectionIndex = 0; - private readonly Lock @lock = new Lock(); + private readonly List servers = []; + private readonly BlockingCollection globalQueue = []; + private readonly ConcurrentDictionary> pendingResults = new(); + private CancellationTokenSource distributorCts = new(); + private Task? distributorTask; private readonly LoadBalancingStrategy loadBalancingStrategy; + private bool disposedValue; + public event EventHandler? MetricsUpdated; - public RemoteExecutor(string url) : this([url], LoadBalancingStrategy.RoundRobin) + public RemoteExecutor(string url) : this([url], LoadBalancingStrategy.ResourceAware) { } - public RemoteExecutor(string[] urls, LoadBalancingStrategy loadBalancingStrategy = LoadBalancingStrategy.RoundRobin) + public RemoteExecutor(string[] urls, LoadBalancingStrategy loadBalancingStrategy = LoadBalancingStrategy.ResourceAware) { this.loadBalancingStrategy = loadBalancingStrategy; foreach (string url in urls) { + Uri baseUri = new(url); + Uri signalRUri = new(baseUri, "/remote"); + HubConnection connection = new HubConnectionBuilder() - .WithUrl(url) + .WithUrl(signalRUri) .WithAutomaticReconnect() + .ConfigureLogging(logging => + { + _ = logging.AddProvider(new RemoteExecLoggerProvider()); + }) .Build(); - connections.Add(connection); + HttpClient httpClient = new() + { + BaseAddress = baseUri + }; + + ServerConnection serverConnection = new ServerConnection(connection, httpClient); + servers.Add(serverConnection); } } public async Task StartAsync(CancellationToken cancellationToken = default) { + distributorCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + List startTasks = []; - foreach (HubConnection connection in connections) + foreach (ServerConnection server in servers) { - _ = connection.On("MetricsUpdated", metrics => + _ = server.Connection.On("MetricsUpdated", metrics => { - serverMetrics[connection] = metrics; - MetricsUpdated?.Invoke(this, new ServerMetricsUpdatedEventArgs(connection, metrics)); + server.Metrics = metrics; + MetricsUpdated?.Invoke(this, new ServerMetricsUpdatedEventArgs(server.Connection, metrics)); }); - _ = connection.On($"RequestAssembly", async (string assemblyName, Guid requestId) => + _ = server.Connection.On("TaskResult", (taskId, result) => + { + if (pendingResults.TryRemove(taskId, out TaskCompletionSource? tcs)) + { + tcs.SetResult(result); + } + }); + + _ = server.Connection.On($"RequestAssembly", async (string assemblyName, Guid requestId) => { Assembly? assembly = AppDomain.CurrentDomain.GetAssemblies().FirstOrDefault(a => a.GetName().FullName == assemblyName) ?? Assembly.Load(new AssemblyName(assemblyName)); byte[] dllBytes = await File.ReadAllBytesAsync(assembly.Location!); - Channel channel = Channel.CreateUnbounded(); - - foreach (byte b in dllBytes) - { - await channel.Writer.WriteAsync(b); - } - - channel.Writer.Complete(); - - await connection.InvokeAsync("ProvideAssembly", requestId, channel.Reader); + ByteArrayContent content = new(dllBytes); + content.Headers.ContentType = new System.Net.Http.Headers.MediaTypeHeaderValue("application/octet-stream"); + _ = await server.HttpClient.PostAsync($"/provide-assembly?requestId={requestId}", content); }); - startTasks.Add(connection.StartAsync(cancellationToken) + startTasks.Add(server.Connection.StartAsync(cancellationToken) .ContinueWith(async (task, state) => { - HubConnection conn = (HubConnection)state!; - serverMetrics[conn] = await conn.InvokeAsync("GetMetrics", cancellationToken); - MetricsUpdated?.Invoke(this, new ServerMetricsUpdatedEventArgs(conn, serverMetrics[conn])); - }, connection, TaskScheduler.Default).Unwrap()); + ServerConnection conn = (ServerConnection)state!; + conn.Metrics = await conn.Connection.InvokeAsync("GetMetrics", cancellationToken); + MetricsUpdated?.Invoke(this, new ServerMetricsUpdatedEventArgs(conn.Connection, conn.Metrics)); + + await conn.Connection.SendAsync("StartTaskStream", conn.TaskChannel.Reader, cancellationToken); + }, server, TaskScheduler.Default).Unwrap()); } await Task.WhenAll(startTasks); + + distributorTask = Task.Run(() => DistributorLoop(distributorCts.Token), distributorCts.Token); } public async Task StopAsync(CancellationToken cancellationToken = default) { + await distributorCts.CancelAsync(); + + if (distributorTask != null) + { + try + { + await distributorTask; + } + catch (OperationCanceledException) + { + // Expected + } + } + + foreach (ServerConnection server in servers) + { + server.TaskChannel.Writer.Complete(); + } + List stopTasks = []; - foreach (HubConnection connection in connections) + foreach (ServerConnection server in servers) { - stopTasks.Add(connection.StopAsync(cancellationToken)); + stopTasks.Add(server.Connection.StopAsync(cancellationToken)); } await Task.WhenAll(stopTasks); @@ -94,52 +137,15 @@ public class RemoteExecutor public Dictionary GetCurrentServerMetrics() { - Dictionary metrics = []; - - foreach (HubConnection connection in connections) - { - if (serverMetrics.TryGetValue(connection, out ServerMetrics? newServerMetrics)) - { - metrics[newServerMetrics.ServerId] = newServerMetrics; - } - } - - return metrics; + return servers + .Select(server => server.Metrics) + .Where(metrics => metrics != null) + .ToDictionary(metrics => metrics!.ServerId, metrics => metrics!); } - public bool TryExecute(TDelegate del, out TResult? result, params object[] args) where TDelegate : Delegate + public async Task Execute(TDelegate del, params object[] args) where TDelegate : Delegate { - try - { - result = Execute(del, args); - return true; - } - catch - { - result = default; - return false; - } - } - - public bool TryExecute(TDelegate del, out TResult? result, [NotNullWhen(false)] out Exception? exception, params object[] args) where TDelegate : Delegate - { - try - { - result = Execute(del, args); - exception = null; - return true; - } - catch (Exception ex) - { - result = default; - exception = ex; - return false; - } - } - - public TResult Execute(TDelegate del, params object[] args) where TDelegate : Delegate - { - object? execResult = Execute(del, args); + object? execResult = await Execute(del, args); if (execResult is TResult typedResult) { @@ -155,7 +161,7 @@ public class RemoteExecutor } } - public object? Execute(T del, params object[] args) where T : Delegate + public async Task Execute(T del, params object[] args) where T : Delegate { MethodInfo method = del.Method; Type declaringType = method.DeclaringType!; @@ -175,12 +181,20 @@ public class RemoteExecutor Arguments = args }; - HubConnection connection = GetNextConnection(); + Guid taskId = Guid.NewGuid(); + TaskCompletionSource tcs = new(); + pendingResults[taskId] = tcs; - RemoteExecutionResult result = connection - .InvokeAsync("Execute", request) - .GetAwaiter() - .GetResult(); + PendingTask pendingTask = new PendingTask + { + TaskId = taskId, + Request = request, + EnqueuedAt = DateTime.UtcNow + }; + + globalQueue.Add(pendingTask); + + RemoteExecutionResult result = await tcs.Task; if (result.Exception != null) { @@ -190,62 +204,145 @@ public class RemoteExecutor return result.Result; } - private HubConnection GetNextConnection() + private async Task DistributorLoop(CancellationToken cancellationToken) { - if (connections.Count == 0) + while (!cancellationToken.IsCancellationRequested) { - throw new InvalidOperationException("No connections available"); + try + { + // Check if there are tasks in the global queue + PendingTask pendingTask = globalQueue.Take(cancellationToken); + + // Select the best server based on metrics + ServerConnection? bestServer = SelectBestServer(); + + if (bestServer != null) + { + // Create task item with ID + TaskItem taskItem = new TaskItem + { + TaskId = pendingTask.TaskId, + Request = pendingTask.Request + }; + + // Push to server's channel - SignalR will stream it + await bestServer.TaskChannel.Writer.WriteAsync(taskItem, cancellationToken); + } + else + { + // No available server, re-enqueue + globalQueue.Add(pendingTask, cancellationToken); + await Task.Delay(100, cancellationToken); + } + } + catch (OperationCanceledException) + { + break; + } + catch (Exception ex) + { + Debug.WriteLine($"DistributorLoop exception: {ex}"); + await Task.Delay(100, cancellationToken); + } + } + } + + private ServerConnection? SelectBestServer() + { + if (servers.Count == 0) + { + return null; } return loadBalancingStrategy switch { - LoadBalancingStrategy.RoundRobin => GetRoundRobinConnection(), - LoadBalancingStrategy.Random => GetRandomConnection(), - LoadBalancingStrategy.LeastConnections => GetLeastConnections(), - LoadBalancingStrategy.LeastActiveTasks => GetLeastActiveTasksConnections(), - LoadBalancingStrategy.ResourceAware => GetResourceAwareConnections(), - _ => throw new NotSupportedException($"Load balancing strategy {loadBalancingStrategy} is not supported") + LoadBalancingStrategy.ResourceAware => servers.MinBy(s => + { + if (s.Metrics == null) + { + return double.MaxValue; + } + + double cpuScore = s.Metrics.CpuUsage; + double activeTaskScore = s.Metrics.ActiveTasks * 10; + double backlogScore = s.TaskChannel.Reader.Count * 50; + + return cpuScore + activeTaskScore + backlogScore; + }), + LoadBalancingStrategy.LeastBacklog => servers.MinBy(s => s.TaskChannel.Reader.Count), + + _ => servers.FirstOrDefault() }; } - private HubConnection GetRoundRobinConnection() + private sealed class ServerConnection(HubConnection connection, HttpClient httpClient) { - lock (@lock) + public HubConnection Connection { get; } = connection; + public HttpClient HttpClient { get; } = httpClient; + public Channel TaskChannel { get; } = Channel.CreateUnbounded(); + public ServerMetrics? Metrics { get; set; } + } + + private sealed class PendingTask + { + public required Guid TaskId { get; init; } + public required RemoteExecutionRequest Request { get; init; } + public required DateTime EnqueuedAt { get; init; } + } + + // S2930: Dispose distributorCts when no longer needed + protected virtual void Dispose(bool disposing) + { + if (!disposedValue) { - HubConnection connection = connections[_currentConnectionIndex]; - _currentConnectionIndex = (_currentConnectionIndex + 1) % connections.Count; - return connection; + if (disposing) + { + distributorCts.Dispose(); + // Dispose managed state (managed objects) here if needed + } + + // Free unmanaged resources (unmanaged objects) and override finalizer if needed + // Set large fields to null if needed + + disposedValue = true; } } - private HubConnection GetRandomConnection() + public void Dispose() { - int index = Random.Shared.Next(connections.Count); - return connections[index]; - } - - private HubConnection GetLeastConnections() - { - // Purely looks at how many clients are connected to the Hub - return connections.OrderBy(c => serverMetrics.TryGetValue(c, out ServerMetrics? m) ? m.ActiveConnections : 0).First(); - } - - private HubConnection GetLeastActiveTasksConnections() - { - return connections.OrderBy(c => serverMetrics.TryGetValue(c, out ServerMetrics? m) ? m.ActiveTasks : 0).First(); - } - - private HubConnection GetResourceAwareConnections() - { - return connections.OrderBy(c => - { - if (!serverMetrics.TryGetValue(c, out ServerMetrics? m)) - { - return 0; - } - - // Simple heuristic: CPU percentage + (Memory in MB / 1024) - return m.CpuUsage + (m.TotalMemoryUsage / 1024 / 1024 / 100); - }).First(); + // Do not change this code. Put cleanup code in 'Dispose(bool disposing)' method + Dispose(disposing: true); + GC.SuppressFinalize(this); } } + +internal class RemoteExecLoggerProvider : ILoggerProvider +{ + public ILogger CreateLogger(string categoryName) + { + return new RemoteExecLogger(); + } + + public void Dispose() + { + throw new NotImplementedException(); + } +} + +internal class RemoteExecLogger : ILogger +{ + public IDisposable? BeginScope(TState state) where TState : notnull + { + return null; + } + + public bool IsEnabled(LogLevel logLevel) + { + return true; + } + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) + { + Console.WriteLine($"[{logLevel}] {formatter(state, exception)}"); + } +} \ No newline at end of file diff --git a/RemoteExec.Server/Controllers/AssemblyController.cs b/RemoteExec.Server/Controllers/AssemblyController.cs new file mode 100644 index 0000000..ecb0097 --- /dev/null +++ b/RemoteExec.Server/Controllers/AssemblyController.cs @@ -0,0 +1,32 @@ +using Microsoft.AspNetCore.Mvc; + +using RemoteExec.Server.Hubs; + +namespace RemoteExec.Server.Controllers; + +[ApiController] +[Route("/")] +public class AssemblyController(ILogger logger) : ControllerBase +{ + [HttpPost("provide-assembly")] + public async Task ProvideAssembly([FromQuery] Guid requestId) + { + try + { + using MemoryStream ms = new MemoryStream(); + await Request.Body.CopyToAsync(ms); + byte[] assemblyBytes = ms.ToArray(); + + logger.LogInformation("Received assembly for request {RequestId}, size: {Size} bytes", requestId, assemblyBytes.Length); + + await RemoteExecutionHub.ProvideAssembly(requestId, assemblyBytes); + + return Ok(); + } + catch (Exception ex) + { + logger.LogError(ex, "Error providing assembly for request {RequestId}", requestId); + return StatusCode(500, ex.Message); + } + } +} \ No newline at end of file diff --git a/RemoteExec.Server/Hubs/RemoteExecutionHub.cs b/RemoteExec.Server/Hubs/RemoteExecutionHub.cs index 1832e63..a1de283 100644 --- a/RemoteExec.Server/Hubs/RemoteExecutionHub.cs +++ b/RemoteExec.Server/Hubs/RemoteExecutionHub.cs @@ -6,7 +6,6 @@ using System.Collections.Concurrent; using System.Diagnostics; using System.Reflection; using System.Text.Json; -using System.Threading.Channels; namespace RemoteExec.Server.Hubs; @@ -16,10 +15,13 @@ public class RemoteExecutionHub(ILogger logger) : Hub private static readonly ConcurrentDictionary> pendingAssemblyRequests = new(); + private static ServerMetrics? lastMetrics; private static DateTime lastMetricsTimestamp; private static TimeSpan lastTotalProcessorTime; private static int activeTasks = 0; + private static readonly int maxConcurrentTasks = Environment.ProcessorCount * 2; + private static readonly SemaphoreSlim taskSemaphore = new SemaphoreSlim(maxConcurrentTasks, maxConcurrentTasks); public override Task OnConnectedAsync() { @@ -43,10 +45,62 @@ public class RemoteExecutionHub(ILogger logger) : Hub return base.OnDisconnectedAsync(exception); } + public async Task StartTaskStream(IAsyncEnumerable taskStream) + { + logger.LogInformation("Starting task stream for connection {ConnectionId}", Context.ConnectionId); + + try + { + await foreach (TaskItem taskItem in taskStream.WithCancellation(Context.ConnectionAborted)) + { + // Wait for available slot before processing + await taskSemaphore.WaitAsync(Context.ConnectionAborted); + + // Process task asynchronously without blocking the stream + _ = Task.Run(async () => + { + try + { + RemoteExecutionResult result = await ExecuteTask(taskItem.Request); + await Clients.Caller.SendAsync("TaskResult", taskItem.TaskId, result); + } + catch (Exception ex) + { + logger.LogError(ex, "Error processing task {TaskId}", taskItem.TaskId); + RemoteExecutionResult errorResult = new RemoteExecutionResult + { + Exception = ex.ToString() + }; + await Clients.Caller.SendAsync("TaskResult", taskItem.TaskId, errorResult); + } + finally + { + _ = taskSemaphore.Release(); + } + }, Context.ConnectionAborted); + } + } + catch (OperationCanceledException ex) + { + logger.LogError(ex, "Task stream for connection {ConnectionId} was canceled", Context.ConnectionId); + } + catch (Exception ex) + { + logger.LogError(ex, "Error in task stream for connection {ConnectionId}", Context.ConnectionId); + } + } + public async Task Execute(RemoteExecutionRequest req) + { + return await ExecuteTask(req); + } + + private async Task ExecuteTask(RemoteExecutionRequest req) { _ = Interlocked.Increment(ref activeTasks); + logger.LogInformation(req.MethodName); + try { if (!connections.TryGetValue(Context.ConnectionId, out RemoteJobAssemblyLoadContext? assemblyLoadContext)) @@ -155,20 +209,8 @@ public class RemoteExecutionHub(ILogger logger) : Hub } } - public async Task ProvideAssembly(Guid requestId, ChannelReader stream) + public static async Task ProvideAssembly(Guid requestId, byte[] assemblyBytes) { - using MemoryStream ms = new MemoryStream(); - - while (await stream.WaitToReadAsync()) - { - while (stream.TryRead(out byte item)) - { - ms.WriteByte(item); - } - } - - byte[] assemblyBytes = ms.ToArray(); - if (pendingAssemblyRequests.TryRemove(requestId, out TaskCompletionSource? tcs)) { tcs.SetResult(assemblyBytes); @@ -183,6 +225,24 @@ public class RemoteExecutionHub(ILogger logger) : Hub public static async Task BroadcastMetricsAsync(IHubContext hubContext) { ServerMetrics metrics = await GetServerMetrics(); + + // Only broadcast if metrics have changed by a significant amount + if (lastMetrics is not null) + { + double cpuDiff = Math.Abs(metrics.CpuUsage - lastMetrics.CpuUsage); + long memoryDiff = Math.Abs(metrics.TotalMemoryUsage - lastMetrics.TotalMemoryUsage); + int connectionsDiff = Math.Abs(metrics.ActiveConnections - lastMetrics.ActiveConnections); + int tasksDiff = Math.Abs(metrics.ActiveTasks - lastMetrics.ActiveTasks); + int maxTasksDiff = Math.Abs(metrics.MaxConcurrentTasks - lastMetrics.MaxConcurrentTasks); + + if (cpuDiff < 1.0 && memoryDiff < 10 * 1024 * 1024 && connectionsDiff == 0 && tasksDiff == 0 && maxTasksDiff == 0) + { + return; // No significant change + } + } + + lastMetrics = metrics; + await hubContext.Clients.All.SendAsync("MetricsUpdated", metrics); } @@ -211,6 +271,7 @@ public class RemoteExecutionHub(ILogger logger) : Hub ServerId = Environment.MachineName, ActiveConnections = connections.Count, ActiveTasks = activeTasks, + MaxConcurrentTasks = maxConcurrentTasks, TotalMemoryUsage = currentProcess.WorkingSet64, CpuUsage = Math.Clamp(Math.Round(cpuUsagePercent, 2), 0, 100), Timestamp = currentTime diff --git a/RemoteExec.Server/Services/MetricsBroadcastService.cs b/RemoteExec.Server/Services/MetricsBroadcastService.cs index f7fd24f..0855265 100644 --- a/RemoteExec.Server/Services/MetricsBroadcastService.cs +++ b/RemoteExec.Server/Services/MetricsBroadcastService.cs @@ -6,7 +6,7 @@ namespace RemoteExec.Server.Services; public class MetricsBroadcastService(IHubContext hubContext, ILogger logger) : BackgroundService { - private readonly TimeSpan _broadcastInterval = TimeSpan.FromSeconds(2); + private readonly TimeSpan _broadcastInterval = TimeSpan.FromMilliseconds(500); protected override async Task ExecuteAsync(CancellationToken stoppingToken) { diff --git a/RemoteExec.Shared/ServerMetrics.cs b/RemoteExec.Shared/ServerMetrics.cs index 588f905..b3a2f2d 100644 --- a/RemoteExec.Shared/ServerMetrics.cs +++ b/RemoteExec.Shared/ServerMetrics.cs @@ -4,6 +4,7 @@ public sealed class ServerMetrics { public int ActiveConnections { get; set; } public int ActiveTasks { get; set; } + public int MaxConcurrentTasks { get; set; } public long TotalMemoryUsage { get; set; } public double CpuUsage { get; set; } public DateTime Timestamp { get; set; } diff --git a/RemoteExec.Shared/TaskItem.cs b/RemoteExec.Shared/TaskItem.cs new file mode 100644 index 0000000..0965379 --- /dev/null +++ b/RemoteExec.Shared/TaskItem.cs @@ -0,0 +1,7 @@ +namespace RemoteExec.Shared; + +public class TaskItem +{ + public Guid TaskId { get; set; } + public required RemoteExecutionRequest Request { get; set; } +} \ No newline at end of file diff --git a/RemoteExec/Program.cs b/RemoteExec/Program.cs index 82b9126..ea23267 100644 --- a/RemoteExec/Program.cs +++ b/RemoteExec/Program.cs @@ -17,16 +17,16 @@ singleHostExecutor.MetricsUpdated += (sender, e) => await singleHostExecutor.StartAsync(); -bool success = singleHostExecutor.TryExecute(Multiply, out int result, 2, 4); -Console.WriteLine($"Success: {success}, Result: {result}"); - -result = singleHostExecutor.Execute>, int>(Multiply, 3, 5); -Console.WriteLine($"Result: {result}"); +await Parallel.ForAsync(0, 1000, async (i, cancellationToken) => +{ + int r = await singleHostExecutor.Execute>, int>(Multiply, i, i + 1); + Console.WriteLine($"Multiply {i} * {i + 1} = {r}"); +}); await singleHostExecutor.StopAsync(); static async Task Multiply(int x, int y) { - await Task.Delay(5000); + await Task.Delay(Random.Shared.Next(100, 500)); return x.Multiply(y); } \ No newline at end of file