From 3ef62cee4554e86229d5c78db314f7f35e9d82f2 Mon Sep 17 00:00:00 2001 From: James Vaughan Date: Tue, 6 Oct 2026 16:20:22 -0400 Subject: [PATCH] RunServer's Runqueue should have a single Run Queue instead of one per client --- src/XTMF2.Client/Program.cs | 31 ++++---- src/XTMF2/Bus/RemoteRunRegistry.cs | 72 +++++++++++++++++ src/XTMF2/Bus/RunServerBus.cs | 2 + .../Bus/TestRemoteRunRegistry.cs | 78 +++++++++++++++++++ 4 files changed, 169 insertions(+), 14 deletions(-) diff --git a/src/XTMF2.Client/Program.cs b/src/XTMF2.Client/Program.cs index 10da6f7..fb37024 100644 --- a/src/XTMF2.Client/Program.cs +++ b/src/XTMF2.Client/Program.cs @@ -122,7 +122,12 @@ static void Main(string[] args) Console.WriteLine("Unable to create a connection to the host!"); return; } - RunClient(serverStream, dllsToLoad); + using (var remoteRunRegistry = new RemoteRunRegistry()) + { + RunClient(serverStream, dllsToLoad, config: null, + usePrivateWorkspace: false, remoteEstimationRegistry: null, + remoteRunRegistry: remoteRunRegistry, allowDeployment: false); + } } finally { @@ -250,10 +255,9 @@ private static void RunTcpClientUnsecured(TcpClient client, List extraDl { try { - RunClient(stream, extraDlls, usePrivateWorkspace: true, - allowDeployment: false, + RunClient(stream, extraDlls, config: null, usePrivateWorkspace: true, remoteEstimationRegistry: remoteEstimationRegistry, - remoteRunRegistry: remoteRunRegistry); + remoteRunRegistry: remoteRunRegistry, allowDeployment: false); } catch (Exception ex) { @@ -282,10 +286,9 @@ private static void RunTcpClient(TcpClient client, X509Certificate2 certificate, { try { - RunClient(stream, extraDlls, usePrivateWorkspace: true, - allowDeployment: true, + RunClient(stream, extraDlls, config: null, usePrivateWorkspace: true, remoteEstimationRegistry: remoteEstimationRegistry, - remoteRunRegistry: remoteRunRegistry); + remoteRunRegistry: remoteRunRegistry, allowDeployment: true); } catch (Exception ex) { @@ -298,9 +301,9 @@ private static void RunTcpClient(TcpClient client, X509Certificate2 certificate, } } - private static void RunClient(Stream serverStream, List extraDlls, SystemConfiguration? config = null, - bool usePrivateWorkspace = false, RemoteSharedEstimationRegistry? remoteEstimationRegistry = null, - RemoteRunRegistry? remoteRunRegistry = null, bool allowDeployment = false) + private static void RunClient(Stream serverStream, List extraDlls, SystemConfiguration? config, + bool usePrivateWorkspace, RemoteSharedEstimationRegistry? remoteEstimationRegistry, + RemoteRunRegistry remoteRunRegistry, bool allowDeployment) { var runtimeStopwatch = Stopwatch.StartNew(); LogStartup("Initializing the XTMF runtime for a client connection."); @@ -325,16 +328,16 @@ private static void RunClient(Stream serverStream, List extraDlls, Syste clientBus.SetDeploymentGate( () => { - remoteRunRegistry?.BeginDrain(); + remoteRunRegistry.BeginDrain(); registry.BeginDrain(); }, () => { - remoteRunRegistry?.EndDrain(); + remoteRunRegistry.EndDrain(); registry.EndDrain(); }, () => - (remoteRunRegistry?.IsIdle ?? true) && registry.IsIdle, + remoteRunRegistry.IsIdle && registry.IsIdle, RestartWithOriginalArguments); clientBus.SetSharedActivityProviders(registry.GetActiveActivities, sharedEstimationWorker.GetActiveActivities); @@ -344,7 +347,7 @@ private static void RunClient(Stream serverStream, List extraDlls, Syste } finally { - remoteRunRegistry?.Detach(clientBus); + remoteRunRegistry.Detach(clientBus); } } diff --git a/src/XTMF2/Bus/RemoteRunRegistry.cs b/src/XTMF2/Bus/RemoteRunRegistry.cs index 413be4c..51b3638 100644 --- a/src/XTMF2/Bus/RemoteRunRegistry.cs +++ b/src/XTMF2/Bus/RemoteRunRegistry.cs @@ -59,14 +59,27 @@ public sealed class RemoteRunRegistry : IDisposable private readonly string _storageDirectory; private bool _disposed; + /// + /// Gets whether the RunServer scheduler has no active or queued work. + /// public bool IsIdle => _scheduler.IsIdle; + /// + /// Stops the registry from accepting new work while existing work drains. + /// public void BeginDrain() => _scheduler.BeginDrain(); + /// + /// Allows the registry to accept work after a drain has been cancelled. + /// public void EndDrain() => _scheduler.EndDrain(); + /// + /// Creates a registry that owns remote run state and the process-wide RunServer scheduler. + /// + /// The directory for persisted run manifests and artifacts; when , the default per-user XTMF2 directory is used. public RemoteRunRegistry(string? storageDirectory = null) { _storageDirectory = storageDirectory ?? Path.Combine( @@ -76,6 +89,20 @@ public RemoteRunRegistry(string? storageDirectory = null) LoadPersistedJobs(); } + /// + /// Registers a remote run and queues it on this registry's scheduler. + /// + /// The connected RunServer bus that receives run output and completion notifications. + /// The prepared run context to execute. + /// The display name of the run. + /// The execution mode of the run. + /// The run's private working directory. + /// The model-system start point to execute. + /// The serialized model system, used to identify the submitted run. + /// The identifier of the owning project, if known. + /// The identifier of the model system, if known. + /// The identifier of the owning user, if known. + /// if the run was registered and queued; otherwise, if the registry is draining or the run ID is already registered. public bool Submit(RunServerBus observer, RunContext context, string runName, RunMode runMode, string workingDirectory, string startToExecute, byte[] modelSystem, Guid? projectId = null, Guid? modelSystemId = null, Guid? ownerUserId = null) @@ -107,6 +134,10 @@ public bool Submit(RunServerBus observer, RunContext context, string runName, internal Scheduler.ReservationLease ReserveWorkerSlot(RunContext context, int expectedSlotCount) => _scheduler.ReserveGroup(context.ID, context, expectedSlotCount); + /// + /// Attaches a connected bus to retained remote runs so it can receive their subsequent updates. + /// + /// The bus to attach as a run observer. public void Attach(RunServerBus observer) { lock (_sync) @@ -116,6 +147,10 @@ public void Attach(RunServerBus observer) } } + /// + /// Detaches a bus from retained remote runs without cancelling the runs. + /// + /// The bus to detach from run notifications. public void Detach(RunServerBus observer) { lock (_sync) @@ -125,12 +160,20 @@ public void Detach(RunServerBus observer) } } + /// + /// Gets snapshots of all retained remote runs. + /// + /// A snapshot of each run currently retained by the registry. public IReadOnlyList GetSnapshots() { lock (_sync) return _jobs.Values.Select(job => job.GetSnapshot()).ToArray(); } + /// + /// Gets the active and queued run and worker-reservation activities in the shared scheduler. + /// + /// The scheduler's current activity inventory. public IReadOnlyList GetActiveActivities() { lock (_sync) @@ -154,6 +197,12 @@ public IReadOnlyList GetActiveActivities() } } + /// + /// Reads a run's retained artifact archive, rebuilding it from the private run directory when necessary. + /// + /// The identifier of the run whose artifacts are requested. + /// Receives the archive bytes when available; otherwise, . + /// if the artifact archive was read successfully; otherwise, . public bool TryReadArtifacts(string runId, out byte[]? archive) { RemoteRunJob? job; @@ -197,6 +246,11 @@ public bool TryReadArtifacts(string runId, out byte[]? archive) } } + /// + /// Acknowledges receipt of a completed run's artifacts and removes the retained archive. + /// + /// The identifier of the run whose artifact receipt is acknowledged. + /// if the run exists and is no longer running; otherwise, . public bool AcknowledgeReceived(string runId) { RemoteRunJob? job; @@ -237,6 +291,12 @@ private static bool IsPrivateRunDirectory(string runId, string workingDirectory) } } + /// + /// Deletes a completed or interrupted run and its files from the registry's private workspace. + /// + /// The identifier of the run to delete. + /// Receives an explanation if deletion fails; otherwise, . + /// if the run and its files were deleted; otherwise, . public bool DeleteRun(string runId, out string? error) { error = null; @@ -288,6 +348,10 @@ public bool DeleteRun(string runId, out string? error) } } + /// + /// Requests cancellation of a queued or running remote run. + /// + /// The identifier of the run to cancel. public void Cancel(string runId) { lock (_sync) @@ -297,9 +361,17 @@ public void Cancel(string runId) } } + /// + /// Kills a queued or running remote run, removing queued work or terminating the active run as applicable. + /// + /// The identifier of the run to kill. + /// if a matching run was found and handled; otherwise, . public bool Kill(string runId) => _scheduler.Kill(runId); + /// + /// Stops the scheduler and releases resources owned by this registry. + /// public void Dispose() { lock (_sync) diff --git a/src/XTMF2/Bus/RunServerBus.cs b/src/XTMF2/Bus/RunServerBus.cs index d566f5f..d8ba502 100644 --- a/src/XTMF2/Bus/RunServerBus.cs +++ b/src/XTMF2/Bus/RunServerBus.cs @@ -108,6 +108,8 @@ public RunServerBus(Stream serverStream, bool streamOwner, XTMFRuntime runtime, bool allowDeployment = false) { Runtime = runtime; + // Standalone debug/test buses own their scheduler. Production RunClient sessions + // receive a process-owned registry so independent host connections share one queue. _remoteRunRegistry = remoteRunRegistry; _runScheduler = remoteRunRegistry is null ? new Scheduler(this, runLocal) : null; _clientHost = serverStream; diff --git a/tests/XTMF2.UnitTests/Bus/TestRemoteRunRegistry.cs b/tests/XTMF2.UnitTests/Bus/TestRemoteRunRegistry.cs index 40b5c17..27cdfc4 100644 --- a/tests/XTMF2.UnitTests/Bus/TestRemoteRunRegistry.cs +++ b/tests/XTMF2.UnitTests/Bus/TestRemoteRunRegistry.cs @@ -1,17 +1,91 @@ using System; using System.IO; +using System.Linq; using System.Security.Cryptography; using System.Text; using System.Text.Json; using System.Text.Json.Serialization; +using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using XTMF2.Bus; +using XTMF2.Bus.Optimization; namespace XTMF2.UnitTests.Bus; [TestClass] public class TestRemoteRunRegistry { + [TestMethod] + public async Task SharedWorkerReservation_BlocksOrdinaryRunsFromIndependentObservers() + { + var directory = Path.Combine(Path.GetTempPath(), "XTMF2-RemoteRunRegistry-" + Guid.NewGuid().ToString("N")); + Directory.CreateDirectory(directory); + try + { + using var registry = new RemoteRunRegistry(directory); + using var firstObserver = CreateObserver(); + using var secondObserver = CreateObserver(); + Assert.IsTrue(RunContext.CreateRunContext(null!, "worker-run", [], Path.GetTempPath(), + "Start", RunMode.Normal, out var workerContext)); + Assert.IsTrue(RunContext.CreateRunContext(null!, "first-ordinary-run", [], Path.GetTempPath(), + "Start", RunMode.Normal, out var firstRun)); + Assert.IsTrue(RunContext.CreateRunContext(null!, "second-ordinary-run", [], Path.GetTempPath(), + "Start", RunMode.Normal, out var secondRun)); + + using var workerLease = registry.ReserveWorkerSlot(workerContext, 1); + await workerLease.Started.WaitAsync(TimeSpan.FromSeconds(3)); + Assert.IsTrue(registry.Submit(firstObserver, firstRun, "first", RunMode.Normal, + firstRun.WorkingDirectory, firstRun.StartToExecute, [])); + Assert.IsTrue(registry.Submit(secondObserver, secondRun, "second", RunMode.Normal, + secondRun.WorkingDirectory, secondRun.StartToExecute, [])); + + var activities = registry.GetActiveActivities(); + Assert.HasCount(3, activities); + Assert.AreEqual("worker-run", activities[0].RunId); + Assert.AreEqual(RunServerActivityState.Running, activities[0].State); + CollectionAssert.AreEqual(new[] { "first-ordinary-run", "second-ordinary-run" }, + activities.Skip(1).Select(activity => activity.RunId).ToArray()); + Assert.IsTrue(activities.Skip(1).All(activity => activity.State == RunServerActivityState.Queued)); + + registry.Dispose(); + workerLease.Dispose(); + } + finally + { + Directory.Delete(directory, recursive: true); + } + } + + [TestMethod] + public async Task SeparateRegistries_HaveIndependentExecutionSlots() + { + var firstDirectory = Path.Combine(Path.GetTempPath(), "XTMF2-RemoteRunRegistry-" + Guid.NewGuid().ToString("N")); + var secondDirectory = Path.Combine(Path.GetTempPath(), "XTMF2-RemoteRunRegistry-" + Guid.NewGuid().ToString("N")); + Directory.CreateDirectory(firstDirectory); + Directory.CreateDirectory(secondDirectory); + try + { + using var firstRegistry = new RemoteRunRegistry(firstDirectory); + using var secondRegistry = new RemoteRunRegistry(secondDirectory); + Assert.IsTrue(RunContext.CreateRunContext(null!, "first-process-run", [], Path.GetTempPath(), + "Start", RunMode.Normal, out var firstContext)); + Assert.IsTrue(RunContext.CreateRunContext(null!, "second-process-run", [], Path.GetTempPath(), + "Start", RunMode.Normal, out var secondContext)); + + using var firstLease = firstRegistry.ReserveWorkerSlot(firstContext, 1); + using var secondLease = secondRegistry.ReserveWorkerSlot(secondContext, 1); + await Task.WhenAll(firstLease.Started, secondLease.Started).WaitAsync(TimeSpan.FromSeconds(3)); + + Assert.AreEqual(RunServerActivityState.Running, firstRegistry.GetActiveActivities().Single().State); + Assert.AreEqual(RunServerActivityState.Running, secondRegistry.GetActiveActivities().Single().State); + } + finally + { + Directory.Delete(firstDirectory, recursive: true); + Directory.Delete(secondDirectory, recursive: true); + } + } + [TestMethod] public void RegistryRestart_MarksPersistedActiveRunInterrupted() { @@ -47,6 +121,10 @@ public void RegistryRestart_MarksPersistedActiveRunInterrupted() } } + private static RunServerBus CreateObserver() + => new(new MemoryStream(), true, runtime: null!, remoteRunRegistry: null, + allowDeployment: false); + [TestMethod] public void DeleteRun_RemovesWorkspaceArchiveAndManifest() {