diff --git a/src/D2BotNG/Program.cs b/src/D2BotNG/Program.cs index b09adc7..76a3c8f 100644 --- a/src/D2BotNG/Program.cs +++ b/src/D2BotNG/Program.cs @@ -124,7 +124,9 @@ private static void Main(string[] args) // EngineHostedService.StartAsync (and any handoff RehydrateAsync inside it) reads // MessageWindow.Handle, so it must be valid by then. In GUI mode, MainForm no // longer switches the handle when it loads — the message-only window owns it - // for the full process lifetime. + // for the full process lifetime. The window runs on its own pump thread, so this + // returns once the handle exists and nothing here shares a pump with it: senders + // block on SendMessageW, and the UI thread below is about to host WebView2. var messageWindow = app.Services.GetRequiredService(); messageWindow.CreateMessageOnlyWindow(); diff --git a/src/D2BotNG/Services/D2BSMessageHandler.cs b/src/D2BotNG/Services/D2BSMessageHandler.cs index 3ff9164..f98abb2 100644 --- a/src/D2BotNG/Services/D2BSMessageHandler.cs +++ b/src/D2BotNG/Services/D2BSMessageHandler.cs @@ -77,8 +77,24 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) { _logger.LogInformation("D2BS message handler started"); - await foreach (var msg in _messageWindow.Messages.ReadAllAsync(stoppingToken)) + await foreach (var raw in _messageWindow.Messages.ReadAllAsync(stoppingToken)) { + // Decoding and JSON parsing happen here rather than in the WndProc: the sending game + // is blocked in SendMessageW for the whole of that call, and the rest of the fleet is + // queued behind it. Parse returns null for a payload it cannot read (already logged), + // and the buffer goes back to the pool either way. + D2BSMessage? msg; + try + { + msg = _messageWindow.Parse(raw); + } + finally + { + raw.Release(); + } + + if (msg == null) continue; + try { await HandleMessageAsync(msg); diff --git a/src/D2BotNG/Windows/MessageWindow.cs b/src/D2BotNG/Windows/MessageWindow.cs index 2e7a487..5d28060 100644 --- a/src/D2BotNG/Windows/MessageWindow.cs +++ b/src/D2BotNG/Windows/MessageWindow.cs @@ -1,3 +1,4 @@ +using System.Buffers; using System.Collections.Concurrent; using System.Runtime.InteropServices; using System.Text; @@ -32,10 +33,29 @@ public enum MessageType /// Created early in startup so the HWND is stable for the full process lifetime /// (handoff rehydration relies on this). /// +/// +/// The window lives on its own dedicated pump thread, and its WndProc does nothing but +/// copy the payload and queue it. Both halves of that matter, because a sender is blocked +/// for the whole of the WndProc: D2BS sends WM_COPYDATA with a bare SendMessageW +/// (no timeout), so the game's thread stalls until this process's pump has serviced it. +/// +/// The window used to be created on the main thread, which then went on to run +/// Application.Run — so every game was queueing behind WebView2, WinForms painting +/// and the titlebar drag loop, on top of each other. And the WndProc itself decoded UTF-8, +/// scanned the whole payload for "heartBeat" and ran a full JsonSerializer.Deserialize +/// of an envelope whose single argument is the entire escaped characterState snapshot — +/// all of it O(payload), all of it with N games waiting their turn. That is what made the +/// per-second capture hitch scale with instance count (d2bsng#11). +/// +/// A dedicated thread also means headless mode has a real pump. It previously had none at +/// all and worked only because Main is [STAThread] and a managed blocking wait +/// on an STA thread happens to dispatch inter-thread sent messages. +/// public class MessageWindow : IDisposable { private readonly ILogger _logger; - private readonly Channel _messageChannel; + private readonly IHostApplicationLifetime _lifetime; + private readonly Channel _messageChannel; /// /// When each sender last reported in, stamped on the message pump. Liveness must not be a @@ -47,15 +67,25 @@ public class MessageWindow : IDisposable private nint _wndProcPtr; private WndProcDelegate? _wndProcDelegate; - private bool _disposed; + private Thread? _pumpThread; + /// Written by Dispose, read on the pump thread — hence volatile. + private volatile bool _disposed; - public MessageWindow(ILogger logger) + public MessageWindow(ILogger logger, IHostApplicationLifetime lifetime) { _logger = logger; - _messageChannel = Channel.CreateUnbounded(new UnboundedChannelOptions + _lifetime = lifetime; + _messageChannel = Channel.CreateUnbounded(new UnboundedChannelOptions { - SingleReader = false, - SingleWriter = true + // One reader: the consumer owns each payload buffer until it releases it, and + // per-sender ordering only holds while a single reader drains the queue. + SingleReader = true, + SingleWriter = true, + // Stated rather than left to the default, because the whole point of this class + // rests on it: true would run the waiting consumer's continuation inline on + // TryWrite, which puts the parse and the SQLite capture ingest back on the pump + // thread — with the sending game blocked in SendMessageW for all of it. + AllowSynchronousContinuations = false }); } @@ -65,9 +95,11 @@ public MessageWindow(ILogger logger) public nint Handle { get; private set; } /// - /// Channel reader for processing incoming D2BS messages. + /// Channel reader for processing incoming D2BS messages. Items are raw payloads: decode and + /// parse them with , then hand the buffer back with + /// . /// - public ChannelReader Messages => _messageChannel.Reader; + public ChannelReader Messages => _messageChannel.Reader; /// /// When the given sender last sent a heartbeat, as observed on the message pump. @@ -82,8 +114,9 @@ public bool TryGetLastHeartbeat(nint senderHandle, out DateTime at) => public void ForgetHandle(nint senderHandle) => _lastHeartbeatAt.TryRemove(senderHandle, out _); /// - /// Creates the message-only window. Call once from Program.Main before any hosted - /// service runs — handoff rehydration reads Handle. + /// Creates the message-only window on its own pump thread and blocks until its handle is + /// valid. Call once from Program.Main before any hosted service runs — handoff rehydration + /// reads Handle. /// public void CreateMessageOnlyWindow() { @@ -93,6 +126,52 @@ public void CreateMessageOnlyWindow() return; } + using var ready = new ManualResetEventSlim(false); + Exception? startupError = null; + + _pumpThread = new Thread(() => + { + try + { + CreateWindow(); + } + catch (Exception ex) + { + startupError = ex; + return; + } + finally + { + // Signalled whether or not creation succeeded, so a failure surfaces as the + // exception below instead of hanging startup. + // ReSharper disable once AccessToDisposedClosure — Set happens-before the Wait returns + ready.Set(); + } + + RunMessageLoop(); + }) + { + Name = "D2BotNG WM_COPYDATA pump", + IsBackground = true + }; + + _pumpThread.Start(); + ready.Wait(); + + if (startupError != null) + { + _pumpThread = null; + throw startupError; + } + } + + /// + /// Registers the window class and creates the window. Runs on the pump thread — a window + /// belongs to the thread that created it, and only that thread's loop dispatches its + /// messages. + /// + private void CreateWindow() + { // Keep delegate alive _wndProcDelegate = WndProc; _wndProcPtr = Marshal.GetFunctionPointerForDelegate(_wndProcDelegate); @@ -115,35 +194,96 @@ public void CreateMessageOnlyWindow() } // Create message-only window - Handle = CreateWindowExW( + var handle = CreateWindowExW( 0, className, "D2BotNG", 0, 0, 0, 0, 0, HWND_MESSAGE, 0, GetModuleHandle(null), 0); - if (Handle == 0) + if (handle == 0) { var error = Marshal.GetLastWin32Error(); throw new InvalidOperationException($"Failed to create message window: {error}"); } - _logger.LogDebug("Created message-only window with handle: {Handle}", Handle); + // Published last, and after the log rather than before it: a throwing sink between the + // two would leave a non-zero Handle behind a thread that is about to exit, and the + // re-entry guard in CreateMessageOnlyWindow would then answer a retry with a warning + // and no pump. + _logger.LogDebug("Created message-only window with handle: {Handle}", handle); + Handle = handle; + } + + /// + /// The pump. Runs until the window is destroyed (Dispose posts WM_CLOSE, whose WM_DESTROY + /// posts the WM_QUIT that ends this loop). + /// + /// + /// Any exit that Dispose did not ask for takes the whole process down, which is not + /// dramatics. D2BS sends with a bare SendMessageW and no timeout, so a dead pump + /// behind a live HWND blocks every game forever on its next send — heartbeat thread + /// included, so the watchdog restarts each game into the same wedge while the manager keeps + /// serving the UI and looks healthy. Exiting destroys the window, and a send to a dead HWND + /// fails immediately instead of hanging. This is only a risk because the pump moved off the + /// main thread: it used to BE the process's pump, so its death was the process's death. + /// + private void RunMessageLoop() + { + try + { + while (true) + { + var result = GetMessageW(out var msg, 0, 0, 0); + if (result == 0) + { + break; // WM_QUIT + } + + if (result == -1) + { + _logger.LogError("GetMessage failed on the WM_COPYDATA pump: {Error}", + Marshal.GetLastWin32Error()); + break; + } + + // No TranslateMessage: a message-only window receives no keyboard input. + DispatchMessageW(ref msg); + } + } + catch (Exception ex) + { + _logger.LogCritical(ex, "WM_COPYDATA pump faulted"); + } + + if (_disposed) + { + _logger.LogDebug("WM_COPYDATA pump stopped"); + return; + } + + // Cleared first. The window died with the thread that owned it, and ProfileEngine keeps + // handing this value to games and resending it on a missed heartbeat; left set, the + // whole fleet times out, gets killed as unresponsive, and restarts into the same dead + // HWND on a loop, with nothing in the log but "missed heartbeat". + Handle = 0; + _logger.LogCritical("WM_COPYDATA pump stopped unexpectedly — no game can reach the manager, shutting down"); + _lifetime.StopApplication(); } /// /// Process an incoming WM_COPYDATA message. Call from WndProc. /// + /// + /// Deliberately does no more than stamp a heartbeat or copy the payload out: the sending + /// game is blocked in SendMessageW for exactly as long as this takes, and every + /// other game is queued behind it. The copy itself is unavoidable — the COPYDATASTRUCT + /// buffer is only valid for the duration of the call. Decoding and parsing happen on the + /// consumer, in . + /// public void HandleCopyData(nint wParam, nint lParam) { try { var copyData = Marshal.PtrToStructure(lParam); - var bytes = new byte[copyData.cbData]; - Marshal.Copy(copyData.lpData, bytes, 0, copyData.cbData); - - // Remove null terminator if present - var length = bytes.Length; - while (length > 0 && bytes[length - 1] == 0) length--; - var messageType = (MessageType)copyData.dwData.ToInt64(); // Heartbeats are recorded here and never enqueued. kolbot's dedicated heartbeat @@ -163,38 +303,36 @@ public void HandleCopyData(nint wParam, nint lParam) return; } - var data = Encoding.UTF8.GetString(bytes, 0, length); - - _logger.LogDebug("WM_COPYDATA received: sender={Sender}, type={Type}, len={Len}, data={Data}", - wParam, messageType, copyData.cbData, data); - - // Normalize heartbeat event. Only reachable for a sender that doesn't set the - // 0xBBBB mode (handled by the fast path above) — it goes down the queue and the - // consumer's "heartBeat" case records it, later but correctly. - if (data.Contains("heartBeat")) + if (copyData.cbData < 0) { - data = JsonSerializer.Serialize(new ProfileMessage - { - Function = "heartBeat" - }); + _logger.LogWarning("Ignoring WM_COPYDATA from {Sender} with negative length {Len}", + wParam, copyData.cbData); + return; } + var raw = D2BSRawMessage.Rent(wParam, messageType, copyData.cbData); + var queued = false; try { - var message = new D2BSMessage + if (copyData.cbData > 0) { - SenderHandle = wParam, - Message = JsonSerializer.Deserialize(data)! - }; + Marshal.Copy(copyData.lpData, raw.Buffer, 0, copyData.cbData); + } - if (!_messageChannel.Writer.TryWrite(message)) + queued = _messageChannel.Writer.TryWrite(raw); + if (!queued) { + // Unbounded channel, so only a completed writer gets here (shutdown). _logger.LogWarning("Failed to queue D2BS message"); } } - catch (Exception ex) + finally { - _logger.LogError(ex, "Error handling WM_COPYDATA for data {data}", data); + // Ownership passes to the consumer only once the write lands. Anything that + // throws in between (a null lpData, a logging sink) would otherwise lose the + // rental silently, and a sender repeating the fault at 1Hz would quietly undo + // the pooling this exists for. + if (!queued) raw.Release(); } } catch (Exception ex) @@ -203,13 +341,83 @@ public void HandleCopyData(nint wParam, nint lParam) } } + /// + /// Decodes and parses a queued payload. Runs on the consumer, off the pump. Returns null if + /// the payload is not a message we can read — the error is logged here rather than thrown, + /// since one unreadable message must not stop the queue. Does not release the buffer; the + /// caller owns it either way. + /// + public D2BSMessage? Parse(D2BSRawMessage raw) + { + var data = string.Empty; + try + { + // Remove null terminator if present + var length = raw.Length; + while (length > 0 && raw.Buffer[length - 1] == 0) length--; + + data = Encoding.UTF8.GetString(raw.Buffer, 0, length); + + _logger.LogDebug("WM_COPYDATA received: sender={Sender}, type={Type}, len={Len}, data={Data}", + raw.SenderHandle, raw.Type, raw.Length, data); + + // Normalize heartbeat event. Only reachable for a sender that doesn't set the + // 0xBBBB mode (handled by the fast path in HandleCopyData) — its heartbeat comes + // down the queue and the consumer's "heartBeat" case records it, later but + // correctly. + if (data.Contains("heartBeat")) + { + data = JsonSerializer.Serialize(new ProfileMessage + { + Function = "heartBeat" + }); + } + + return new D2BSMessage + { + SenderHandle = raw.SenderHandle, + Message = JsonSerializer.Deserialize(data)! + }; + } + catch (Exception ex) + { + // Length as well as the text: a failure in the decode itself leaves data empty, and + // then the length is the only thing that says what arrived. + _logger.LogError(ex, "Error handling WM_COPYDATA from {Sender} ({Len} bytes) for data {data}", + raw.SenderHandle, raw.Length, data); + return null; + } + } + private nint WndProc(nint hWnd, uint msg, nint wParam, nint lParam) { - if (msg != WM_COPYDATA) - return DefWindowProcW(hWnd, msg, wParam, lParam); - HandleCopyData(wParam, lParam); - return 1; + switch (msg) + { + case WM_COPYDATA: + try + { + HandleCopyData(wParam, lParam); + } + catch (Exception ex) + { + // HandleCopyData catches its own, so this is the belt to that braces: an + // exception unwinding through DispatchMessageW's native frames would take + // the pump with it, and a dead pump hangs every game (see RunMessageLoop). + _logger.LogError(ex, "Unhandled error dispatching WM_COPYDATA"); + } + return 1; + case WM_CLOSE: + // Dispose posts this. DestroyWindow has thread affinity, so it has to happen + // here on the pump thread rather than in Dispose itself. + DestroyWindow(hWnd); + return 0; + case WM_DESTROY: + PostQuitMessage(0); // ends RunMessageLoop + return 0; + default: + return DefWindowProcW(hWnd, msg, wParam, lParam); + } } public void Dispose() @@ -217,16 +425,92 @@ public void Dispose() if (_disposed) return; _disposed = true; + // DestroyWindow only works from the owning thread, so ask the pump to close itself and + // wait for the loop to unwind. The writer is completed after that, so the pump cannot + // still be enqueueing into a completed channel. + if (Handle != 0) + { + PostMessage(Handle, WM_CLOSE, 0, 0); + } + + if (_pumpThread is { IsAlive: true } && !_pumpThread.Join(TimeSpan.FromSeconds(5))) + { + _logger.LogWarning("WM_COPYDATA pump did not stop within 5s"); + } + + _pumpThread = null; + Handle = 0; + _messageChannel.Writer.Complete(); - if (Handle != 0) + // Drain whatever the consumer will now never see, so pooled buffers go back. + while (_messageChannel.Reader.TryRead(out var raw)) { - DestroyWindow(Handle); - Handle = 0; + raw.Release(); } } } +/// +/// A WM_COPYDATA payload as it came off the wire, queued for the consumer to decode. Buffers +/// come from a pool because a characterState snapshot is hundreds of KB and arrives once a +/// second per running game, which is enough allocation to keep the large object heap busy. +/// +public sealed class D2BSRawMessage +{ + /// + /// Payloads above this are allocated instead of pooled. Comfortably above a full PlugY stash + /// snapshot, and the cap matters because ArrayPool.Create — unlike + /// ArrayPool<T>.Shared — registers no Gen2 trim callback, so whatever a bucket + /// retains it retains for the life of the process. Every doubling of this number doubles the + /// largest bucket's permanent footprint for the sake of an outlier that is rented once. + /// + private const int MaxPooledPayload = 1024 * 1024; + + private static readonly ArrayPool Pool = ArrayPool.Create(MaxPooledPayload, 8); + + private readonly bool _pooled; + private int _released; + + private D2BSRawMessage(nint senderHandle, MessageType type, byte[] buffer, int length, bool pooled) + { + SenderHandle = senderHandle; + Type = type; + Buffer = buffer; + Length = length; + _pooled = pooled; + } + + public nint SenderHandle { get; } + + public MessageType Type { get; } + + /// The payload buffer. May be longer than — it is pooled. + public byte[] Buffer { get; } + + /// How many bytes of the sender actually wrote. + public int Length { get; } + + public static D2BSRawMessage Rent(nint senderHandle, MessageType type, int length) + { + var pooled = length <= MaxPooledPayload; + var buffer = pooled ? Pool.Rent(length) : new byte[length]; + return new D2BSRawMessage(senderHandle, type, buffer, length, pooled); + } + + /// + /// Hands the buffer back. A message has exactly one owner — whoever took it off the channel + /// — and this is idempotent for that owner's benefit, not as a licence to share: returning + /// one buffer twice hands the same array to two renters, which corrupts silently rather than + /// throwing. Interlocked because the guard is worthless if it can itself race. + /// + public void Release() + { + if (!_pooled || Interlocked.Exchange(ref _released, 1) != 0) return; + Pool.Return(Buffer); + } +} + /// /// Represents a JSON message received from D2BS via WM_COPYDATA that was serialized using JSON. /// diff --git a/src/D2BotNG/Windows/NativeMethods.cs b/src/D2BotNG/Windows/NativeMethods.cs index c4ae57e..31476f5 100644 --- a/src/D2BotNG/Windows/NativeMethods.cs +++ b/src/D2BotNG/Windows/NativeMethods.cs @@ -165,6 +165,17 @@ public static extern nint CreateWindowExW( [DllImport("user32.dll")] public static extern bool PostMessage(nint hWnd, uint Msg, nint wParam, nint lParam); + // Blocking message-loop primitives. Only the MessageWindow pump thread uses these — + // returns >0 for a message, 0 for WM_QUIT, -1 for an error. + [DllImport("user32.dll", SetLastError = true)] + public static extern int GetMessageW(out MSG lpMsg, nint hWnd, uint wMsgFilterMin, uint wMsgFilterMax); + + [DllImport("user32.dll")] + public static extern nint DispatchMessageW(ref MSG lpMsg); + + [DllImport("user32.dll")] + public static extern void PostQuitMessage(int nExitCode); + [DllImport("user32.dll", SetLastError = true, CharSet = CharSet.Auto)] public static extern nint SendMessageTimeout( nint hWnd, diff --git a/src/D2BotNG/Windows/NativeTypes.cs b/src/D2BotNG/Windows/NativeTypes.cs index da1347c..494aacc 100644 --- a/src/D2BotNG/Windows/NativeTypes.cs +++ b/src/D2BotNG/Windows/NativeTypes.cs @@ -21,6 +21,24 @@ public struct COPYDATASTRUCT public nint lpData; } + [StructLayout(LayoutKind.Sequential)] + public struct POINT + { + public int x; + public int y; + } + + [StructLayout(LayoutKind.Sequential)] + public struct MSG + { + public nint hwnd; + public uint message; + public nint wParam; + public nint lParam; + public uint time; + public POINT pt; + } + [StructLayout(LayoutKind.Sequential, CharSet = CharSet.Unicode)] public struct WNDCLASSEXW { @@ -135,6 +153,7 @@ public enum SE_OBJECT_TYPE // Window messages (UINT) public const uint WM_NULL = 0x0000; // no-op; used as a liveness ping via SendMessageTimeout + public const uint WM_DESTROY = 0x0002; public const uint WM_CLOSE = 0x0010; public const uint WM_SETTEXT = 0x000C; public const uint WM_COPYDATA = 0x004A;