diff --git a/RemoteExec.Client.DependencyInjection/RemoteExec.Client.DependencyInjection.csproj b/RemoteExec.Client.DependencyInjection/RemoteExec.Client.DependencyInjection.csproj new file mode 100644 index 0000000..87def2e --- /dev/null +++ b/RemoteExec.Client.DependencyInjection/RemoteExec.Client.DependencyInjection.csproj @@ -0,0 +1,13 @@ + + + + net10.0 + enable + enable + + + + + + + diff --git a/RemoteExec.Client.DependencyInjection/RemoteExecutorServiceCollectionExtensions.cs b/RemoteExec.Client.DependencyInjection/RemoteExecutorServiceCollectionExtensions.cs new file mode 100644 index 0000000..89dac4a --- /dev/null +++ b/RemoteExec.Client.DependencyInjection/RemoteExecutorServiceCollectionExtensions.cs @@ -0,0 +1,26 @@ +// In RemoteExec.Client.DependencyInjection (Extension Project) +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; + +namespace RemoteExec.Client.DependencyInjection; + +public static class RemoteExecutorServiceCollectionExtensions +{ + public static IServiceCollection AddRemoteExecutor(this IServiceCollection services, string[] urls, Action? configure = null) + { + RemoteExecutorOptions options = new RemoteExecutorOptions(); + configure?.Invoke(options); + + _ = services.AddSingleton(options); + + _ = services.AddSingleton(sp => + { + RemoteExecutorOptions opt = sp.GetRequiredService(); + ILogger logger = sp.GetRequiredService>(); + + return new RemoteExecutor(urls, opt, logger); + }); + + return services; + } +} \ No newline at end of file diff --git a/RemoteExec.Client/Exceptions/MaxRetriesException.cs b/RemoteExec.Client/Exceptions/MaxRetriesException.cs new file mode 100644 index 0000000..4c96cce --- /dev/null +++ b/RemoteExec.Client/Exceptions/MaxRetriesException.cs @@ -0,0 +1,5 @@ +namespace RemoteExec.Client.Exceptions; + +public class MaxRetriesException(string message) : Exception(message) +{ +} diff --git a/RemoteExec.Client/RemoteExecutionException.cs b/RemoteExec.Client/Exceptions/RemoteExecutionException.cs similarity index 64% rename from RemoteExec.Client/RemoteExecutionException.cs rename to RemoteExec.Client/Exceptions/RemoteExecutionException.cs index edf792b..c2fe637 100644 --- a/RemoteExec.Client/RemoteExecutionException.cs +++ b/RemoteExec.Client/Exceptions/RemoteExecutionException.cs @@ -1,4 +1,4 @@ -namespace RemoteExec.Client; +namespace RemoteExec.Client.Exceptions; public class RemoteExecutionException(string message) : Exception(message) { diff --git a/RemoteExec.Client/ILoadBalancingStrategy.cs b/RemoteExec.Client/ILoadBalancingStrategy.cs new file mode 100644 index 0000000..10cbc78 --- /dev/null +++ b/RemoteExec.Client/ILoadBalancingStrategy.cs @@ -0,0 +1,6 @@ +namespace RemoteExec.Client; + +public interface ILoadBalancingStrategy +{ + ServerConnection? SelectServer(IEnumerable availableServers); +} diff --git a/RemoteExec.Client/LoadBalancingStrategies/LeastBacklogStrategy.cs b/RemoteExec.Client/LoadBalancingStrategies/LeastBacklogStrategy.cs new file mode 100644 index 0000000..f39f8b7 --- /dev/null +++ b/RemoteExec.Client/LoadBalancingStrategies/LeastBacklogStrategy.cs @@ -0,0 +1,9 @@ +namespace RemoteExec.Client.LoadBalancingStrategies; + +public class LeastBacklogStrategy : ILoadBalancingStrategy +{ + public ServerConnection? SelectServer(IEnumerable availableServers) + { + return availableServers.MinBy(s => s.TaskChannel.Reader.Count); + } +} diff --git a/RemoteExec.Client/LoadBalancingStrategies/ResourceAwareStrategy.cs b/RemoteExec.Client/LoadBalancingStrategies/ResourceAwareStrategy.cs new file mode 100644 index 0000000..2b3c08d --- /dev/null +++ b/RemoteExec.Client/LoadBalancingStrategies/ResourceAwareStrategy.cs @@ -0,0 +1,21 @@ +namespace RemoteExec.Client.LoadBalancingStrategies; + +internal class ResourceAwareStrategy : ILoadBalancingStrategy +{ + public ServerConnection? SelectServer(IEnumerable availableServers) + { + return availableServers.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; + }); + } +} diff --git a/RemoteExec.Client/LoadBalancingStrategy.cs b/RemoteExec.Client/LoadBalancingStrategy.cs deleted file mode 100644 index f0392ac..0000000 --- a/RemoteExec.Client/LoadBalancingStrategy.cs +++ /dev/null @@ -1,7 +0,0 @@ -namespace RemoteExec.Client; - -public enum LoadBalancingStrategy -{ - ResourceAware, - LeastBacklog -} \ No newline at end of file diff --git a/RemoteExec.Client/RemoteExecutor.ConnectionManagement.cs b/RemoteExec.Client/RemoteExecutor.ConnectionManagement.cs index f20229e..787c68a 100644 --- a/RemoteExec.Client/RemoteExecutor.ConnectionManagement.cs +++ b/RemoteExec.Client/RemoteExecutor.ConnectionManagement.cs @@ -1,6 +1,7 @@ using Microsoft.AspNetCore.SignalR.Client; using Microsoft.Extensions.Logging; +using RemoteExec.Client.Exceptions; using RemoteExec.Shared; using System.Collections.Concurrent; @@ -92,8 +93,8 @@ public partial class RemoteExecutor private async Task AttemptReconnectionAsync(ServerConnection server, CancellationToken cancellationToken) { - int retryDelay = 5000; // Start with 5 seconds - int maxRetryDelay = 60000; // Max 1 minute + int retryDelay = (int)options.ServerReconnectInitialDelay.TotalMilliseconds; + int maxRetryDelay = (int)options.ServerReconnectMaxDelay.TotalMilliseconds; while (!cancellationToken.IsCancellationRequested && server.Connection.State == HubConnectionState.Disconnected) { @@ -151,8 +152,17 @@ public partial class RemoteExecutor { if (taskDict.TryRemove(id, out PendingTask? task) && pendingResults.ContainsKey(id)) { - globalQueue.Add(task); - count++; + task.RetryCount++; + + if (task.RetryCount <= options.MaxTaskRetries) + { + globalQueue.Add(task); + count++; + } + else if (pendingResults.TryRemove(id, out TaskCompletionSource? tcs)) + { + tcs.SetException(new MaxRetriesException("Max retry attempts reached.")); + } } } diff --git a/RemoteExec.Client/RemoteExecutor.InternalTypes.cs b/RemoteExec.Client/RemoteExecutor.InternalTypes.cs index 737ccdc..4daec93 100644 --- a/RemoteExec.Client/RemoteExecutor.InternalTypes.cs +++ b/RemoteExec.Client/RemoteExecutor.InternalTypes.cs @@ -1,26 +1,15 @@ -using Microsoft.AspNetCore.SignalR.Client; - using RemoteExec.Shared; -using System.Threading.Channels; - namespace RemoteExec.Client; public partial class RemoteExecutor { - private sealed class ServerConnection(HubConnection connection, HttpClient httpClient) - { - public HubConnection Connection { get; } = connection; - public HttpClient HttpClient { get; } = httpClient; - public Channel TaskChannel { get; } = Channel.CreateUnbounded(); - public ServerMetrics? Metrics { get; set; } - public Func? ClosedEventHandler { get; set; } - } private sealed class PendingTask { public required Guid TaskId { get; init; } public required RemoteExecutionRequest Request { get; init; } public required DateTime EnqueuedAt { get; init; } + public int RetryCount { get; set; } = 0; } } \ No newline at end of file diff --git a/RemoteExec.Client/RemoteExecutor.TaskDistribution.cs b/RemoteExec.Client/RemoteExecutor.TaskDistribution.cs index 8a56985..7b72906 100644 --- a/RemoteExec.Client/RemoteExecutor.TaskDistribution.cs +++ b/RemoteExec.Client/RemoteExecutor.TaskDistribution.cs @@ -79,24 +79,6 @@ public partial class RemoteExecutor return null; } - return loadBalancingStrategy switch - { - LoadBalancingStrategy.ResourceAware => connectedServers.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 => connectedServers.MinBy(s => s.TaskChannel.Reader.Count), - - _ => connectedServers.FirstOrDefault() - }; + return options.Strategy.SelectServer(connectedServers); } } \ No newline at end of file diff --git a/RemoteExec.Client/RemoteExecutor.cs b/RemoteExec.Client/RemoteExecutor.cs index 89af28a..06ee796 100644 --- a/RemoteExec.Client/RemoteExecutor.cs +++ b/RemoteExec.Client/RemoteExecutor.cs @@ -2,6 +2,7 @@ using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; +using RemoteExec.Client.Exceptions; using RemoteExec.Shared; using System.Collections.Concurrent; @@ -12,8 +13,8 @@ namespace RemoteExec.Client; public partial class RemoteExecutor : IAsyncDisposable { + private readonly BlockingCollection globalQueue; private readonly List servers = []; - private readonly BlockingCollection globalQueue = []; private readonly ConcurrentDictionary> pendingResults = new(); private readonly ConcurrentDictionary> serverAssignedTasks = new(); @@ -21,38 +22,65 @@ public partial class RemoteExecutor : IAsyncDisposable private CancellationTokenSource distributorCts = new(); private Task? distributorTask; - private readonly LoadBalancingStrategy loadBalancingStrategy; + private readonly RemoteExecutorOptions options = new(); private readonly ILogger logger; private bool disposedValue; public event EventHandler? MetricsUpdated; - public RemoteExecutor(string url) : this([url], LoadBalancingStrategy.ResourceAware, NullLogger.Instance) + public RemoteExecutor(string url) : this([url], new RemoteExecutorOptions(), NullLogger.Instance) { } - public RemoteExecutor(string url, ILogger logger) : this([url], LoadBalancingStrategy.ResourceAware, logger) + public RemoteExecutor(string url, ILogger logger) : this([url], new RemoteExecutorOptions(), logger) { } - public RemoteExecutor(string url, LoadBalancingStrategy loadBalancingStrategy) : this([url], loadBalancingStrategy, NullLogger.Instance) + public RemoteExecutor(string url, Action configure) : this([url], NullLogger.Instance, configure) { } - public RemoteExecutor(string[] urls) : this(urls, LoadBalancingStrategy.ResourceAware, NullLogger.Instance) + public RemoteExecutor(string url, ILogger logger, Action configure) : this([url], logger, configure) { } - public RemoteExecutor(string[] urls, LoadBalancingStrategy loadBalancingStrategy) : this(urls, loadBalancingStrategy, NullLogger.Instance) + public RemoteExecutor(string[] urls) : this(urls, new RemoteExecutorOptions(), NullLogger.Instance) { } - public RemoteExecutor(string[] urls, LoadBalancingStrategy loadBalancingStrategy, ILogger logger) + public RemoteExecutor(string[] urls, ILogger logger) : this(urls, new RemoteExecutorOptions(), logger) + { + } + + public RemoteExecutor(string[] urls, Action configure) : this(urls, NullLogger.Instance, configure) + { + } + + public RemoteExecutor(string[] urls, ILogger logger, Action configure) { - this.loadBalancingStrategy = loadBalancingStrategy; this.logger = logger; + options = new RemoteExecutorOptions(); + globalQueue = new BlockingCollection(options.GlobalQueueCapacity); + + configure(options); + + InitializeServers(urls); + } + + public RemoteExecutor(string[] urls, RemoteExecutorOptions options, ILogger logger) + { + this.logger = logger; + this.options = options; + + globalQueue = new BlockingCollection(this.options.GlobalQueueCapacity); + + InitializeServers(urls); + } + + private void InitializeServers(string[] urls) + { foreach (string url in urls) { Uri baseUri = new(url); @@ -85,9 +113,9 @@ public partial class RemoteExecutor : IAsyncDisposable .ToDictionary(metrics => metrics!.ServerId, metrics => metrics!); } - public async Task Execute(TDelegate del, params object[] args) where TDelegate : Delegate + public async Task ExecuteAsync(TDelegate @delegate, CancellationToken cancellationToken, params object[] args) where TDelegate : Delegate { - object? execResult = await Execute(del, args); + object? execResult = await ExecuteAsync(@delegate, cancellationToken, args); if (execResult is TResult typedResult) { @@ -103,12 +131,21 @@ public partial class RemoteExecutor : IAsyncDisposable } } - public async Task Execute(T del, params object[] args) where T : Delegate + public async Task ExecuteAsync(TDelegate @delegate, params object[] args) where TDelegate : Delegate { - MethodInfo method = del.Method; + return await ExecuteAsync(@delegate, CancellationToken.None, args); + } + + public async Task ExecuteAsync(T @delegate, CancellationToken cancellationToken, params object[] args) where T : Delegate + { + MethodInfo method = @delegate.Method; Type declaringType = method.DeclaringType!; Assembly assembly = declaringType.Assembly; + using CancellationTokenSource cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, distributorCts.Token); + using CancellationTokenSource timeoutCts = new CancellationTokenSource(options.ExecutionTimeout); + using CancellationTokenSource finalCts = CancellationTokenSource.CreateLinkedTokenSource(cts.Token, timeoutCts.Token); + if (!method.IsStatic) { throw new InvalidOperationException("Only static methods supported"); @@ -141,14 +178,26 @@ public partial class RemoteExecutor : IAsyncDisposable globalQueue.Add(pendingTask); - RemoteExecutionResult result = await tcs.Task; - - if (result.Exception != null) + try { - throw new RemoteExecutionException(result.Exception); - } + RemoteExecutionResult result = await tcs.Task.WaitAsync(finalCts.Token); - return result.Result; + if (result.Exception != null) + { + throw new RemoteExecutionException(result.Exception); + } + + return result.Result; + } + finally + { + _ = pendingResults.TryRemove(taskId, out _); + } + } + + public Task ExecuteAsync(T @delegate, params object[] args) where T : Delegate + { + return ExecuteAsync(@delegate, CancellationToken.None, args); } protected virtual async Task DisposeAsync(bool disposing) diff --git a/RemoteExec.Client/RemoteExecutorOptions.cs b/RemoteExec.Client/RemoteExecutorOptions.cs new file mode 100644 index 0000000..f4afd03 --- /dev/null +++ b/RemoteExec.Client/RemoteExecutorOptions.cs @@ -0,0 +1,21 @@ +using RemoteExec.Client.LoadBalancingStrategies; + +namespace RemoteExec.Client; + +public class RemoteExecutorOptions +{ + public ILoadBalancingStrategy Strategy { get; set; } = new ResourceAwareStrategy(); + + // How long to wait for a result before throwing a TimeoutException + public TimeSpan ExecutionTimeout { get; set; } = TimeSpan.FromMinutes(5); + + // Backoff settings for server reconnections + public TimeSpan ServerReconnectInitialDelay { get; set; } = TimeSpan.FromSeconds(5); + public TimeSpan ServerReconnectMaxDelay { get; set; } = TimeSpan.FromSeconds(60); + + // Reliability: How many times a task can be requeued before failing + public int MaxTaskRetries { get; set; } = 3; + + // Capacity: Stop accepting Execute calls if the global queue is too full + public int GlobalQueueCapacity { get; set; } = 1000; +} diff --git a/RemoteExec.Client/ServerConnection.cs b/RemoteExec.Client/ServerConnection.cs new file mode 100644 index 0000000..27aa4ad --- /dev/null +++ b/RemoteExec.Client/ServerConnection.cs @@ -0,0 +1,16 @@ +using Microsoft.AspNetCore.SignalR.Client; + +using RemoteExec.Shared; + +using System.Threading.Channels; + +namespace RemoteExec.Client; + +public class ServerConnection(HubConnection connection, HttpClient httpClient) +{ + public HubConnection Connection { get; } = connection; + public HttpClient HttpClient { get; } = httpClient; + public Channel TaskChannel { get; } = Channel.CreateUnbounded(); + public ServerMetrics? Metrics { get; set; } + public Func? ClosedEventHandler { get; set; } +} \ No newline at end of file diff --git a/RemoteExec.slnx b/RemoteExec.slnx index dfc2fd0..e89b600 100644 --- a/RemoteExec.slnx +++ b/RemoteExec.slnx @@ -1,4 +1,5 @@ + diff --git a/RemoteExec/Program.cs b/RemoteExec/Program.cs index ea23267..a6c7451 100644 --- a/RemoteExec/Program.cs +++ b/RemoteExec/Program.cs @@ -19,7 +19,7 @@ await singleHostExecutor.StartAsync(); await Parallel.ForAsync(0, 1000, async (i, cancellationToken) => { - int r = await singleHostExecutor.Execute>, int>(Multiply, i, i + 1); + int r = await singleHostExecutor.ExecuteAsync>, int>(Multiply, i, i + 1); Console.WriteLine($"Multiply {i} * {i + 1} = {r}"); });