diff --git a/src/XTMF2.GUI/RunController.cs b/src/XTMF2.GUI/RunController.cs index 7ffedc28..58b0bcd4 100644 --- a/src/XTMF2.GUI/RunController.cs +++ b/src/XTMF2.GUI/RunController.cs @@ -1524,7 +1524,7 @@ private void OnRemoteRunArtifactsAvailable(object? sender, RemoteRunArtifactsRes return; _recoveredRemoteRuns.TryGetValue(response.RunId, out snapshot); } - if (response.Archive is null) + if (response.ArchivePath is null) { RunsViewModel.NotifyArtifactTransferFailed(response.RunId, response.Error ?? "The RunServer returned no artifact archive."); @@ -1533,7 +1533,7 @@ private void OnRemoteRunArtifactsAvailable(object? sender, RemoteRunArtifactsRes try { - using var archiveStream = new MemoryStream(response.Archive, writable: false); + using var archiveStream = File.OpenRead(response.ArchivePath); ExtractRunArtifacts(archiveStream, receipt.TargetDirectory); RunsViewModel.NotifyArtifactsTransferred(response.RunId); if (snapshot is not null) diff --git a/src/XTMF2/Bus/HostBus.cs b/src/XTMF2/Bus/HostBus.cs index dc2195d2..b51add8a 100644 --- a/src/XTMF2/Bus/HostBus.cs +++ b/src/XTMF2/Bus/HostBus.cs @@ -384,8 +384,25 @@ private Thread StartListenner() this, SharedEstimationProtocol.ReadRemoteRunSnapshotsPayload(reader))); break; case SharedEstimationMessageType.RemoteRunArtifacts: - IgnoreWarnings(() => RemoteRunArtifactsAvailable?.Invoke( - this, SharedEstimationProtocol.ReadRemoteRunArtifactsPayload(reader))); + { + var archivePath = Path.Combine(Path.GetTempPath(), + $"XTMF2-{Guid.NewGuid():N}.zip"); + try + { + RemoteRunArtifactsResponse response; + using (var archive = new FileStream(archivePath, FileMode.CreateNew, + FileAccess.Write, FileShare.Read)) + { + response = SharedEstimationProtocol.ReadRemoteRunArtifactsPayload( + reader, archive, archivePath); + } + IgnoreWarnings(() => RemoteRunArtifactsAvailable?.Invoke(this, response)); + } + finally + { + try { File.Delete(archivePath); } catch { } + } + } break; case SharedEstimationMessageType.ServerActivitySnapshots: var activityResponse = SharedEstimationProtocol.ReadServerActivitySnapshotsPayload(reader); diff --git a/src/XTMF2/Bus/Optimization/SharedEstimationProtocol.cs b/src/XTMF2/Bus/Optimization/SharedEstimationProtocol.cs index 2b7f5702..b436dece 100644 --- a/src/XTMF2/Bus/Optimization/SharedEstimationProtocol.cs +++ b/src/XTMF2/Bus/Optimization/SharedEstimationProtocol.cs @@ -894,27 +894,43 @@ public static string ReadGetRemoteRunArtifacts(BinaryReader reader) internal static string ReadGetRemoteRunArtifactsPayload(BinaryReader reader) => ReadRequiredString(reader); - public static void WriteRemoteRunArtifacts(BinaryWriter writer, string runId, byte[]? archive, string? error = null) + public static void WriteRemoteRunArtifacts(BinaryWriter writer, string runId, Stream? archive, + long archiveLength, string? error = null) { WriteHeader(writer, SharedEstimationMessageType.RemoteRunArtifacts); WriteRequiredString(writer, runId); writer.Write(archive is not null); if (archive is not null) - WriteBytes(writer, archive); + { + if (archiveLength < 0) + throw new ArgumentOutOfRangeException(nameof(archiveLength)); + writer.Write(archiveLength); + CopyExactly(archive, writer.BaseStream, archiveLength); + } WriteOptionalString(writer, error); } - public static RemoteRunArtifactsResponse ReadRemoteRunArtifacts(BinaryReader reader) + public static RemoteRunArtifactsResponse ReadRemoteRunArtifacts(BinaryReader reader, + Stream archiveDestination, string archivePath) { ReadExpectedHeader(reader, SharedEstimationMessageType.RemoteRunArtifacts); - return ReadRemoteRunArtifactsPayload(reader); + return ReadRemoteRunArtifactsPayload(reader, archiveDestination, archivePath); } - internal static RemoteRunArtifactsResponse ReadRemoteRunArtifactsPayload(BinaryReader reader) + internal static RemoteRunArtifactsResponse ReadRemoteRunArtifactsPayload(BinaryReader reader, + Stream archiveDestination, string archivePath) { var runId = ReadRequiredString(reader); - var archive = reader.ReadBoolean() ? ReadBytes(reader) : null; - return new RemoteRunArtifactsResponse(runId, archive, ReadOptionalString(reader)); + var hasArchive = reader.ReadBoolean(); + if (hasArchive) + { + var archiveLength = reader.ReadInt64(); + if (archiveLength < 0) + throw new InvalidDataException($"Invalid remote run artifact length: {archiveLength}."); + CopyExactly(reader.BaseStream, archiveDestination, archiveLength); + } + return new RemoteRunArtifactsResponse(runId, hasArchive ? archivePath : null, + ReadOptionalString(reader)); } public static void WriteAcknowledgeRemoteRun(BinaryWriter writer, string runId) @@ -1133,6 +1149,19 @@ private static byte[] ReadBytes(BinaryReader reader) : throw new EndOfStreamException("The shared estimation payload was truncated."); } + private static void CopyExactly(Stream source, Stream destination, long length) + { + var buffer = new byte[81920]; + while (length > 0) + { + int read = source.Read(buffer, 0, (int)Math.Min(buffer.Length, length)); + if (read == 0) + throw new EndOfStreamException("The remote run artifact archive was truncated."); + destination.Write(buffer, 0, read); + length -= read; + } + } + private static void WriteDoubles(BinaryWriter writer, IReadOnlyList values) { WriteCount(writer, values.Count); diff --git a/src/XTMF2/Bus/RemoteRunRegistry.cs b/src/XTMF2/Bus/RemoteRunRegistry.cs index 51b3638d..d2563c4c 100644 --- a/src/XTMF2/Bus/RemoteRunRegistry.cs +++ b/src/XTMF2/Bus/RemoteRunRegistry.cs @@ -20,7 +20,7 @@ public enum RemoteRunState public sealed record RemoteRunParameterValue(int NodeIndex, double Value); -public sealed record RemoteRunArtifactsResponse(string RunId, byte[]? Archive, string? Error); +public sealed record RemoteRunArtifactsResponse(string RunId, string? ArchivePath, string? Error); public sealed record RemoteRunSnapshot( string RunId, @@ -246,6 +246,46 @@ public bool TryReadArtifacts(string runId, out byte[]? archive) } } + internal bool TryOpenArtifacts(string runId, out FileStream? archive) + { + archive = null; + RemoteRunJob? job; + string path; + lock (_sync) + { + if (!_jobs.TryGetValue(runId, out job)) + return false; + path = GetArtifactPath(runId); + } + + try + { + if (!File.Exists(path)) + { + var snapshot = job.GetSnapshot(); + if (snapshot.State == RemoteRunState.Running || !IsPrivateRunDirectory(runId, snapshot.WorkingDirectory) || + !Directory.Exists(snapshot.WorkingDirectory)) + return false; + + var temporary = path + ".tmp"; + if (File.Exists(temporary)) + File.Delete(temporary); + ZipFile.CreateFromDirectory(snapshot.WorkingDirectory, temporary, CompressionLevel.Fastest, false); + File.Move(temporary, path, true); + job.SetArtifactsAvailable(); + } + + archive = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.Read); + return true; + } + catch (Exception exception) when (exception is IOException or UnauthorizedAccessException) + { + archive?.Dispose(); + archive = null; + return false; + } + } + /// /// Acknowledges receipt of a completed run's artifacts and removes the retained archive. /// diff --git a/src/XTMF2/Bus/RunServerBus.cs b/src/XTMF2/Bus/RunServerBus.cs index d8ba502c..a62a15c2 100644 --- a/src/XTMF2/Bus/RunServerBus.cs +++ b/src/XTMF2/Bus/RunServerBus.cs @@ -418,8 +418,9 @@ public void SendServerActivitySnapshots(string requestId, IReadOnlyList snapshots) => WriteSharedEstimation(writer => SharedEstimationProtocol.WriteRemoteRunSnapshots(writer, snapshots)); - public void SendRemoteRunArtifacts(string runId, byte[]? archive, string? error) - => WriteSharedEstimation(writer => SharedEstimationProtocol.WriteRemoteRunArtifacts(writer, runId, archive, error)); + public void SendRemoteRunArtifacts(string runId, Stream? archive, long archiveLength, string? error) + => WriteSharedEstimation(writer => + SharedEstimationProtocol.WriteRemoteRunArtifacts(writer, runId, archive, archiveLength, error)); public void SendRemoteRunDeletionResponse(RemoteRunDeletionResponse response) => WriteSharedEstimation(writer => SharedEstimationProtocol.WriteRemoteRunDeleted(writer, response)); @@ -662,11 +663,14 @@ public void ProcessRequests() case SharedEstimationMessageType.GetRemoteRunArtifacts: { var runId = SharedEstimationProtocol.ReadGetRemoteRunArtifactsPayload(reader); - byte[]? archive = null; + FileStream? archive = null; var found = _remoteRunRegistry is not null && - _remoteRunRegistry.TryReadArtifacts(runId, out archive); - SendRemoteRunArtifacts(runId, found ? archive : null, - found ? null : "No unacknowledged artifact archive is available for this run."); + _remoteRunRegistry.TryOpenArtifacts(runId, out archive); + using (archive) + { + SendRemoteRunArtifacts(runId, archive, archive?.Length ?? 0, + found ? null : "No unacknowledged artifact archive is available for this run."); + } } break; case SharedEstimationMessageType.AcknowledgeRemoteRun: diff --git a/tests/XTMF2.UnitTests/Bus/Optimization/TestSharedEstimationProtocol.cs b/tests/XTMF2.UnitTests/Bus/Optimization/TestSharedEstimationProtocol.cs index 74f9d6a2..a0aea668 100644 --- a/tests/XTMF2.UnitTests/Bus/Optimization/TestSharedEstimationProtocol.cs +++ b/tests/XTMF2.UnitTests/Bus/Optimization/TestSharedEstimationProtocol.cs @@ -285,7 +285,8 @@ [new RemoteRunParameterValue(3, 0.75)], null, null, null, false, timestamp, "model-hash-2", RemoteRunState.Completed, "Complete", 8, 0.01, [], [new RemoteRunParameterValue(5, 1.1)], null, null, true, timestamp) }; - var archive = new byte[] { 4, 5, 6, 7 }; + var archive = Enumerable.Range(0, 1_000_001).Select(value => (byte)value).ToArray(); + using var archiveSource = new MemoryStream(archive, writable: false); using var stream = new MemoryStream(); using (var writer = new BinaryWriter(stream, System.Text.Encoding.UTF8, true)) @@ -293,12 +294,14 @@ [new RemoteRunParameterValue(3, 0.75)], null, null, null, false, timestamp, SharedEstimationProtocol.WriteQueryRemoteRuns(writer); SharedEstimationProtocol.WriteRemoteRunSnapshots(writer, snapshots); SharedEstimationProtocol.WriteGetRemoteRunArtifacts(writer, "run-2"); - SharedEstimationProtocol.WriteRemoteRunArtifacts(writer, "run-2", archive); + SharedEstimationProtocol.WriteRemoteRunArtifacts(writer, "run-2", archiveSource, + archiveSource.Length); SharedEstimationProtocol.WriteAcknowledgeRemoteRun(writer, "run-2"); } stream.Position = 0; using var reader = new BinaryReader(stream, System.Text.Encoding.UTF8, true); + using var receivedArchive = new MemoryStream(); SharedEstimationProtocol.ReadQueryRemoteRuns(reader); var readSnapshots = SharedEstimationProtocol.ReadRemoteRunSnapshots(reader); Assert.HasCount(2, readSnapshots); @@ -313,9 +316,10 @@ [new RemoteRunParameterValue(3, 0.75)], null, null, null, false, timestamp, Assert.AreEqual(modelSystemId, readSnapshots[0].ModelSystemId); Assert.AreEqual(ownerUserId, readSnapshots[0].OwnerUserId); Assert.AreEqual("run-2", SharedEstimationProtocol.ReadGetRemoteRunArtifacts(reader)); - var response = SharedEstimationProtocol.ReadRemoteRunArtifacts(reader); + var response = SharedEstimationProtocol.ReadRemoteRunArtifacts(reader, receivedArchive, "archive.zip"); Assert.AreEqual("run-2", response.RunId); - CollectionAssert.AreEqual(archive, response.Archive); + Assert.AreEqual("archive.zip", response.ArchivePath); + CollectionAssert.AreEqual(archive, receivedArchive.ToArray()); Assert.IsNull(response.Error); Assert.AreEqual("run-2", SharedEstimationProtocol.ReadAcknowledgeRemoteRun(reader)); } diff --git a/tests/XTMF2.UnitTests/Bus/TestSharedEstimationWorkerParticipant.cs b/tests/XTMF2.UnitTests/Bus/TestSharedEstimationWorkerParticipant.cs index 377d9fef..fdd3f054 100644 --- a/tests/XTMF2.UnitTests/Bus/TestSharedEstimationWorkerParticipant.cs +++ b/tests/XTMF2.UnitTests/Bus/TestSharedEstimationWorkerParticipant.cs @@ -17,7 +17,7 @@ public class TestSharedEstimationWorkerParticipant [TestMethod] public void Evaluate_UsesPreparedModelSystemAndReturnsFitness() { - RunInModelSystemContext("SharedEstimationWorkerParticipant", (user, projectSession, session) => + RunInModelSystemContext("Evaluate_UsesPreparedModelSystemAndReturnsFitness", (user, projectSession, session) => { XTMF2.Editing.CommandError error = null; var modelSystem = session.ModelSystem; diff --git a/tests/XTMF2.UnitTests/ModelSystemConstruct/Parameters/Compiler/TestNotOperator.cs b/tests/XTMF2.UnitTests/ModelSystemConstruct/Parameters/Compiler/TestNotOperator.cs index ca54f6fe..d7768ac3 100644 --- a/tests/XTMF2.UnitTests/ModelSystemConstruct/Parameters/Compiler/TestNotOperator.cs +++ b/tests/XTMF2.UnitTests/ModelSystemConstruct/Parameters/Compiler/TestNotOperator.cs @@ -100,7 +100,7 @@ public void TestNonBoolTextFails() [TestMethod] public void TestNotVariable() { - TestHelper.RunInModelSystemContext("TestBadVariableNames", (User user, ProjectSession project, ModelSystemSession session) => + TestHelper.RunInModelSystemContext("TestNotVariable", (User user, ProjectSession project, ModelSystemSession session) => { var nodes = new List() { diff --git a/tests/XTMF2.UnitTests/TestRun.cs b/tests/XTMF2.UnitTests/TestRun.cs index dc3db7cf..52b420b3 100644 --- a/tests/XTMF2.UnitTests/TestRun.cs +++ b/tests/XTMF2.UnitTests/TestRun.cs @@ -232,7 +232,7 @@ public void ReportRunProgress() }; Assert.IsTrue(clientBus.RunModelSystem(msSession, Path.Combine(pSession.RunsDirectory, "ReportRunProgress"), "Start", out var id, out error), error?.Message); // give the models system some time to complete - if (!sim.Wait(2000)) + if (!sim.Wait(5000)) { Assert.Fail("The model system failed to execute in time!"); }