From bd5b2a5c2e5b8ad86655c8803ecc12856b0c4c9b Mon Sep 17 00:00:00 2001 From: James Vaughan Date: Thu, 24 Sep 2026 01:23:24 -0400 Subject: [PATCH 01/18] Implement Multi-Node Estimation --- README.md | 17 + src/XTMF2.Client/Program.cs | 26 +- src/XTMF2.GUI/Properties/RunServerEndpoint.cs | 9 +- src/XTMF2.GUI/Properties/Settings.cs | 2 + src/XTMF2.GUI/RunController.cs | 347 ++++++++++ .../ViewModels/ModelSystemEditorViewModel.cs | 67 +- .../RemoteEstimationWorkerViewModel.cs | 36 + src/XTMF2.GUI/ViewModels/RunViewModel.cs | 74 ++ src/XTMF2.GUI/ViewModels/RunsViewModel.cs | 16 + src/XTMF2.GUI/Views/EstimationDialog.axaml | 16 + src/XTMF2.GUI/Views/EstimationDialog.axaml.cs | 17 + .../Views/RunConfigurationDialog.axaml | 50 +- .../Views/RunConfigurationDialog.axaml.cs | 127 +++- src/XTMF2.GUI/Views/RunsView.axaml | 40 +- src/XTMF2/Bus/HostBus.cs | 104 ++- .../Optimization/EstimationAlgorithmConfig.cs | 18 + .../EstimationAlgorithms/GeneticAlgorithm.cs | 40 +- .../NelderMeadAlgorithm.cs | 37 +- .../ParticleSwarmAlgorithm.cs | 36 +- .../Bus/Optimization/IEstimationAlgorithm.cs | 22 + ...emoteSharedEstimationCoordinatorSession.cs | 53 ++ .../Bus/RemoteSharedEstimationRegistry.cs | 305 +++++++++ src/XTMF2/Bus/RunContext.cs | 213 +++++- src/XTMF2/Bus/RunServerBus.cs | 100 ++- src/XTMF2/Bus/SharedEstimationCoordinator.cs | 313 +++++++++ .../Bus/SharedEstimationCoordinatorRun.cs | 127 ++++ src/XTMF2/Bus/SharedEstimationProtocol.cs | 643 ++++++++++++++++++ src/XTMF2/Bus/SharedEstimationWorker.cs | 15 + .../Bus/SharedEstimationWorkerConnection.cs | 101 +++ .../Bus/SharedEstimationWorkerParticipant.cs | 50 ++ src/XTMF2/Bus/SharedEstimationWorkerPool.cs | 252 +++++++ .../Bus/SharedEstimationWorkerSession.cs | 123 ++++ src/XTMF2/Editing/ModelSystemSession.cs | 45 ++ src/XTMF2/ModelSystemConstruct/ModelSystem.cs | 27 + .../Headless/RunConfigurationDialogTests.cs | 44 ++ .../Properties/TestRunServerEndpoint.cs | 64 ++ .../Bus/TestSharedEstimationCoordinator.cs | 96 +++ .../Bus/TestSharedEstimationEndToEnd.cs | 175 +++++ .../Bus/TestSharedEstimationProtocol.cs | 237 +++++++ .../TestSharedEstimationWorkerParticipant.cs | 57 ++ .../Modules/PathValidationModule.cs | 24 + 41 files changed, 4094 insertions(+), 71 deletions(-) create mode 100644 src/XTMF2.GUI/ViewModels/RemoteEstimationWorkerViewModel.cs create mode 100644 src/XTMF2/Bus/RemoteSharedEstimationCoordinatorSession.cs create mode 100644 src/XTMF2/Bus/RemoteSharedEstimationRegistry.cs create mode 100644 src/XTMF2/Bus/SharedEstimationCoordinator.cs create mode 100644 src/XTMF2/Bus/SharedEstimationCoordinatorRun.cs create mode 100644 src/XTMF2/Bus/SharedEstimationProtocol.cs create mode 100644 src/XTMF2/Bus/SharedEstimationWorker.cs create mode 100644 src/XTMF2/Bus/SharedEstimationWorkerConnection.cs create mode 100644 src/XTMF2/Bus/SharedEstimationWorkerParticipant.cs create mode 100644 src/XTMF2/Bus/SharedEstimationWorkerPool.cs create mode 100644 src/XTMF2/Bus/SharedEstimationWorkerSession.cs create mode 100644 tests/XTMF2.GUI.Tests/Headless/RunConfigurationDialogTests.cs create mode 100644 tests/XTMF2.GUI.Tests/Properties/TestRunServerEndpoint.cs create mode 100644 tests/XTMF2.UnitTests/Bus/TestSharedEstimationCoordinator.cs create mode 100644 tests/XTMF2.UnitTests/Bus/TestSharedEstimationEndToEnd.cs create mode 100644 tests/XTMF2.UnitTests/Bus/TestSharedEstimationProtocol.cs create mode 100644 tests/XTMF2.UnitTests/Bus/TestSharedEstimationWorkerParticipant.cs create mode 100644 tests/XTMF2.UnitTests/Modules/PathValidationModule.cs diff --git a/README.md b/README.md index 7315d56..29ed1f8 100644 --- a/README.md +++ b/README.md @@ -68,6 +68,23 @@ In XTMF2, open **Settings**, open **RunServers**, and add a remote endpoint. Ent The GUI pins the server certificate to this fingerprint and authenticates with the token before creating the RunServer bus. A mismatched certificate or token is rejected. Do not expose the TCP port to untrusted networks; use firewall rules or a private network as appropriate. +## Distributed estimation + +Estimation runs can evaluate candidate parameter vectors concurrently across multiple connected RunServers. Normal model runs and calibration runs continue to use one RunServer. + +To start a distributed estimation run: + +1. Connect the required RunServers from **Settings** > **RunServers**. +2. Open the model system and choose **Run Estimation**. +3. Select two or more RunServers in the run configuration dialog and choose the orchestrator RunServer. The other selected servers are workers. +4. If the model has an estimation `InputDirectory`, review its worker-local value for each selected worker. These values are saved in the GUI settings by model node ID and are reused on later estimation runs. + +The orchestrator RunServer owns the estimation algorithm and connects directly to the workers. Each worker constructs and validates its own local copy of the model. Input-directory overrides are applied only to that worker's copy; the saved model system is not modified. The orchestrator must be able to reach every worker's configured TCP endpoint. + +After submission, the GUI is not required for the optimization to continue. The orchestrator persists `estimation-completion.json` in the run directory when the job finishes, including the best parameter values and completion status. + +RunServers that receive path overrides must be running the current shared-estimation worker build. Runs without path overrides retain the version-1 shared-estimation protocol and remain compatible with workers that support the original distributed-estimation protocol. + ## Main Branches There are 4 major branches for XTMF 2 intended for different purposes: diff --git a/src/XTMF2.Client/Program.cs b/src/XTMF2.Client/Program.cs index ac9ec8c..5f10c64 100644 --- a/src/XTMF2.Client/Program.cs +++ b/src/XTMF2.Client/Program.cs @@ -179,6 +179,7 @@ private static void RunTcpServer(string address, int port, string? securityDirec Console.Out.Flush(); using (tcpListener) using (var shutdown = new CancellationTokenSource()) + using (var remoteEstimationRegistry = new RemoteSharedEstimationRegistry()) { Console.CancelKeyPress += (_, eventArgs) => { @@ -212,13 +213,14 @@ private static void RunTcpServer(string address, int port, string? securityDirec Console.Out.Flush(); var acceptedClient = client; _ = securityDirectory is null - ? Task.Run(() => RunTcpClientUnsecured(acceptedClient, extraDlls)) - : Task.Run(() => RunTcpClient(acceptedClient, certificate!, token!, extraDlls)); + ? Task.Run(() => RunTcpClientUnsecured(acceptedClient, extraDlls, remoteEstimationRegistry)) + : Task.Run(() => RunTcpClient(acceptedClient, certificate!, token!, extraDlls, remoteEstimationRegistry)); } } } - private static void RunTcpClientUnsecured(TcpClient client, List extraDlls) + private static void RunTcpClientUnsecured(TcpClient client, List extraDlls, + RemoteSharedEstimationRegistry remoteEstimationRegistry) { var remoteEndpoint = client.Client.RemoteEndPoint?.ToString() ?? "unknown endpoint"; Console.WriteLine($"Local RunServer connection accepted from {remoteEndpoint}"); @@ -228,7 +230,8 @@ private static void RunTcpClientUnsecured(TcpClient client, List extraDl { try { - RunClient(stream, extraDlls, usePrivateWorkspace: true); + RunClient(stream, extraDlls, usePrivateWorkspace: true, + remoteEstimationRegistry: remoteEstimationRegistry); } catch (Exception ex) { @@ -238,7 +241,8 @@ private static void RunTcpClientUnsecured(TcpClient client, List extraDl } } - private static void RunTcpClient(TcpClient client, X509Certificate2 certificate, string token, List extraDlls) + private static void RunTcpClient(TcpClient client, X509Certificate2 certificate, string token, List extraDlls, + RemoteSharedEstimationRegistry remoteEstimationRegistry) { var remoteEndpoint = client.Client.RemoteEndPoint?.ToString() ?? "unknown endpoint"; using (client) @@ -255,7 +259,8 @@ private static void RunTcpClient(TcpClient client, X509Certificate2 certificate, { try { - RunClient(stream, extraDlls, usePrivateWorkspace: true); + RunClient(stream, extraDlls, usePrivateWorkspace: true, + remoteEstimationRegistry: remoteEstimationRegistry); } catch (Exception ex) { @@ -268,7 +273,8 @@ private static void RunTcpClient(TcpClient client, X509Certificate2 certificate, } } - private static void RunClient(Stream serverStream, List extraDlls, SystemConfiguration? config = null, bool usePrivateWorkspace = false) + private static void RunClient(Stream serverStream, List extraDlls, SystemConfiguration? config = null, + bool usePrivateWorkspace = false, RemoteSharedEstimationRegistry? remoteEstimationRegistry = null) { var runtime = XTMFRuntime.CreateRuntime(config); var loadedConfig = runtime.SystemConfiguration; @@ -276,7 +282,13 @@ private static void RunClient(Stream serverStream, List extraDlls, Syste { loadedConfig.LoadAssembly(dll); } + using var ownedRemoteEstimationRegistry = remoteEstimationRegistry is null + ? new RemoteSharedEstimationRegistry() + : null; + var registry = remoteEstimationRegistry ?? ownedRemoteEstimationRegistry!; using var clientBus = new RunServerBus(serverStream, true, runtime, extraDlls, System.Diagnostics.Debugger.IsAttached, usePrivateWorkspace); + using var sharedEstimationWorker = clientBus.AttachSharedEstimationWorker(); + using var remoteCoordinator = new RemoteSharedEstimationCoordinatorSession(clientBus, registry); clientBus.ProcessRequests(); } } diff --git a/src/XTMF2.GUI/Properties/RunServerEndpoint.cs b/src/XTMF2.GUI/Properties/RunServerEndpoint.cs index 58fc97c..b965d52 100644 --- a/src/XTMF2.GUI/Properties/RunServerEndpoint.cs +++ b/src/XTMF2.GUI/Properties/RunServerEndpoint.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; namespace XTMF2.GUI.Properties; @@ -15,6 +16,7 @@ public sealed class RunServerEndpoint public bool IsLocal { get; set; } public string Token { get; set; } = string.Empty; public string CertificateFingerprint { get; set; } = string.Empty; + public Dictionary BasicParameterOverrides { get; set; } = new(StringComparer.OrdinalIgnoreCase); public static RunServerEndpoint CreateLocal() => new() @@ -26,7 +28,8 @@ public static RunServerEndpoint CreateLocal() Port = 0, IsLocal = true, Token = string.Empty, - CertificateFingerprint = string.Empty + CertificateFingerprint = string.Empty, + BasicParameterOverrides = new(StringComparer.OrdinalIgnoreCase) }; public RunServerEndpoint Clone() @@ -39,6 +42,8 @@ public RunServerEndpoint Clone() Port = Port, IsLocal = IsLocal, Token = Token, - CertificateFingerprint = CertificateFingerprint + CertificateFingerprint = CertificateFingerprint, + BasicParameterOverrides = new Dictionary(BasicParameterOverrides ?? new Dictionary(), + StringComparer.OrdinalIgnoreCase) }; } diff --git a/src/XTMF2.GUI/Properties/Settings.cs b/src/XTMF2.GUI/Properties/Settings.cs index 9dac72f..b046fb4 100644 --- a/src/XTMF2.GUI/Properties/Settings.cs +++ b/src/XTMF2.GUI/Properties/Settings.cs @@ -149,6 +149,8 @@ private static List NormalizeRunServers(IEnumerable 65535) continue; + endpoint.BasicParameterOverrides ??= new(StringComparer.OrdinalIgnoreCase); + if (normalized.Any(existing => existing.Id == endpoint.Id)) continue; diff --git a/src/XTMF2.GUI/RunController.cs b/src/XTMF2.GUI/RunController.cs index ad9a3c6..6488314 100644 --- a/src/XTMF2.GUI/RunController.cs +++ b/src/XTMF2.GUI/RunController.cs @@ -25,9 +25,11 @@ You should have received a copy of the GNU General Public License using System.Threading; using System.Threading.Tasks; using XTMF2.Bus; +using XTMF2.Bus.Optimization; using XTMF2.Editing; using XTMF2.GUI.ViewModels; using XTMF2.GUI.Properties; +using XTMF2.ModelSystemConstruct; using System.Linq; @@ -67,6 +69,11 @@ public class RunController : IDisposable private readonly Dictionary _sessionsByRunId = new(); private readonly Dictionary _hostBusesByRunId = new(); private readonly Dictionary _runDirectoriesByRunId = new(); + private readonly Dictionary> + _remoteEstimationMetadata = new(); + private readonly Dictionary> + _remoteWorkerEndpointsByRunId = new(); + private readonly HashSet _completedRemoteRuns = new(StringComparer.Ordinal); /// /// Fires when an estimation or calibration run completes and has results ready to be @@ -247,6 +254,51 @@ private void OnClientIterationProgressAvailable( RunsViewModel.NotifyIterationProgress(runID, iteration, fitness, values); } + private void OnSharedEstimationProgress(object? sender, SharedEstimationProgress progress) + { + RunsViewModel.NotifyStatus(progress.RunId, + $"[Remote estimation] iteration {progress.Iteration}: best fitness = {progress.BestFitness:G6}"); + } + + private void OnSharedEstimationWorkerControlAcknowledged( + object? sender, SharedEstimationWorkerControlAcknowledgement acknowledgement) + { + RunsViewModel.NotifyRemoteWorkerAcknowledgement(acknowledgement); + } + + private void OnSharedEstimationCompleted(object? sender, SharedEstimationCompletion completion) + { + lock (_sessionsByRunId) + { + if (!_completedRemoteRuns.Add(completion.RunId)) + return; + } + if (!completion.Succeeded) + { + RunsViewModel.NotifyError(completion.RunId, + completion.FailureReason ?? "Remote estimation failed.", String.Empty, null, null); + return; + } + + (ModelSystemSession Session, User User) entry; + IReadOnlyList<(int nodeIndex, string name, double min, double max)> metadata; + lock (_sessionsByRunId) + { + if (!_remoteEstimationMetadata.TryGetValue(completion.RunId, out var remoteMetadata)) + return; + if (!_sessionsByRunId.TryGetValue(completion.RunId, out entry) || + remoteMetadata is null) + return; + metadata = remoteMetadata; + } + if (completion.BestParameters.Count != metadata.Count) + return; + var results = metadata.Select((item, index) => (item.nodeIndex, completion.BestParameters[index])).ToArray(); + RunsViewModel.NotifyOptimizationResults(completion.RunId, entry.Session, entry.User, results); + OptimizationResultsAvailable?.Invoke(completion.RunId, entry.Session, results); + RunsViewModel.NotifyFinished(completion.RunId); + } + private void OnClientOptimizationResultsAvailable( object sender, string runID, IReadOnlyList<(int nodeIndex, double value)> results) { @@ -313,6 +365,278 @@ public bool SendEstimationRun( [NotNullWhen(false)] out CommandError? error) => SendRun(projectSession, msSession, user, startToExecute, runName, RunMode.Estimation, endpointId, out id, out error); + /// + /// Starts an estimation run across the selected connected RunServers. + /// + public bool SendSharedEstimationRun( + Project projectSession, + ModelSystemSession msSession, + User user, + string startToExecute, + string runName, + IReadOnlyList workerEndpointIds, + IReadOnlyDictionary>? basicParameterOverridesByWorker, + [NotNullWhen(true)] out string? id, + [NotNullWhen(false)] out CommandError? error) + { + id = null; + error = null; + if (String.IsNullOrWhiteSpace(projectSession.ProjectDirectory)) + { + error = new CommandError("Project directory is not set."); + return false; + } + if (workerEndpointIds.Count == 0) + { + error = new CommandError("At least one shared-estimation worker is required."); + return false; + } + + var modelSystem = msSession.ModelSystem; + var enabledEntries = modelSystem.EstimationGroups + .Where(group => group.IsEnabled) + .SelectMany(group => group.Parameters) + .Where(entry => entry.IsEnabled) + .ToArray(); + var metadata = msSession.GetOptimizationParameterMeta(RunMode.Estimation); + if (enabledEntries.Length == 0 || enabledEntries.Length != metadata.Count) + { + error = new CommandError("Estimation requires at least one enabled estimation parameter."); + return false; + } + if (modelSystem.EstimationFitnessNode is null) + { + error = new CommandError("Estimation requires a fitness node."); + return false; + } + + var parameterEntries = enabledEntries + .Select((entry, index) => (entry, nodeIndex: metadata[index].nodeIndex)) + .ToArray(); + + var runId = Guid.NewGuid().ToString(); + var runDirectory = Path.Combine(projectSession.ProjectDirectory, "runs", runName); + using var modelStream = new MemoryStream(); + if (!msSession.Save(out error, modelStream)) + return false; + + var pool = new SharedEstimationWorkerPool(); + foreach (var endpointId in workerEndpointIds.Distinct(StringComparer.Ordinal)) + { + if (!_connections.TryGet(endpointId, out var hostBus) || hostBus is null) + { + pool.Dispose(); + error = new CommandError($"RunServer '{endpointId}' is not connected."); + return false; + } + if (!pool.AddExistingWorker(endpointId, endpointId, hostBus, out var workerError)) + { + pool.Dispose(); + error = new CommandError(workerError ?? $"Unable to add RunServer '{endpointId}'."); + return false; + } + } + + var lower = parameterEntries.Select(item => item.entry.Min).ToArray(); + var upper = parameterEntries.Select(item => item.entry.Max).ToArray(); + var initial = parameterEntries.Select(item => item.entry.NullHypothesis).ToArray(); + var algorithm = modelSystem.EstimationAlgorithmConfig.CreateAlgorithm( + parameterEntries.Length, lower, upper, initial, + modelSystem.EstimationObjective == EstimationObjective.Maximize); + var request = new SharedEstimationRunRequest( + runId, runDirectory, startToExecute, modelStream.ToArray()); + if (!pool.StartRun(request, out var startError, basicParameterOverridesByWorker)) + { + pool.Dispose(); + error = new CommandError(startError ?? "Unable to start shared estimation."); + return false; + } + + var serverLabel = $"Shared ({workerEndpointIds.Count} workers)"; + var runViewModel = RunsViewModel.AddRun(runId, runName, runDirectory, serverLabel, msSession, user); + var cancellation = new CancellationTokenSource(); + runViewModel.SetRunMode(RunMode.Estimation, metadata, () => + { + cancellation.Cancel(); + pool.CancelRun("Cancelled by user."); + }); + lock (_sessionsByRunId) + _sessionsByRunId[runId] = (msSession, user); + + _ = Task.Run(() => + { + try + { + var runner = new SharedEstimationCoordinatorRun(runId, algorithm, pool.Coordinator); + var completion = runner.Execute( + progress: progress => RunsViewModel.NotifyStatus(runId, + $"[Shared estimation] iteration {progress.Iteration}: best fitness = {progress.BestFitness:G6}"), + cancellationToken: cancellation.Token); + if (!completion.Succeeded) + { + RunsViewModel.NotifyError(runId, completion.FailureReason ?? "Shared estimation failed.", + String.Empty, null, null); + return; + } + + var results = parameterEntries + .Select((item, index) => (nodeIndex: item.nodeIndex, value: completion.BestParameters[index])) + .ToArray(); + RunsViewModel.NotifyOptimizationResults(runId, msSession, user, results); + OptimizationResultsAvailable?.Invoke(runId, msSession, results); + RunsViewModel.NotifyFinished(runId); + } + finally + { + cancellation.Dispose(); + pool.Dispose(); + lock (_sessionsByRunId) + _sessionsByRunId.Remove(runId); + } + }); + + id = runId; + return true; + } + + public bool SendRemoteSharedEstimationRun( + Project projectSession, + ModelSystemSession msSession, + User user, + string startToExecute, + string runName, + string orchestratorEndpointId, + IReadOnlyList workerEndpointIds, + IReadOnlyDictionary>? basicParameterOverridesByWorker, + [NotNullWhen(true)] out string? id, + [NotNullWhen(false)] out CommandError? error) + { + id = null; + error = null; + if (!_connections.TryGet(orchestratorEndpointId, out var orchestrator) || orchestrator is null) + { + error = new CommandError($"RunServer '{orchestratorEndpointId}' is not connected."); + return false; + } + var modelSystem = msSession.ModelSystem; + var enabledEntries = modelSystem.EstimationGroups + .Where(group => group.IsEnabled) + .SelectMany(group => group.Parameters) + .Where(entry => entry.IsEnabled) + .ToArray(); + var metadata = msSession.GetOptimizationParameterMeta(RunMode.Estimation); + if (enabledEntries.Length == 0 || enabledEntries.Length != metadata.Count) + { + error = new CommandError("Estimation requires at least one enabled estimation parameter."); + return false; + } + if (modelSystem.EstimationFitnessNode is null) + { + error = new CommandError("Estimation requires a fitness node."); + return false; + } + + var endpoints = GetConnectedRunServers() + .Where(endpoint => workerEndpointIds.Contains(endpoint.Id, StringComparer.Ordinal) && + endpoint.Id != orchestratorEndpointId) + .ToArray(); + if (endpoints.Length == 0) + { + error = new CommandError("Select at least one worker RunServer in addition to the orchestrator."); + return false; + } + + using var modelStream = new MemoryStream(); + if (!msSession.Save(out error, modelStream)) + return false; + var entries = enabledEntries + .Select((entry, index) => (entry, nodeIndex: metadata[index].nodeIndex)) + .ToArray(); + var request = new SharedEstimationCoordinatorRequest( + new SharedEstimationRunRequest(Guid.NewGuid().ToString(), + Path.Combine(projectSession.ProjectDirectory!, "runs", runName), + startToExecute, modelStream.ToArray()), + endpoints.Select(endpoint => new SharedEstimationWorkerEndpoint( + endpoint.Id, endpoint.Id, endpoint.Address, endpoint.Port, endpoint.Token, + endpoint.CertificateFingerprint, + basicParameterOverridesByWorker is not null && + basicParameterOverridesByWorker.TryGetValue(endpoint.Id, out var overrides) + ? overrides + : null)).ToArray(), + modelSystem.EstimationAlgorithmConfig.AlgorithmId, + modelSystem.EstimationAlgorithmConfig.GetParameters(), + entries.Select(item => item.entry.Min).ToArray(), + entries.Select(item => item.entry.Max).ToArray(), + entries.Select(item => item.entry.NullHypothesis).ToArray(), + modelSystem.EstimationObjective == EstimationObjective.Maximize); + + if (!orchestrator.StartRemoteSharedEstimation(request, out error)) + return false; + + id = request.Run.RunId; + var submittedRunId = request.Run.RunId; + var runDirectory = request.Run.WorkingDirectory; + var runViewModel = RunsViewModel.AddRun(id, runName, runDirectory, + $"{orchestratorEndpointId} (orchestrator, {endpoints.Length} workers)", msSession, user); + runViewModel.SetRunMode(RunMode.Estimation, metadata, () => + { + orchestrator.CancelSharedEstimation(submittedRunId, "Cancelled by user.", out _); + }); + var initialWorkers = request.Workers.ToDictionary(worker => worker.WorkerId, + worker => worker, StringComparer.Ordinal); + var availableWorkers = GetConnectedRunServers() + .Where(endpoint => endpoint.Id != orchestratorEndpointId) + .Select(endpoint => initialWorkers.TryGetValue(endpoint.Id, out var initial) + ? initial + : new SharedEstimationWorkerEndpoint(endpoint.Id, endpoint.Id, endpoint.Address, + endpoint.Port, endpoint.Token, endpoint.CertificateFingerprint)) + .ToArray(); + lock (_sessionsByRunId) + { + _hostBusesByRunId[submittedRunId] = orchestrator; + _remoteWorkerEndpointsByRunId[submittedRunId] = availableWorkers.ToDictionary(worker => worker.WorkerId, + worker => worker, StringComparer.Ordinal); + } + runViewModel.SetRemoteEstimationWorkers( + availableWorkers.Select(ToRunServerEndpoint).ToArray(), + initialWorkers.Keys.ToArray(), + workerId => ChangeRemoteEstimationWorker(submittedRunId, workerId, add: true), + workerId => ChangeRemoteEstimationWorker(submittedRunId, workerId, add: false)); + lock (_sessionsByRunId) + _sessionsByRunId[id] = (msSession, user); + lock (_sessionsByRunId) + _remoteEstimationMetadata[id] = metadata; + return true; + } + + private static RunServerEndpoint ToRunServerEndpoint(SharedEstimationWorkerEndpoint endpoint) + => new() + { + Id = endpoint.EndpointId, + Name = Settings.Default.RunServers.FirstOrDefault(candidate => + string.Equals(candidate.Id, endpoint.EndpointId, StringComparison.Ordinal))?.Name + ?? endpoint.EndpointId, + Address = endpoint.Address, + Port = endpoint.Port, + Token = endpoint.Token, + CertificateFingerprint = endpoint.CertificateFingerprint + }; + + private string? ChangeRemoteEstimationWorker(string runId, string workerId, bool add) + { + lock (_sessionsByRunId) + { + if (!_hostBusesByRunId.TryGetValue(runId, out var bus) || + !_remoteWorkerEndpointsByRunId.TryGetValue(runId, out var workers) || + !workers.TryGetValue(workerId, out var worker)) + return "The remote estimation connection or worker is unavailable."; + var sent = add + ? bus.AddRemoteEstimationWorker(runId, worker, out var error) + : bus.RemoveRemoteEstimationWorker(runId, worker, out error); + return sent ? null : error?.Message ?? "Unable to send the worker change."; + } + } + /// /// Sends a calibration run (proportional-update loop) to the client process. /// @@ -498,8 +822,27 @@ private void SubscribeToHostBus(HostBus hostBus) hostBus.ClientErrorWhenRunningModelSystem += OnClientErrorWhenRunningModelSystem; hostBus.ClientOptimizationResultsAvailable += OnClientOptimizationResultsAvailable; hostBus.ClientIterationProgressAvailable += OnClientIterationProgressAvailable; + hostBus.SharedEstimationProgressAvailable += OnSharedEstimationProgress; + hostBus.SharedEstimationCompleted += OnSharedEstimationCompleted; + hostBus.SharedEstimationWorkerControlAcknowledged += OnSharedEstimationWorkerControlAcknowledged; + hostBus.SharedEstimationJobSnapshotsAvailable += OnSharedEstimationJobSnapshotsAvailable; hostBus.ClientRunArtifactsReceived += OnClientRunArtifactsReceived; hostBus.Disconnected += OnHostBusDisconnected; + hostBus.QuerySharedEstimationJobs(out _); + } + + private void OnSharedEstimationJobSnapshotsAvailable( + object? sender, IReadOnlyList snapshots) + { + foreach (var snapshot in snapshots) + { + if (snapshot.ActiveWorkerIds is not null) + RunsViewModel.NotifyRemoteWorkerSnapshot(snapshot.RunId, snapshot.ActiveWorkerIds); + if (snapshot.Progress is not null) + OnSharedEstimationProgress(sender, snapshot.Progress); + if (snapshot.Completion is not null) + OnSharedEstimationCompleted(sender, snapshot.Completion); + } } private void OnHostBusDisconnected(object? sender, EventArgs e) @@ -521,6 +864,10 @@ private void UnsubscribeFromHostBus(HostBus hostBus) hostBus.ClientErrorWhenRunningModelSystem -= OnClientErrorWhenRunningModelSystem; hostBus.ClientOptimizationResultsAvailable -= OnClientOptimizationResultsAvailable; hostBus.ClientIterationProgressAvailable -= OnClientIterationProgressAvailable; + hostBus.SharedEstimationProgressAvailable -= OnSharedEstimationProgress; + hostBus.SharedEstimationCompleted -= OnSharedEstimationCompleted; + hostBus.SharedEstimationWorkerControlAcknowledged -= OnSharedEstimationWorkerControlAcknowledged; + hostBus.SharedEstimationJobSnapshotsAvailable -= OnSharedEstimationJobSnapshotsAvailable; hostBus.ClientRunArtifactsReceived -= OnClientRunArtifactsReceived; hostBus.Disconnected -= OnHostBusDisconnected; } diff --git a/src/XTMF2.GUI/ViewModels/ModelSystemEditorViewModel.cs b/src/XTMF2.GUI/ViewModels/ModelSystemEditorViewModel.cs index 4626118..a910052 100644 --- a/src/XTMF2.GUI/ViewModels/ModelSystemEditorViewModel.cs +++ b/src/XTMF2.GUI/ViewModels/ModelSystemEditorViewModel.cs @@ -2759,6 +2759,25 @@ public async Task RemoveNodeFromCalibrationAsync(NodeViewModel nvm) return CollectFunctionNodes(Session.ModelSystem.GlobalBoundary); } + public System.Collections.Generic.List GetInputDirectoryNodes() + { + return CollectInputDirectoryNodes(Session.ModelSystem.GlobalBoundary); + } + + private static System.Collections.Generic.List CollectInputDirectoryNodes( + XTMF2.ModelSystemConstruct.Boundary boundary) + { + var result = new System.Collections.Generic.List(); + foreach (var node in boundary.Modules) + { + if (node.Type == typeof(XTMF2.RuntimeModules.BasicParameter)) + result.Add(node); + } + foreach (var child in boundary.Boundaries) + result.AddRange(CollectInputDirectoryNodes(child)); + return result; + } + private static System.Collections.Generic.List CollectFunctionNodes( XTMF2.ModelSystemConstruct.Boundary boundary) { @@ -3502,10 +3521,10 @@ private async Task RunModelSystem() defaultRunName, availableStarts.Select(start => start.Name).ToList()); if (runConfiguration is null) return; - var (runName, endpointId, startToExecute) = runConfiguration.Value; + var (runName, endpointIds, startToExecute, _, _) = runConfiguration.Value; var project = Session.Project; - if (!_runController.SendRun(project, Session, User, startToExecute, runName, endpointId, out _, out var runError)) + if (!_runController.SendRun(project, Session, User, startToExecute, runName, endpointIds[0], out _, out var runError)) { ShowToast($"Failed to start run: {runError?.Message}", isError: true, durationMs: 6000); return; @@ -3529,15 +3548,31 @@ private async Task RunEstimation() } var defaultRunName = $"Estimation_{ModelSystemHeader.Name ?? "Run"}_{DateTime.Now:yyyyMMdd_HHmmss}"; + var selectedInputDirectoryId = Session.ModelSystem.EstimationInputDirectoryNode?.Id; + var inputDirectory = Session.GetStringBasicParameterMeta() + .Where(parameter => selectedInputDirectoryId is not null && parameter.nodeId == selectedInputDirectoryId) + .ToList(); + var inputDirectoryParameters = inputDirectory + .Select(parameter => new RunConfigurationDialog.PathParameter( + parameter.nodeIndex, parameter.nodeId, parameter.name, parameter.value)) + .ToList(); var runConfiguration = await ConfigureRunAsync( "Run Estimation", defaultRunName, - availableStarts.Select(start => start.Name).ToList()); + availableStarts.Select(start => start.Name).ToList(), + allowMultipleRunServers: true, + inputDirectoryParameters); if (runConfiguration is null) return; - var (runName, endpointId, startToExecute) = runConfiguration.Value; + var (runName, endpointIds, startToExecute, pathOverrides, orchestratorEndpointId) = runConfiguration.Value; var project = Session.Project; - if (!_runController.SendEstimationRun(project, Session, User, startToExecute, runName, endpointId, out _, out var runError)) + CommandError? runError; + bool started = endpointIds.Count > 1 && orchestratorEndpointId is not null + ? _runController.SendRemoteSharedEstimationRun(project, Session, User, startToExecute, runName, + orchestratorEndpointId, endpointIds, pathOverrides, out _, out runError) + : _runController.SendEstimationRun(project, Session, User, startToExecute, runName, + endpointIds[0], out _, out runError); + if (!started) { ShowToast($"Failed to start estimation run: {runError?.Message}", isError: true, durationMs: 6000); return; @@ -3566,10 +3601,10 @@ private async Task RunCalibration() defaultRunName, availableStarts.Select(start => start.Name).ToList()); if (runConfiguration is null) return; - var (runName, endpointId, startToExecute) = runConfiguration.Value; + var (runName, endpointIds, startToExecute, _, _) = runConfiguration.Value; var project = Session.Project; - if (!_runController.SendCalibrationRun(project, Session, User, startToExecute, runName, endpointId, out _, out var runError)) + if (!_runController.SendCalibrationRun(project, Session, User, startToExecute, runName, endpointIds[0], out _, out var runError)) { ShowToast($"Failed to start calibration run: {runError?.Message}", isError: true, durationMs: 6000); return; @@ -3579,10 +3614,14 @@ private async Task RunCalibration() RunStarted?.Invoke(); } - private async Task<(string RunName, string EndpointId, string StartName)?> ConfigureRunAsync( + private async Task<(string RunName, IReadOnlyList EndpointIds, string StartName, + IReadOnlyDictionary> PathOverrides, + string? OrchestratorEndpointId)?> ConfigureRunAsync( string title, string defaultRunName, - IReadOnlyList startNames) + IReadOnlyList startNames, + bool allowMultipleRunServers = false, + IReadOnlyList? pathParameters = null) { if (_runController is null) return null; @@ -3594,17 +3633,19 @@ private async Task RunCalibration() return null; } - var dialog = new RunConfigurationDialog(title, defaultRunName, endpoints, startNames); + var dialog = new RunConfigurationDialog(title, defaultRunName, endpoints, startNames, + allowMultipleRunServers, pathParameters); await dialog.ShowDialog(ParentWindow!); if (dialog.WasCancelled) return null; var runName = dialog.RunName?.Trim(); - var endpointId = dialog.SelectedRunServer?.Id; + var endpointIds = dialog.SelectedRunServers.Select(endpoint => endpoint.Id).ToArray(); var startName = dialog.SelectedStartName; - return string.IsNullOrEmpty(runName) || endpointId is null || string.IsNullOrEmpty(startName) + var orchestratorEndpointId = allowMultipleRunServers ? dialog.SelectedCoordinatorRunServer?.Id : null; + return string.IsNullOrEmpty(runName) || endpointIds.Length == 0 || string.IsNullOrEmpty(startName) ? null - : (runName, endpointId, startName); + : (runName, endpointIds, startName, dialog.PathOverrides, orchestratorEndpointId); } /// Save the model system to its project file. diff --git a/src/XTMF2.GUI/ViewModels/RemoteEstimationWorkerViewModel.cs b/src/XTMF2.GUI/ViewModels/RemoteEstimationWorkerViewModel.cs new file mode 100644 index 0000000..e34c2ac --- /dev/null +++ b/src/XTMF2.GUI/ViewModels/RemoteEstimationWorkerViewModel.cs @@ -0,0 +1,36 @@ +using CommunityToolkit.Mvvm.ComponentModel; +using XTMF2.GUI.Properties; + +namespace XTMF2.GUI.ViewModels; + +public sealed partial class RemoteEstimationWorkerViewModel : ObservableObject +{ + public RunServerEndpoint Endpoint { get; } + public string WorkerId => Endpoint.Id; + public string Name => Endpoint.Name; + + [ObservableProperty] + private bool _isActive; + + [ObservableProperty] + private bool _isBusy; + + public bool CanRemove => IsActive; + public bool IsInactive => !IsActive; + public bool IsNotBusy => !IsBusy; + + partial void OnIsActiveChanged(bool value) + { + OnPropertyChanged(nameof(CanRemove)); + OnPropertyChanged(nameof(IsInactive)); + } + + partial void OnIsBusyChanged(bool value) + => OnPropertyChanged(nameof(IsNotBusy)); + + public RemoteEstimationWorkerViewModel(RunServerEndpoint endpoint, bool isActive) + { + Endpoint = endpoint; + IsActive = isActive; + } +} diff --git a/src/XTMF2.GUI/ViewModels/RunViewModel.cs b/src/XTMF2.GUI/ViewModels/RunViewModel.cs index 643af33..5788544 100644 --- a/src/XTMF2.GUI/ViewModels/RunViewModel.cs +++ b/src/XTMF2.GUI/ViewModels/RunViewModel.cs @@ -24,6 +24,7 @@ You should have received a copy of the GNU General Public License using System.Linq; using System.Runtime.InteropServices; using XTMF2.Configuration; +using XTMF2.GUI.Properties; using CommunityToolkit.Mvvm.ComponentModel; using CommunityToolkit.Mvvm.Input; using XTMF2.Bus; @@ -225,6 +226,11 @@ private static bool TryOpenDirectoryInFileExplorer(string directoryPath) /// Complete history of all completed iterations (oldest first). public ObservableCollection IterationHistory { get; } = new(); + public ObservableCollection RemoteWorkers { get; } = new(); + + [ObservableProperty] + private bool _isRemoteSharedEstimation; + /// The iteration currently shown in the detail panel; auto-advances with each new iteration. [ObservableProperty] private OptimizationIterationViewModel? _selectedIteration; @@ -251,6 +257,74 @@ internal void SetRunMode(RunMode runMode, CancelRunCommand.NotifyCanExecuteChanged(); } + internal void SetRemoteEstimationWorkers( + IReadOnlyList endpoints, + IReadOnlyCollection activeWorkerIds, + Func addWorker, + Func removeWorker) + { + IsRemoteSharedEstimation = true; + _addRemoteWorkerAction = addWorker; + _removeRemoteWorkerAction = removeWorker; + RemoteWorkers.Clear(); + foreach (var endpoint in endpoints) + RemoteWorkers.Add(new RemoteEstimationWorkerViewModel(endpoint, + activeWorkerIds.Contains(endpoint.Id, StringComparer.Ordinal))); + } + + internal void ApplyRemoteWorkerSnapshot(IReadOnlyCollection activeWorkerIds) + { + foreach (var worker in RemoteWorkers) + { + worker.IsActive = activeWorkerIds.Contains(worker.WorkerId, StringComparer.Ordinal); + worker.IsBusy = false; + } + } + + internal void ApplyRemoteWorkerAcknowledgement(SharedEstimationWorkerControlAcknowledgement acknowledgement) + { + var worker = RemoteWorkers.FirstOrDefault(candidate => + string.Equals(candidate.WorkerId, acknowledgement.WorkerId, StringComparison.Ordinal)); + if (worker is null) + return; + worker.IsBusy = false; + if (acknowledgement.Succeeded) + worker.IsActive = acknowledgement.Add; + else + AppendStatus($"[Remote estimation] Worker change failed: {acknowledgement.Error}"); + } + + private Func? _addRemoteWorkerAction; + private Func? _removeRemoteWorkerAction; + + [RelayCommand] + private void AddRemoteWorker(RemoteEstimationWorkerViewModel worker) + { + if (worker.IsActive || worker.IsBusy || _addRemoteWorkerAction is null) + return; + worker.IsBusy = true; + var error = _addRemoteWorkerAction(worker.WorkerId); + if (error is not null) + { + worker.IsBusy = false; + AppendStatus($"[Remote estimation] Unable to add worker: {error}"); + } + } + + [RelayCommand] + private void RemoveRemoteWorker(RemoteEstimationWorkerViewModel worker) + { + if (!worker.IsActive || worker.IsBusy || _removeRemoteWorkerAction is null) + return; + worker.IsBusy = true; + var error = _removeRemoteWorkerAction(worker.WorkerId); + if (error is not null) + { + worker.IsBusy = false; + AppendStatus($"[Remote estimation] Unable to remove worker: {error}"); + } + } + /// Called on the UI thread whenever the client sends per-iteration progress. internal void UpdateIterationProgress(int iteration, double fitness, IReadOnlyList<(int nodeIndex, double value)> values) { diff --git a/src/XTMF2.GUI/ViewModels/RunsViewModel.cs b/src/XTMF2.GUI/ViewModels/RunsViewModel.cs index 8be9faf..f7e162d 100644 --- a/src/XTMF2.GUI/ViewModels/RunsViewModel.cs +++ b/src/XTMF2.GUI/ViewModels/RunsViewModel.cs @@ -23,6 +23,7 @@ You should have received a copy of the GNU General Public License using Avalonia.Threading; using CommunityToolkit.Mvvm.ComponentModel; using CommunityToolkit.Mvvm.Input; +using XTMF2.Bus; using XTMF2.Editing; namespace XTMF2.GUI.ViewModels; @@ -158,6 +159,21 @@ internal void NotifyIterationProgress( Dispatcher.UIThread.Post(() => vm.UpdateIterationProgress(iteration, fitness, values)); } + internal void NotifyRemoteWorkerSnapshot(string runId, IReadOnlyCollection activeWorkerIds) + { + var vm = FindRun(runId); + if (vm is null) return; + Dispatcher.UIThread.Post(() => vm.ApplyRemoteWorkerSnapshot(activeWorkerIds)); + } + + internal void NotifyRemoteWorkerAcknowledgement( + SharedEstimationWorkerControlAcknowledgement acknowledgement) + { + var vm = FindRun(acknowledgement.RunId); + if (vm is null) return; + Dispatcher.UIThread.Post(() => vm.ApplyRemoteWorkerAcknowledgement(acknowledgement)); + } + private RunViewModel? FindRun(string runId) { lock (_runsLock) diff --git a/src/XTMF2.GUI/Views/EstimationDialog.axaml b/src/XTMF2.GUI/Views/EstimationDialog.axaml index ea0732b..32fc7e0 100644 --- a/src/XTMF2.GUI/Views/EstimationDialog.axaml +++ b/src/XTMF2.GUI/Views/EstimationDialog.axaml @@ -52,6 +52,22 @@