From 0852becc758aee2305abebe421e1ce812d209794 Mon Sep 17 00:00:00 2001 From: Resurrected Trader Date: Sat, 5 Sep 2026 08:50:48 +0100 Subject: [PATCH 1/2] perf: Take WM_COPYDATA off the UI thread and out of the WndProc Games send the manager WM_COPYDATA with a bare blocking SendMessageW, so a game's thread stalls for exactly as long as our WndProc takes, and the whole fleet queues on one pump. Two things made that expensive. The window was created on the main thread, which then ran Application.Run -- so every game was queueing behind WinForms painting, the titlebar drag loop and WebView2, on top of each other. It now lives on its own pump thread that does nothing else. That also gives headless mode a real pump: it had none, and worked only because Main is [STAThread] and a managed blocking wait on an STA thread happens to dispatch inter-thread sent messages. And the WndProc itself decoded UTF-8, scanned the whole payload for "heartBeat" and ran a full JsonSerializer.Deserialize of an envelope whose one argument is the entire escaped characterState snapshot -- all O(payload), all with N games waiting. It now stamps the heartbeat or copies the bytes into a pooled buffer and returns; the consumer decodes and parses. The copy is the only part that has to happen there, since the COPYDATASTRUCT buffer dies with the call. Together that is what made the per-second capture hitch scale with instance count (ResurrectedTrader/d2bsng#11). Moving the pump off the main thread means its death is no longer the process's death, so an unexpected exit now clears Handle and stops the application: a dead pump behind a live HWND blocks every game forever on its next send, heartbeat included, so the watchdog would restart each game into the same wedge while the manager kept serving the UI and looked healthy. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01CJNhrx7ZamSEaeREus4aKS --- src/D2BotNG/Program.cs | 4 +- src/D2BotNG/Services/D2BSMessageHandler.cs | 18 +- src/D2BotNG/Windows/MessageWindow.cs | 371 ++++++++++++++++++--- src/D2BotNG/Windows/NativeMethods.cs | 11 + src/D2BotNG/Windows/NativeTypes.cs | 19 ++ 5 files changed, 375 insertions(+), 48 deletions(-) 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..72c002f 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,14 +67,19 @@ 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, + // 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 }); } @@ -65,9 +90,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 +109,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 +121,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 +189,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 +298,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 +336,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 +420,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; From 74304f9724633670724e2716a0956270e9c6347f Mon Sep 17 00:00:00 2001 From: Resurrected Trader Date: Sat, 5 Sep 2026 11:49:49 +0100 Subject: [PATCH 2/2] docs: Pin AllowSynchronousContinuations on the WM_COPYDATA queue It is already the default, but the previous commit's whole premise rests on it: true would run the waiting consumer's continuation inline on TryWrite, putting the parse and the SQLite capture ingest back on the pump thread with the sending game blocked in SendMessageW for all of it. Stated explicitly so that flipping it reads as the decision it would be. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01CJNhrx7ZamSEaeREus4aKS --- src/D2BotNG/Windows/MessageWindow.cs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/D2BotNG/Windows/MessageWindow.cs b/src/D2BotNG/Windows/MessageWindow.cs index 72c002f..5d28060 100644 --- a/src/D2BotNG/Windows/MessageWindow.cs +++ b/src/D2BotNG/Windows/MessageWindow.cs @@ -80,7 +80,12 @@ public MessageWindow(ILogger logger, IHostApplicationLifetime lif // 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 + 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 }); }