diff --git a/RemoteExec.Client/RemoteExecLoggerProvider.cs b/RemoteExec.Client/RemoteExecLoggerProvider.cs new file mode 100644 index 0000000..f8ae5f6 --- /dev/null +++ b/RemoteExec.Client/RemoteExecLoggerProvider.cs @@ -0,0 +1,34 @@ +using Microsoft.Extensions.Logging; + +namespace RemoteExec.Client; + +internal class RemoteExecLoggerProvider : ILoggerProvider +{ + private bool disposedValue; + private readonly ILogger logger; + + public RemoteExecLoggerProvider(ILogger logger) + { + this.logger = logger; + } + + public ILogger CreateLogger(string categoryName) + { + return logger; + } + + protected virtual void Dispose(bool disposing) + { + if (!disposedValue) + { + // No managed or unmanaged resources to dispose + disposedValue = true; + } + } + + public void Dispose() + { + Dispose(disposing: true); + GC.SuppressFinalize(this); + } +} \ No newline at end of file diff --git a/RemoteExec.Client/RemoteExecutor.cs b/RemoteExec.Client/RemoteExecutor.cs index e799a64..5154359 100644 --- a/RemoteExec.Client/RemoteExecutor.cs +++ b/RemoteExec.Client/RemoteExecutor.cs @@ -1,5 +1,6 @@ using Microsoft.AspNetCore.SignalR.Client; using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; using RemoteExec.Shared; @@ -19,18 +20,35 @@ public class RemoteExecutor : IDisposable private CancellationTokenSource distributorCts = new(); private Task? distributorTask; private readonly LoadBalancingStrategy loadBalancingStrategy; - + private readonly ILogger logger; private bool disposedValue; public event EventHandler? MetricsUpdated; - public RemoteExecutor(string url) : this([url], LoadBalancingStrategy.ResourceAware) + public RemoteExecutor(string url) : this([url], LoadBalancingStrategy.ResourceAware, NullLogger.Instance) { } - public RemoteExecutor(string[] urls, LoadBalancingStrategy loadBalancingStrategy = LoadBalancingStrategy.ResourceAware) + public RemoteExecutor(string url, ILogger logger) : this([url], LoadBalancingStrategy.ResourceAware, logger) + { + } + + public RemoteExecutor(string url, LoadBalancingStrategy loadBalancingStrategy) : this([url], loadBalancingStrategy, NullLogger.Instance) + { + } + + public RemoteExecutor(string[] urls) : this(urls, LoadBalancingStrategy.ResourceAware, NullLogger.Instance) + { + } + + public RemoteExecutor(string[] urls, LoadBalancingStrategy loadBalancingStrategy) : this(urls, loadBalancingStrategy, NullLogger.Instance) + { + } + + public RemoteExecutor(string[] urls, LoadBalancingStrategy loadBalancingStrategy, ILogger logger) { this.loadBalancingStrategy = loadBalancingStrategy; + this.logger = logger; foreach (string url in urls) { @@ -42,7 +60,7 @@ public class RemoteExecutor : IDisposable .WithAutomaticReconnect() .ConfigureLogging(logging => { - _ = logging.AddProvider(new RemoteExecLoggerProvider()); + _ = logging.AddProvider(new RemoteExecLoggerProvider(logger)); }) .Build(); @@ -106,6 +124,8 @@ public class RemoteExecutor : IDisposable public async Task StopAsync(CancellationToken cancellationToken = default) { + logger.LogInformation("Stopping RemoteExecutor..."); + await distributorCts.CancelAsync(); if (distributorTask != null) @@ -113,13 +133,15 @@ public class RemoteExecutor : IDisposable try { await distributorTask; + logger.LogDebug("Distributor task completed successfully"); } - catch (OperationCanceledException) + catch (OperationCanceledException ex) { - // Expected + logger.LogError(ex, "Distributor task was canceled"); } } + logger.LogDebug("Completing task channels for {ServerCount} servers", servers.Count); foreach (ServerConnection server in servers) { server.TaskChannel.Writer.Complete(); @@ -133,6 +155,7 @@ public class RemoteExecutor : IDisposable } await Task.WhenAll(stopTasks); + logger.LogInformation("RemoteExecutor stopped successfully"); } public Dictionary GetCurrentServerMetrics() @@ -315,34 +338,3 @@ public class RemoteExecutor : IDisposable 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/Hubs/RemoteExecutionHub.cs b/RemoteExec.Server/Hubs/RemoteExecutionHub.cs index a1de283..6434cf9 100644 --- a/RemoteExec.Server/Hubs/RemoteExecutionHub.cs +++ b/RemoteExec.Server/Hubs/RemoteExecutionHub.cs @@ -15,6 +15,9 @@ public class RemoteExecutionHub(ILogger logger) : Hub private static readonly ConcurrentDictionary> pendingAssemblyRequests = new(); + // Track pending assembly requests per connection to avoid duplicate requests + private static readonly ConcurrentDictionary>> pendingAssemblyRequestsByConnection = new(); + private static ServerMetrics? lastMetrics; private static DateTime lastMetricsTimestamp; private static TimeSpan lastTotalProcessorTime; @@ -28,6 +31,7 @@ public class RemoteExecutionHub(ILogger logger) : Hub RemoteJobAssemblyLoadContext assemblyLoadContext = new RemoteJobAssemblyLoadContext($"RemoteJob_{Guid.NewGuid()}"); _ = connections.TryAdd(Context.ConnectionId, assemblyLoadContext); + _ = pendingAssemblyRequestsByConnection.TryAdd(Context.ConnectionId, new ConcurrentDictionary>()); logger.LogInformation("Connection {ConnectionId} established", Context.ConnectionId); @@ -42,6 +46,8 @@ public class RemoteExecutionHub(ILogger logger) : Hub logger.LogInformation("Connection {ConnectionId} disconnected", Context.ConnectionId); } + _ = pendingAssemblyRequestsByConnection.TryRemove(Context.ConnectionId, out _); + return base.OnDisconnectedAsync(exception); } @@ -99,8 +105,6 @@ public class RemoteExecutionHub(ILogger logger) : Hub { _ = Interlocked.Increment(ref activeTasks); - logger.LogInformation(req.MethodName); - try { if (!connections.TryGetValue(Context.ConnectionId, out RemoteJobAssemblyLoadContext? assemblyLoadContext)) @@ -282,17 +286,36 @@ public class RemoteExecutionHub(ILogger logger) : Hub { try { - Guid guid = Guid.NewGuid(); - TaskCompletionSource tcs = new TaskCompletionSource(); + if (!pendingAssemblyRequestsByConnection.TryGetValue(Context.ConnectionId, out ConcurrentDictionary>? connectionPendingRequests)) + { + throw new InvalidOperationException("Connection not found"); + } - _ = pendingAssemblyRequests.TryAdd(guid, tcs); + Task assemblyBytesTask = connectionPendingRequests.GetOrAdd(assemblyName, key => + { + return Task.Run(async () => + { + try + { + Guid guid = Guid.NewGuid(); + TaskCompletionSource tcs = new TaskCompletionSource(); - await Clients.Caller.SendAsync("RequestAssembly", assemblyName, guid); + _ = pendingAssemblyRequests.TryAdd(guid, tcs); - // Wait for the assembly with a timeout - byte[] assemblyBytes = await tcs.Task.WaitAsync(TimeSpan.FromSeconds(30)); + await Clients.Caller.SendAsync("RequestAssembly", key, guid); + + // Wait for the assembly with a timeout + return await tcs.Task.WaitAsync(TimeSpan.FromSeconds(30)); + } + finally + { + _ = connectionPendingRequests.TryRemove(key, out _); + } + }); + }); + + byte[] assemblyBytes = await assemblyBytesTask; - // Return a temporary assembly just for metadata inspection using MemoryStream ms = new MemoryStream(assemblyBytes); return Assembly.Load(assemblyBytes); } @@ -307,16 +330,34 @@ public class RemoteExecutionHub(ILogger logger) : Hub { string assemblyName = assembly.GetName().FullName!; - Guid guid = Guid.NewGuid(); - TaskCompletionSource tcs = new TaskCompletionSource(); + if (!pendingAssemblyRequestsByConnection.TryGetValue(Context.ConnectionId, out ConcurrentDictionary>? connectionPendingRequests)) + { + throw new InvalidOperationException("Connection not found"); + } - _ = pendingAssemblyRequests.TryAdd(guid, tcs); + Task assemblyBytesTask = connectionPendingRequests.GetOrAdd(assemblyName, key => + { + return Task.Run(async () => + { + try + { + Guid guid = Guid.NewGuid(); + TaskCompletionSource tcs = new TaskCompletionSource(); - await Clients.Caller.SendAsync("RequestAssembly", assemblyName, guid); + _ = pendingAssemblyRequests.TryAdd(guid, tcs); - byte[] assemblyBytes = await tcs.Task.WaitAsync(TimeSpan.FromSeconds(30)); + await Clients.Caller.SendAsync("RequestAssembly", key, guid); - return assemblyBytes; + return await tcs.Task.WaitAsync(TimeSpan.FromSeconds(30)); + } + finally + { + _ = connectionPendingRequests.TryRemove(key, out _); + } + }); + }); + + return await assemblyBytesTask; } private async Task PreLoadReferencedAssembliesAsync(RemoteJobAssemblyLoadContext assemblyLoadContext, Assembly assembly)