diff --git a/source/Halibut.TestUtils.CompatBinary.Base/BackwardsCompatProgramBase.cs b/source/Halibut.TestUtils.CompatBinary.Base/BackwardsCompatProgramBase.cs index 80b624fee..f5f4a027c 100644 --- a/source/Halibut.TestUtils.CompatBinary.Base/BackwardsCompatProgramBase.cs +++ b/source/Halibut.TestUtils.CompatBinary.Base/BackwardsCompatProgramBase.cs @@ -17,7 +17,7 @@ public static async Task Main() }); var mode = SettingsHelper.GetSetting("mode") ?? string.Empty; - Console.WriteLine($"Mode is: {mode}"); + NonBlockingConsole.WriteLine($"Mode is: {mode}"); if (mode.Equals("serviceonly")) { @@ -29,7 +29,8 @@ public static async Task Main() } else { - Console.WriteLine($"Unknown mode: {mode}"); + NonBlockingConsole.WriteLine($"Unknown mode: {mode}"); + NonBlockingConsole.FlushBeforeExit(); throw new Exception($"Unknown mode: {mode}"); } @@ -47,7 +48,7 @@ static async Task RunExternalService(CancellationToken cancellationToken) if (serviceConnectionType is ServiceConnectionType.Polling or ServiceConnectionType.PollingOverWebSocket) { addressToPoll = SettingsHelper.GetSetting("octopusservercommsport"); - Console.WriteLine($"Will poll: {addressToPoll}"); + NonBlockingConsole.WriteLine($"Will poll: {addressToPoll}"); } var proxyDetails = SettingsHelper.GetProxyDetails(); @@ -67,22 +68,22 @@ static async Task RunExternalService(CancellationToken cancellationToken) case ServiceConnectionType.PollingOverWebSocket: FixHungWebSocketsHack.EnableHack(); var sslThubprint = SettingsHelper.GetSetting("sslthubprint"); - Console.WriteLine($"Using SSL thumbprint: {sslThubprint}"); + NonBlockingConsole.WriteLine($"Using SSL thumbprint: {sslThubprint}"); tentaclePolling.Poll(new Uri("poll://SQ-TENTAPOLL"), new ServiceEndPoint(new Uri(addressToPoll), sslThubprint, proxyDetails)); break; case ServiceConnectionType.Listening: var port = tentaclePolling.Listen(); - Console.WriteLine($"Listening on port: {port}"); + NonBlockingConsole.WriteLine($"Listening on port: {port}"); tentaclePolling.Trust(octopusThumbprint); break; default: throw new ArgumentOutOfRangeException(); } - Console.WriteLine("RunningAndReady"); - Console.WriteLine("Will Now sleep"); - await Console.Out.FlushAsync(); + NonBlockingConsole.WriteLine("RunningAndReady"); + NonBlockingConsole.WriteLine("Will Now sleep"); + await NonBlockingConsole.FlushAsync(); // Run until the Program is terminated await StayAliveUntilHelper.WaitUntilSignaledToDie(); diff --git a/source/Halibut.TestUtils.CompatBinary.Base/DelegateCachingService.cs b/source/Halibut.TestUtils.CompatBinary.Base/DelegateCachingService.cs index ba23544e7..d2e3151a4 100644 --- a/source/Halibut.TestUtils.CompatBinary.Base/DelegateCachingService.cs +++ b/source/Halibut.TestUtils.CompatBinary.Base/DelegateCachingService.cs @@ -1,5 +1,6 @@ using System; using Halibut.TestUtils.Contracts; +using Halibut.TestUtils.SampleProgram.Base.LogUtils; namespace Halibut.TestUtils.SampleProgram.Base { @@ -14,37 +15,37 @@ public DelegateCachingService(ICachingService cachingService) public Guid NonCachableCall() { - Console.WriteLine("Forwarding NonCachableCall() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding NonCachableCall() call to delegate"); return cachingService.NonCachableCall(); } public Guid CachableCall() { - Console.WriteLine("Forwarding CachableCall() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding CachableCall() call to delegate"); return cachingService.CachableCall(); } public Guid CachableCall(Guid input) { - Console.WriteLine("Forwarding CachableCall(Guid) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding CachableCall(Guid) call to delegate"); return cachingService.CachableCall(input); } public Guid AnotherCachableCall() { - Console.WriteLine("Forwarding AnotherCachableCall() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding AnotherCachableCall() call to delegate"); return cachingService.AnotherCachableCall(); } public Guid CachableCallThatThrowsAnExceptionWithARandomExceptionMessage(string exceptionMessagePrefix) { - Console.WriteLine("Forwarding CachableCallThatThrowsAnExceptionWithARandomExceptionMessage() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding CachableCallThatThrowsAnExceptionWithARandomExceptionMessage() call to delegate"); return cachingService.CachableCallThatThrowsAnExceptionWithARandomExceptionMessage(exceptionMessagePrefix); } public Guid TwoSecondCachableCall() { - Console.WriteLine("Forwarding TwoSecondCachableCall() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding TwoSecondCachableCall() call to delegate"); return cachingService.TwoSecondCachableCall(); } } diff --git a/source/Halibut.TestUtils.CompatBinary.Base/DelegateEchoService.cs b/source/Halibut.TestUtils.CompatBinary.Base/DelegateEchoService.cs index 0856ea080..aff787986 100644 --- a/source/Halibut.TestUtils.CompatBinary.Base/DelegateEchoService.cs +++ b/source/Halibut.TestUtils.CompatBinary.Base/DelegateEchoService.cs @@ -1,6 +1,7 @@ using System; using System.IO; using Halibut.TestUtils.Contracts; +using Halibut.TestUtils.SampleProgram.Base.LogUtils; using Halibut.TestUtils.SampleProgram.Base.Utils; namespace Halibut.TestUtils.SampleProgram.Base @@ -16,19 +17,19 @@ public DelegateEchoService(IEchoService echoService) public int LongRunningOperation() { - Console.WriteLine("Forwarding LongRunningOperation() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding LongRunningOperation() call to delegate"); return echoService.LongRunningOperation(); } public string SayHello(string name) { - Console.WriteLine("Forwarding SayHello() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding SayHello() call to delegate"); return echoService.SayHello(name); } public bool Crash() { - Console.WriteLine("Forwarding Crash() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Crash() call to delegate"); try { return echoService.Crash(); @@ -44,7 +45,7 @@ public bool Crash() public int CountBytes(DataStream dataStream) { dataStream = dataStream.ConfigureWriterOnReceivedDataStream(); - Console.WriteLine("Forwarding CountBytes() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding CountBytes() call to delegate"); return echoService.CountBytes(dataStream); } } diff --git a/source/Halibut.TestUtils.CompatBinary.Base/DelegateMultipleParametersTestService.cs b/source/Halibut.TestUtils.CompatBinary.Base/DelegateMultipleParametersTestService.cs index 70bec138c..f775a2814 100644 --- a/source/Halibut.TestUtils.CompatBinary.Base/DelegateMultipleParametersTestService.cs +++ b/source/Halibut.TestUtils.CompatBinary.Base/DelegateMultipleParametersTestService.cs @@ -1,5 +1,6 @@ using System; using Halibut.TestUtils.Contracts; +using Halibut.TestUtils.SampleProgram.Base.LogUtils; namespace Halibut.TestUtils.SampleProgram.Base { @@ -14,115 +15,115 @@ public DelegateMultipleParametersTestService(IMultipleParametersTestService mult public void MethodReturningVoid(long a, long b) { - Console.WriteLine("Forwarding MethodReturningVoid() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding MethodReturningVoid() call to delegate"); multipleParametersTestService.MethodReturningVoid(a, b); } public long Add(long a, long b) { - Console.WriteLine("Forwarding Add(long, long) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Add(long, long) call to delegate"); return multipleParametersTestService.Add(a, b); } public double Add(double a, double b) { - Console.WriteLine("Forwarding Add(double, double) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Add(double, double) call to delegate"); return multipleParametersTestService.Add(a, b); } public decimal Add(decimal a, decimal b) { - Console.WriteLine("Forwarding Add(decimal, decimal) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Add(decimal, decimal) call to delegate"); return multipleParametersTestService.Add(a, b); } public string Hello() { - Console.WriteLine("Forwarding Hello() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello() call to delegate"); return multipleParametersTestService.Hello(); } public string Hello(string a) { - Console.WriteLine("Forwarding Hello(1 param) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(1 param) call to delegate"); return multipleParametersTestService.Hello(a); } public string Hello(string a, string b) { - Console.WriteLine("Forwarding Hello(2 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(2 params) call to delegate"); return multipleParametersTestService.Hello(a, b); } public string Hello(string a, string b, string c) { - Console.WriteLine("Forwarding Hello(3 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(3 params) call to delegate"); return multipleParametersTestService.Hello(a, b, c); } public string Hello(string a, string b, string c, string d) { - Console.WriteLine("Forwarding Hello(4 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(4 params) call to delegate"); return multipleParametersTestService.Hello(a, b, c, d); } public string Hello(string a, string b, string c, string d, string e) { - Console.WriteLine("Forwarding Hello(5 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(5 params) call to delegate"); return multipleParametersTestService.Hello(a, b, c, d, e); } public string Hello(string a, string b, string c, string d, string e, string f) { - Console.WriteLine("Forwarding Hello(6 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(6 params) call to delegate"); return multipleParametersTestService.Hello(a, b, c, d, e, f); } public string Hello(string a, string b, string c, string d, string e, string f, string g) { - Console.WriteLine("Forwarding Hello(7 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(7 params) call to delegate"); return multipleParametersTestService.Hello(a, b, c, d, e, f, g); } public string Hello(string a, string b, string c, string d, string e, string f, string g, string h) { - Console.WriteLine("Forwarding Hello(8 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(8 params) call to delegate"); return multipleParametersTestService.Hello(a, b, c, d, e, f, g, h); } public string Hello(string a, string b, string c, string d, string e, string f, string g, string h, string i) { - Console.WriteLine("Forwarding Hello(9 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(9 params) call to delegate"); return multipleParametersTestService.Hello(a, b, c, d, e, f, g, h, i); } public string Hello(string a, string b, string c, string d, string e, string f, string g, string h, string i, string j) { - Console.WriteLine("Forwarding Hello(10 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(10 params) call to delegate"); return multipleParametersTestService.Hello(a, b, c, d, e, f, g, h, i, j); } public string Hello(string a, string b, string c, string d, string e, string f, string g, string h, string i, string j, string k) { - Console.WriteLine("Forwarding Hello(11 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(11 params) call to delegate"); return multipleParametersTestService.Hello(a, b, c, d, e, f, g, h, i, j, k); } public string Ambiguous(string a, string b) { - Console.WriteLine("Forwarding Ambiguous(string, string) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Ambiguous(string, string) call to delegate"); return multipleParametersTestService.Ambiguous(a, b); } public string Ambiguous(string a, Tuple b) { - Console.WriteLine("Forwarding Ambiguous(string, Tuple) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Ambiguous(string, Tuple) call to delegate"); return multipleParametersTestService.Ambiguous(a, b); } public MapLocation GetLocation(MapLocation loc) { - Console.WriteLine("Forwarding GetLocation() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding GetLocation() call to delegate"); return multipleParametersTestService.GetLocation(loc); } } diff --git a/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/NonBlockingConsole.cs b/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/NonBlockingConsole.cs new file mode 100644 index 000000000..5eab161e2 --- /dev/null +++ b/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/NonBlockingConsole.cs @@ -0,0 +1,28 @@ +using System; +using System.Threading.Tasks; + +namespace Halibut.TestUtils.SampleProgram.Base.LogUtils +{ + /// + /// Use this instead of everywhere in the compat binaries. + /// See for why. + /// + public static class NonBlockingConsole + { + static readonly TimeSpan DefaultFlushTimeout = TimeSpan.FromSeconds(10); + + static readonly NonBlockingLineWriter Writer = new(Console.Out, threadName: "CompatBinaryConsoleWriter"); + + public static void WriteLine(string line) => Writer.WriteLine(line); + + /// + /// Waits for queued lines to reach stdout, e.g. so the test process sees "RunningAndReady". + /// + public static Task FlushAsync() => Writer.FlushAsync(DefaultFlushTimeout); + + /// + /// Waits (briefly) for queued lines to reach stdout before the process exits. + /// + public static bool FlushBeforeExit() => Writer.Flush(TimeSpan.FromSeconds(2)); + } +} diff --git a/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/NonBlockingLineWriter.cs b/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/NonBlockingLineWriter.cs new file mode 100644 index 000000000..2041b2524 --- /dev/null +++ b/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/NonBlockingLineWriter.cs @@ -0,0 +1,139 @@ +using System; +using System.Collections.Concurrent; +using System.IO; +using System.Threading; +using System.Threading.Tasks; + +namespace Halibut.TestUtils.SampleProgram.Base.LogUtils +{ + /// + /// Writes lines to a from a dedicated background thread, so callers never block on the write. + /// + /// The compat binaries write their logs to stdout, which the test process reads through a pipe. If the test process + /// stops reading, the pipe buffer (about 4KB on Windows) fills up and a plain Console.WriteLine blocks. Because Halibut + /// logs on the request path, that blocked write freezes the whole compat binary: in-flight requests stop, and new + /// connections are never accepted (see LEV-2253). + /// + /// With this writer, a stalled reader only stalls the background thread. Callers queue lines up to a fixed capacity, + /// after which lines are dropped (and counted) rather than blocking the Halibut protocol. + /// + public class NonBlockingLineWriter : IDisposable + { + public const int DefaultCapacity = 10_000; + + readonly TextWriter target; + readonly BlockingCollection lines; + readonly Thread writerThread; + long droppedLines; + long pendingLines; + + public NonBlockingLineWriter(TextWriter target, int capacity = DefaultCapacity, string threadName = nameof(NonBlockingLineWriter)) + { + if (capacity <= 0) throw new ArgumentOutOfRangeException(nameof(capacity), capacity, "Capacity must be greater than zero."); + + this.target = target ?? throw new ArgumentNullException(nameof(target)); + lines = new BlockingCollection(new ConcurrentQueue(), capacity); + writerThread = new Thread(WriteLoop) + { + IsBackground = true, + Name = threadName + }; + writerThread.Start(); + } + + /// + /// The number of lines dropped because the queue was full, and not yet reported to the target. + /// + public long DroppedLines => Interlocked.Read(ref droppedLines); + + /// + /// Queues a line to be written. Never blocks. If the queue is full the line is dropped, and a count of dropped + /// lines is written to the target once it starts accepting writes again. + /// + public void WriteLine(string line) + { + Interlocked.Increment(ref pendingLines); + + bool added; + try + { + added = lines.TryAdd(line); + } + catch (InvalidOperationException) + { + // Disposed (CompleteAdding was called). + added = false; + } + + if (!added) + { + Interlocked.Decrement(ref pendingLines); + Interlocked.Increment(ref droppedLines); + } + } + + /// + /// Waits until every queued line has been written to the target, or the timeout elapses. + /// Use this where the test process must see a line before continuing, such as "RunningAndReady". + /// + /// True if every queued line was written before the timeout. + public async Task FlushAsync(TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (Interlocked.Read(ref pendingLines) > 0) + { + if (DateTime.UtcNow >= deadline) return false; + await Task.Delay(10).ConfigureAwait(false); + } + + return true; + } + + /// + /// Synchronous version of , for use just before the process exits. + /// + public bool Flush(TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (Interlocked.Read(ref pendingLines) > 0) + { + if (DateTime.UtcNow >= deadline) return false; + Thread.Sleep(10); + } + + return true; + } + + void WriteLoop() + { + foreach (var line in lines.GetConsumingEnumerable()) + { + try + { + var dropped = Interlocked.Exchange(ref droppedLines, 0); + if (dropped > 0) + { + target.WriteLine($"[{nameof(NonBlockingLineWriter)}] {dropped} log line(s) were dropped because the output was not being read."); + } + + target.WriteLine(line); + target.Flush(); + } + catch (Exception) + { + // Nothing sensible can be done if the target is broken (e.g. the pipe was closed), + // and an exception here must not take down the compat binary. + } + finally + { + Interlocked.Decrement(ref pendingLines); + } + } + } + + public void Dispose() + { + lines.CompleteAdding(); + } + } +} diff --git a/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/TestContextConnectionLog.cs b/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/TestContextConnectionLog.cs index 48dc47f6b..dce596355 100644 --- a/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/TestContextConnectionLog.cs +++ b/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/TestContextConnectionLog.cs @@ -41,7 +41,7 @@ void WriteInternal(LogEvent logEvent) if (logEventLogLevel >= logLevel) { var logMessage = string.Format("{5, 16}: {0}:{1} {2} {3} {4}", logEvent.Type, logEvent.Error, endpoint, Thread.CurrentThread.ManagedThreadId, logEvent.FormattedMessage, name); - Console.WriteLine(logMessage); + NonBlockingConsole.WriteLine(logMessage); } } diff --git a/source/Halibut.TestUtils.CompatBinary.Base/ProxyServiceForwardingRequestToClient.cs b/source/Halibut.TestUtils.CompatBinary.Base/ProxyServiceForwardingRequestToClient.cs index daa092a6a..9fd4d01ce 100644 --- a/source/Halibut.TestUtils.CompatBinary.Base/ProxyServiceForwardingRequestToClient.cs +++ b/source/Halibut.TestUtils.CompatBinary.Base/ProxyServiceForwardingRequestToClient.cs @@ -29,14 +29,14 @@ public static async Task Run(CancellationToken cancellationToken) if (serviceConnectionType is ServiceConnectionType.Polling or ServiceConnectionType.PollingOverWebSocket) { proxyClientAddressToPoll = SettingsHelper.GetSetting("octopusservercommsport"); - Console.WriteLine($"Proxy Service will Poll: {proxyClientAddressToPoll}"); + NonBlockingConsole.WriteLine($"Proxy Service will Poll: {proxyClientAddressToPoll}"); } var serviceAddressToConnectTo = string.Empty; if (serviceConnectionType == ServiceConnectionType.Listening) { serviceAddressToConnectTo = SettingsHelper.GetSetting("realServiceListenAddress"); - Console.WriteLine($"Proxy Service will forward request to: {serviceAddressToConnectTo}"); + NonBlockingConsole.WriteLine($"Proxy Service will forward request to: {serviceAddressToConnectTo}"); } // A Halibut Runtime Client which is the previous version of Halibut the Test is trying to test. @@ -46,7 +46,7 @@ public static async Task Run(CancellationToken cancellationToken) .WithLogFactory(new TestContextLogFactory("PreviousVersionOfClient", SettingsHelper.GetHalibutLogLevel())) .Build()) { - Console.WriteLine("Client will trust: " + serviceCert.Thumbprint); + NonBlockingConsole.WriteLine("Client will trust: " + serviceCert.Thumbprint); client.Trust(serviceCert.Thumbprint); ServiceEndPoint realServiceEndpoint; @@ -55,16 +55,16 @@ public static async Task Run(CancellationToken cancellationToken) case ServiceConnectionType.Polling: var clientPollingListeningPort = client.Listen(); // Do not change the log message as it is used by the HalibutTestBinaryRunners - Console.WriteLine("Polling listener is listening on port: " + clientPollingListeningPort); + NonBlockingConsole.WriteLine("Polling listener is listening on port: " + clientPollingListeningPort); realServiceEndpoint = new ServiceEndPoint(new Uri("poll://SQ-TENTAPOLL"), serviceCert.Thumbprint, proxyDetails); break; case ServiceConnectionType.PollingOverWebSocket: using (await TcpPortHelper.WaitForLock(cancellationToken)) { var webSocketListeningPort = WebSocketListeningPort(client, out var webSocketListeningUrl, cancellationToken); - Console.WriteLine($"WebSocket Polling listener is listening on: {webSocketListeningUrl}"); + NonBlockingConsole.WriteLine($"WebSocket Polling listener is listening on: {webSocketListeningUrl}"); // Do not change the log message as it is used by the HalibutTestBinaryRunners - Console.WriteLine("Polling listener is listening on port: " + webSocketListeningPort); + NonBlockingConsole.WriteLine("Polling listener is listening on port: " + webSocketListeningPort); realServiceEndpoint = new ServiceEndPoint(new Uri("poll://SQ-TENTAPOLL"), serviceCert.Thumbprint, proxyDetails); } break; @@ -95,17 +95,17 @@ public static async Task Run(CancellationToken cancellationToken) case ServiceConnectionType.Listening: var proxyServiceListeningPort = proxyService.Listen(); // Do not change the log message as it is used by the HalibutTestBinaryRunners - Console.WriteLine($"Listening on port: {proxyServiceListeningPort}"); + NonBlockingConsole.WriteLine($"Listening on port: {proxyServiceListeningPort}"); proxyService.Trust(clientCert.Thumbprint); break; default: throw new ArgumentOutOfRangeException(); } - Console.WriteLine("RunningAndReady"); - Console.WriteLine("Will Now sleep"); + NonBlockingConsole.WriteLine("RunningAndReady"); + NonBlockingConsole.WriteLine("Will Now sleep"); - await Console.Out.FlushAsync(); + await NonBlockingConsole.FlushAsync(); // Run until the Program is terminated await StayAliveUntilHelper.WaitUntilSignaledToDie(); @@ -125,7 +125,7 @@ public static int WebSocketListeningPort(HalibutRuntime client, out string webSo } catch (HttpListenerException e) { - Console.WriteLine($"Failed to listen for websocket, trying again. {e}"); + NonBlockingConsole.WriteLine($"Failed to listen for websocket, trying again. {e}"); } } diff --git a/source/Halibut.TestUtils.CompatBinary.Base/Services/FixHungWebSocketsHack.cs b/source/Halibut.TestUtils.CompatBinary.Base/Services/FixHungWebSocketsHack.cs index 783da8e25..7f48f1db8 100644 --- a/source/Halibut.TestUtils.CompatBinary.Base/Services/FixHungWebSocketsHack.cs +++ b/source/Halibut.TestUtils.CompatBinary.Base/Services/FixHungWebSocketsHack.cs @@ -25,6 +25,9 @@ public static void EnableHack() { Thread.Sleep(10000); // Print anything to console. + // This deliberately writes to the console directly rather than through NonBlockingConsole, + // since it is the write itself (on this thread) that appears to unstick the CLR. It is on its + // own background thread, so blocking here cannot freeze the Halibut protocol. Console.WriteLine("Compat binary is still running"); } }); diff --git a/source/Halibut.TestUtils.CompatBinary.Base/Services/StayAliveUntilHelper.cs b/source/Halibut.TestUtils.CompatBinary.Base/Services/StayAliveUntilHelper.cs index 6661201bd..f76bff658 100644 --- a/source/Halibut.TestUtils.CompatBinary.Base/Services/StayAliveUntilHelper.cs +++ b/source/Halibut.TestUtils.CompatBinary.Base/Services/StayAliveUntilHelper.cs @@ -2,6 +2,7 @@ using System.IO; using System.Threading; using System.Threading.Tasks; +using Halibut.TestUtils.SampleProgram.Base.LogUtils; namespace Halibut.TestUtils.SampleProgram.Base.Services { @@ -19,7 +20,7 @@ public static async Task WaitUntilSignaledToDie() #endif using (var fileStreamLock = new FileStream(stayAliveFile, FileMode.Open, FileAccess.Read, FileShare.None)) { - Console.WriteLine("Got a lock, going to die now"); + NonBlockingConsole.WriteLine("Got a lock, going to die now"); fileStreamLock.Dispose(); } @@ -29,6 +30,7 @@ public static async Task WaitUntilSignaledToDie() } finally { + NonBlockingConsole.FlushBeforeExit(); Environment.Exit(0); } } diff --git a/source/Halibut.TestUtils.CompatBinary.Base/SettingsHelper.cs b/source/Halibut.TestUtils.CompatBinary.Base/SettingsHelper.cs index 5deec8ffb..180d3f023 100644 --- a/source/Halibut.TestUtils.CompatBinary.Base/SettingsHelper.cs +++ b/source/Halibut.TestUtils.CompatBinary.Base/SettingsHelper.cs @@ -1,6 +1,7 @@ using System; using System.Security.Cryptography.X509Certificates; using Halibut.Logging; +using Halibut.TestUtils.SampleProgram.Base.LogUtils; using Halibut.Transport.Proxy; namespace Halibut.TestUtils.SampleProgram.Base @@ -18,7 +19,7 @@ static class SettingsHelper var port = GetSetting("proxydetails_port"); var type = GetSetting("proxydetails_type"); - Console.WriteLine($"Using Proxy Details Hot:{host} Port:{port} Type:{type} Username:{userName} Password:{password}"); + NonBlockingConsole.WriteLine($"Using Proxy Details Hot:{host} Port:{port} Type:{type} Username:{userName} Password:{password}"); return new ProxyDetails(host, AsInt(port), AsPortType(type), userName, password); @@ -33,7 +34,7 @@ int AsInt(string s) } } - Console.WriteLine("Not using an Http Proxy"); + NonBlockingConsole.WriteLine("Not using an Http Proxy"); return null; } @@ -122,7 +123,7 @@ public static string CompatBinaryStayAliveLockFile() throw new Exception("Env var CompatBinaryStayAliveFilePath must be set"); } - Console.WriteLine($"Will die when the following file can be locked or is deleted '{stayAliveFilePath}'."); + NonBlockingConsole.WriteLine($"Will die when the following file can be locked or is deleted '{stayAliveFilePath}'."); return stayAliveFilePath; } diff --git a/source/Halibut.Tests/Halibut.Tests.csproj b/source/Halibut.Tests/Halibut.Tests.csproj index 526854c7d..e59695d8f 100644 --- a/source/Halibut.Tests/Halibut.Tests.csproj +++ b/source/Halibut.Tests/Halibut.Tests.csproj @@ -17,6 +17,10 @@ net10.0 + + + + PreserveNewest diff --git a/source/Halibut.Tests/Support/BackwardsCompatibility/NonBlockingLineWriterFixture.cs b/source/Halibut.Tests/Support/BackwardsCompatibility/NonBlockingLineWriterFixture.cs new file mode 100644 index 000000000..b40ed816a --- /dev/null +++ b/source/Halibut.Tests/Support/BackwardsCompatibility/NonBlockingLineWriterFixture.cs @@ -0,0 +1,144 @@ +using System; +using System.Diagnostics; +using System.IO; +using System.Linq; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using FluentAssertions; +using Halibut.TestUtils.SampleProgram.Base.LogUtils; +using NUnit.Framework; + +namespace Halibut.Tests.Support.BackwardsCompatibility +{ + public class NonBlockingLineWriterFixture : BaseTest + { + [Test] + public async Task WritesLinesInOrder() + { + var target = new StringWriter(); + using var writer = new NonBlockingLineWriter(target); + + for (var i = 0; i < 100; i++) writer.WriteLine($"line {i}"); + + (await writer.FlushAsync(TimeSpan.FromSeconds(10))).Should().BeTrue(); + + ReadLines(target).Should().Equal(Enumerable.Range(0, 100).Select(i => $"line {i}")); + } + + [Test] + public async Task DoesNotBlockWhenTheTargetIsBlocked_AndReportsDroppedLinesOnceUnblocked() + { + const int capacity = 10; + using var target = new BlockingTextWriter(); + using var writer = new NonBlockingLineWriter(target, capacity); + + // The first line is taken by the writer thread, which then blocks on the target. + writer.WriteLine("first"); + target.WaitUntilAWriteIsBlocked(TimeSpan.FromSeconds(10)).Should().BeTrue(); + + var stopwatch = Stopwatch.StartNew(); + for (var i = 0; i < capacity + 50; i++) writer.WriteLine($"queued {i}"); + stopwatch.Stop(); + + stopwatch.Elapsed.Should().BeLessThan(TimeSpan.FromSeconds(5), "writing must never wait on a blocked target"); + writer.DroppedLines.Should().Be(50); + + target.Unblock(); + (await writer.FlushAsync(TimeSpan.FromSeconds(10))).Should().BeTrue(); + writer.WriteLine("after unblocking"); + (await writer.FlushAsync(TimeSpan.FromSeconds(10))).Should().BeTrue(); + + var lines = ReadLines(target.Written); + lines.Should().StartWith(new[] { "first" }); + lines.Should().ContainInOrder(Enumerable.Range(0, capacity).Select(i => $"queued {i}")); + lines.Should().Contain(l => l.Contains("50 log line(s) were dropped")); + lines.Last().Should().Be("after unblocking"); + writer.DroppedLines.Should().Be(0); + } + + [Test] + public async Task FlushReturnsFalseWhenTheTargetStaysBlocked() + { + using var target = new BlockingTextWriter(); + using var writer = new NonBlockingLineWriter(target); + + writer.WriteLine("stuck"); + target.WaitUntilAWriteIsBlocked(TimeSpan.FromSeconds(10)).Should().BeTrue(); + + (await writer.FlushAsync(TimeSpan.FromMilliseconds(200))).Should().BeFalse(); + writer.Flush(TimeSpan.FromMilliseconds(200)).Should().BeFalse(); + + target.Unblock(); + (await writer.FlushAsync(TimeSpan.FromSeconds(10))).Should().BeTrue(); + } + + [Test] + public async Task KeepsWritingWhenTheTargetThrows() + { + var target = new ThrowOnceTextWriter(); + using var writer = new NonBlockingLineWriter(target); + + writer.WriteLine("throws"); + writer.WriteLine("written"); + (await writer.FlushAsync(TimeSpan.FromSeconds(10))).Should().BeTrue(); + + ReadLines(target.Written).Should().Equal("written"); + } + + static string[] ReadLines(StringWriter writer) => + writer.ToString().Split(new[] { Environment.NewLine }, StringSplitOptions.RemoveEmptyEntries); + + /// + /// Simulates a stdout pipe that nobody is reading: writes block until is called. + /// + class BlockingTextWriter : TextWriter + { + readonly ManualResetEventSlim unblocked = new(false); + readonly ManualResetEventSlim aWriteIsBlocked = new(false); + + public StringWriter Written { get; } = new(); + public override Encoding Encoding => Encoding.UTF8; + + public bool WaitUntilAWriteIsBlocked(TimeSpan timeout) => aWriteIsBlocked.Wait(timeout); + public void Unblock() => unblocked.Set(); + + public override void WriteLine(string? value) + { + if (!unblocked.IsSet) + { + aWriteIsBlocked.Set(); + unblocked.Wait(); + } + + lock (Written) Written.WriteLine(value); + } + + protected override void Dispose(bool disposing) + { + // Never leave the writer thread blocked after a test. + unblocked.Set(); + base.Dispose(disposing); + } + } + + class ThrowOnceTextWriter : TextWriter + { + bool hasThrown; + + public StringWriter Written { get; } = new(); + public override Encoding Encoding => Encoding.UTF8; + + public override void WriteLine(string? value) + { + if (!hasThrown) + { + hasThrown = true; + throw new IOException("The pipe has been ended."); + } + + Written.WriteLine(value); + } + } + } +}