diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/BlockingPipeStreamTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/BlockingPipeStreamTests.cs new file mode 100644 index 000000000..0f2561c2c --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/BlockingPipeStreamTests.cs @@ -0,0 +1,147 @@ +using System.IO.Pipes; +using System.Security.Cryptography; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol; +using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1; + +namespace ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests; + +/// +/// lmxopcua#789: — on Windows the +/// that keeps worker pipe I/O off the completion port. Exercised on a real +/// pipe pair; on Unix the same tests cover the plain pipe it returns there. +/// +public sealed class BlockingPipeStreamTests +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static async Task<(Stream Server, Stream Client)> PairAsync() + { + var name = "otopcua-sw-test-" + Convert.ToHexString(RandomNumberGenerator.GetBytes(6)); + var server = new NamedPipeServerStream(name, PipeDirection.InOut, 1, PipeTransmissionMode.Byte, + PipeOptions.Asynchronous | PipeOptions.CurrentUserOnly); + var client = new NamedPipeClientStream(".", name, PipeDirection.InOut, PipeOptions.Asynchronous | PipeOptions.CurrentUserOnly); + var accept = server.WaitForConnectionAsync(Ct); + client.Connect(TimeSpan.FromSeconds(5)); + await accept; + return (ScriptWorkerFrameCodec.BlockingStream(server), ScriptWorkerFrameCodec.BlockingStream(client)); + } + + private static Envelope Frame(ulong id) => new() + { + ProtocolVersion = ScriptWorkerFrameCodec.ProtocolVersion, + Seq = id, + Heartbeat = new Heartbeat { Inflight = (uint)id }, + }; + + [Fact] + public async Task On_windows_the_pipe_is_wrapped_and_elsewhere_it_is_the_pipe_itself() + { + var (server, client) = await PairAsync(); + using (server) + using (client) + { + if (OperatingSystem.IsWindows()) server.ShouldBeOfType(); + else server.ShouldBeAssignableTo(); + } + } + + /// + /// Full duplex: each end has a read blocked on a dedicated thread while the other direction keeps writing — a + /// handle opened for synchronous I/O would serialise the write behind the blocked read and deadlock here. + /// + [Fact] + public async Task Reads_blocked_on_both_ends_do_not_hold_up_writes_in_either_direction() + { + var (server, client) = await PairAsync(); + using (server) + using (client) + { + const int frames = 2_000; + Task Pump(Stream from, Stream to) + { + var reader = Task.Factory.StartNew(() => + { + var r = new ScriptWorkerFrameReader(from); + var seen = 0; + while (seen < frames && r.Read() is { } f) + { + f.Heartbeat.Inflight.ShouldBe((uint)seen); + seen++; + } + return seen; + }, Ct, TaskCreationOptions.LongRunning, TaskScheduler.Default); + _ = Task.Factory.StartNew(() => + { + for (var i = 0; i < frames; i++) ScriptWorkerFrameCodec.Write(to, Frame((ulong)i), ScriptWorkerFrameCodec.DefaultMaxFrameBytes); + }, Ct, TaskCreationOptions.LongRunning, TaskScheduler.Default); + return reader; + } + + var toClient = Pump(client, server); + var toServer = Pump(server, client); + (await toClient.WaitAsync(TimeSpan.FromSeconds(30), Ct)).ShouldBe(frames); + (await toServer.WaitAsync(TimeSpan.FromSeconds(30), Ct)).ShouldBe(frames); + } + } + + [Fact] + public async Task A_large_frame_crosses_whole() + { + var (server, client) = await PairAsync(); + using (server) + using (client) + { + var big = new Envelope + { + ProtocolVersion = ScriptWorkerFrameCodec.ProtocolVersion, + Load = new LoadScripts { Scripts = { new ScriptSource { Key = "k", Source = new string('x', 1_000_000) } } }, + }; + var read = Task.Factory.StartNew(() => new ScriptWorkerFrameReader(server).Read(), Ct, TaskCreationOptions.LongRunning, + TaskScheduler.Default); + ScriptWorkerFrameCodec.Write(client, big, ScriptWorkerFrameCodec.DefaultMaxFrameBytes); + var got = await read.WaitAsync(TimeSpan.FromSeconds(30), Ct); + got!.Load.Scripts[0].Source.Length.ShouldBe(1_000_000); + } + } + + [Fact] + public async Task A_peer_that_goes_away_ends_a_blocked_read_and_fails_a_write() + { + var (server, client) = await PairAsync(); + using (server) + { + var read = Task.Factory.StartNew(() => server.Read(new byte[16], 0, 16), Ct, TaskCreationOptions.LongRunning, + TaskScheduler.Default); + await Task.Delay(100, Ct); + read.IsCompleted.ShouldBeFalse("positive control: the read is blocked on the kernel"); + client.Dispose(); + (await read.WaitAsync(TimeSpan.FromSeconds(10), Ct)).ShouldBe(0, "end of stream"); + Should.Throw(() => + { + // A pipe can accept a write into its buffer before it notices; the second one must fail. + for (var i = 0; i < 4; i++) server.Write(new byte[64], 0, 64); + }); + } + } + + [Fact] + public async Task Disposing_an_end_wakes_its_own_blocked_reader() + { + var (server, client) = await PairAsync(); + using (client) + { + var read = Task.Factory.StartNew(() => server.Read(new byte[16], 0, 16), Ct, TaskCreationOptions.LongRunning, + TaskScheduler.Default); + await Task.Delay(100, Ct); + read.IsCompleted.ShouldBeFalse("positive control: the read is blocked on the kernel"); + server.Dispose(); + // End of stream, or the disposed / I/O error the readers treat as the pipe ending — never a hang. + int ended; + try { ended = await read.WaitAsync(TimeSpan.FromSeconds(10), Ct); } + catch (Exception ex) when (ex is ObjectDisposedException or IOException) { ended = 0; } + ended.ShouldBe(0); + } + } +}