From 77170147593e9e8138007fc021e158206e347253 Mon Sep 17 00:00:00 2001 From: Baruch Even Date: Fri, 31 Jul 2026 14:54:29 +0300 Subject: [PATCH 1/3] serverbase: only collect when something was allocated A bare timer made an idle server re-mark an unchanged heap every interval, using CPU and waking every core. Gate it on gcCollectMinAllocated, plus one collection once the server goes quiet. (cherry picked from commit bf96205d736e23429d65cee13de9b41565e80b88) --- serverbase/source/served/serverbase.d | 30 ++++++++++++++++++++++++++- 1 file changed, 29 insertions(+), 1 deletion(-) diff --git a/serverbase/source/served/serverbase.d b/serverbase/source/served/serverbase.d index a6cc8bd..601d6fc 100644 --- a/serverbase/source/served/serverbase.d +++ b/serverbase/source/served/serverbase.d @@ -38,6 +38,10 @@ struct LanguageServerConfig int gcCollectSeconds = 30; /// ditto int gcMinimizeTimes = 5; + /// Bytes that must have been allocated since the last collection for the + /// periodic collector to run again. One collection also always runs once the + /// server goes quiet. + size_t gcCollectMinAllocated = 128 * 1024 * 1024; } // dumps a performance/GC trace log to served_trace.log @@ -483,6 +487,12 @@ mixin template LanguageServerRouter(alias ExtensionModule, LanguageServerConfig int gcCollects, totalGcCollects; StopWatch gcInterval; gcInterval.start(); + // Nothing is freed between collections, so usedSize growth is the + // amount allocated, from any thread or fiber. + size_t gcUsedAtLastCollect; + // Detects the busy -> quiet edge. Keyed on messages, not live fibers: a + // fiber can block forever on a client reply, which is idle, not busy. + bool activityThisInterval, activitySinceCollect; void collectGC() { @@ -507,6 +517,8 @@ mixin template LanguageServerRouter(alias ExtensionModule, LanguageServerConfig gcSpeed.stop(); auto after = GC.stats(); + gcUsedAtLastCollect = after.usedSize; + activitySinceCollect = false; if (before != after) tracef("GC run in %s. Freed %s bytes (%s bytes allocated, %s bytes available)", gcSpeed.peek, @@ -562,6 +574,9 @@ mixin template LanguageServerRouter(alias ExtensionModule, LanguageServerConfig pushFiber(msg.fiberName, gotRequest(msg)); else pushFiber(msg.fiberName, gotNotify(msg)); + + static if (serverConfig.gcCollectSeconds > 0) + activityThisInterval = activitySinceCollect = true; } Thread.sleep(loopIterationDelay); synchronized (fibersMutex) @@ -571,7 +586,20 @@ mixin template LanguageServerRouter(alias ExtensionModule, LanguageServerConfig { if (gcInterval.peek > serverConfig.gcCollectSeconds.seconds) { - collectGC(); + import core.memory : GC; + + immutable busy = activityThisInterval; + activityThisInterval = false; + + // busy: collect once enough was allocated to be worth marking + if (GC.stats().usedSize >= gcUsedAtLastCollect + + serverConfig.gcCollectMinAllocated) + collectGC(); + // gone quiet: collect once to release what the work dropped + else if (activitySinceCollect && !busy) + collectGC(); + else + gcInterval.reset(); } } } From 8c96aec7350bce5f40cade4fc1103f8b13589460 Mon Sep 17 00:00:00 2001 From: Baruch Even Date: Fri, 31 Jul 2026 14:54:29 +0300 Subject: [PATCH 2/3] serverbase: wait for input instead of polling while idle The loop slept a fixed interval forever, waking ~100x/second with nothing to do. Block on the reader while no request is buffered, no fiber is in flight and no timeout is pending. waitForData reports whether it really waited, so already-buffered bytes keep the old cadence rather than spin. (cherry picked from commit 36d46460563e208f05f588a67e41eb722bfeb4d4) --- lsp/source/served/lsp/filereader.d | 95 +++++++++++++++++++++++++++ lsp/source/served/lsp/jsonrpc.d | 8 +++ serverbase/source/served/serverbase.d | 33 +++++++++- 3 files changed, 135 insertions(+), 1 deletion(-) diff --git a/lsp/source/served/lsp/filereader.d b/lsp/source/served/lsp/filereader.d index 31af8e8..5727e7d 100644 --- a/lsp/source/served/lsp/filereader.d +++ b/lsp/source/served/lsp/filereader.d @@ -1,7 +1,9 @@ module served.lsp.filereader; +import core.sync.condition; import core.sync.mutex; import core.thread; +import core.time : Duration; import std.algorithm; import std.stdio; @@ -31,6 +33,10 @@ version (Windows) class WindowsStdinReader : FileReader closeEvent.reset(); scope (exit) closeEvent.set(); + // let a waiter notice we stopped instead of sitting out its timeout + scope (exit) + synchronized (mutex) + notifyDataAvailable(); auto stdin = GetStdHandle(STD_INPUT_HANDLE); ubyte[4096] buffer; @@ -54,7 +60,10 @@ version (Windows) class WindowsStdinReader : FileReader return; } synchronized (mutex) + { data ~= buffer[0 .. len]; + notifyDataAvailable(); + } break; case WAIT_FAILED: stderr.writeln("stdin read failed ", GetLastError()); @@ -108,6 +117,10 @@ version (Windows) class WindowsFileReader : FileReader closeEvent.reset(); scope (exit) closeEvent.set(); + // let a waiter notice we stopped instead of sitting out its timeout + scope (exit) + synchronized (mutex) + notifyDataAvailable(); ubyte[4096] buffer; @@ -132,7 +145,10 @@ version (Windows) class WindowsFileReader : FileReader continue; } synchronized (mutex) + { data ~= buffer[0 .. numRead]; + notifyDataAvailable(); + } } } @@ -195,6 +211,10 @@ version (Posix) class PosixFileReader : FileReader closeEvent.reset(); scope (exit) closeEvent.setIfInitialized(); + // let a waiter notice we stopped instead of sitting out its timeout + scope (exit) + synchronized (mutex) + notifyDataAvailable(); int fd = stdFile.fileno; ubyte[4096] buffer; @@ -240,7 +260,10 @@ version (Posix) class PosixFileReader : FileReader else { synchronized (mutex) + { data ~= buffer[0 .. len]; + notifyDataAvailable(); + } } } } @@ -261,6 +284,27 @@ abstract class FileReader : Thread super(&run); isDaemon = true; mutex = new Mutex(); + dataAvailable = new Condition(mutex); + } + + /// Blocks until data is appended or `timeout` elapses. + /// Returns: false without waiting when data is already buffered, so a caller + /// driving a loop can tell it must not treat this as a completed wait. + bool waitForData(Duration timeout) + { + synchronized (mutex) + { + if (data.length) + return false; + dataAvailable.wait(timeout); + return true; + } + } + + /// Wakes `waitForData`. Call while holding `mutex`. + protected void notifyDataAvailable() + { + dataAvailable.notifyAll(); } string yieldLine(bool* whileThisIs = null, bool equalToThis = true) @@ -328,6 +372,7 @@ protected: ubyte[] data; Mutex mutex; + Condition dataAvailable; } /// Creates a new FileReader using the GC reading from stdin using a platform @@ -428,3 +473,53 @@ unittest code = readCodeWithBuffer("lsp/source/served/lsp/filereader.d", slice, 16); assert(code == "module served.ls"); } + + +unittest +{ + import core.time : msecs, seconds; + import std.datetime.stopwatch : AutoStart, StopWatch; + + // concrete reader that never reads anything, so `data` is driven by the test + static class TestReader : FileReader + { + override void stop() {} + override bool isReading() { return true; } + protected override void run() {} + + void append(ubyte[] bytes) + { + synchronized (mutex) + { + data ~= bytes; + notifyDataAvailable(); + } + } + } + + auto reader = new TestReader(); + + // nothing buffered: waits, and reports that it waited + auto sw = StopWatch(AutoStart.yes); + assert(reader.waitForData(50.msecs)); + assert(sw.peek >= 40.msecs); + + // data already buffered: must return false *without* waiting, otherwise a + // caller skipping its sleep on the strength of this call would spin + reader.append(cast(ubyte[]) "Content-Length: 99\r\n\r\npartial".dup); + sw = StopWatch(AutoStart.yes); + assert(!reader.waitForData(5.seconds)); + assert(sw.peek < 1.seconds); + + // appending wakes a waiter well before the timeout + auto empty = new TestReader(); + auto waker = new Thread({ + Thread.sleep(30.msecs); + empty.append(cast(ubyte[]) "x".dup); + }); + waker.start(); + sw = StopWatch(AutoStart.yes); + assert(empty.waitForData(5.seconds)); + assert(sw.peek < 2.seconds); + waker.join(); +} diff --git a/lsp/source/served/lsp/jsonrpc.d b/lsp/source/served/lsp/jsonrpc.d index 94fe18a..521f8fb 100644 --- a/lsp/source/served/lsp/jsonrpc.d +++ b/lsp/source/served/lsp/jsonrpc.d @@ -447,6 +447,14 @@ class RPCProcessor : Fiber return resolveWait(i, timeout); } + /// Blocks until the reader has new input or `timeout` elapses, so a caller + /// driving the loop can wait instead of polling. + /// Returns: false without waiting when input is already buffered. + bool waitForInput(Duration timeout) + { + return reader.waitForData(timeout); + } + private: void onData(RequestMessageRaw req) { diff --git a/serverbase/source/served/serverbase.d b/serverbase/source/served/serverbase.d index 601d6fc..aa2db66 100644 --- a/serverbase/source/served/serverbase.d +++ b/serverbase/source/served/serverbase.d @@ -42,6 +42,10 @@ struct LanguageServerConfig /// periodic collector to run again. One collection also always runs once the /// server goes quiet. size_t gcCollectMinAllocated = 128 * 1024 * 1024; + + /// How long to block waiting for input while idle, in milliseconds. Incoming + /// data wakes it immediately. 0 polls instead. + int idleWaitMsecs = 1000; } // dumps a performance/GC trace log to served_trace.log @@ -563,6 +567,11 @@ mixin template LanguageServerRouter(alias ExtensionModule, LanguageServerConfig static if (is(typeof(ExtensionModule.parallelMain))) pushFiber("parallelMain", &ExtensionModule.parallelMain); + // The RPC manager and parallelMain live forever; anything else is work. + size_t permanentFibers; + synchronized (fibersMutex) + permanentFibers = fibers.length; + while (rpc.state != Fiber.State.TERM) { while (rpc.hasData) @@ -578,7 +587,29 @@ mixin template LanguageServerRouter(alias ExtensionModule, LanguageServerConfig static if (serverConfig.gcCollectSeconds > 0) activityThisInterval = activitySinceCollect = true; } - Thread.sleep(loopIterationDelay); + // Wait for input when there is nothing to do at all. Otherwise keep the + // fixed cadence: a live fiber may need resuming to make progress. + static if (serverConfig.idleWaitMsecs > 0) + { + bool nothingToDo; + synchronized (fibersMutex) + nothingToDo = fibers.length <= permanentFibers; + + // created by parallelMain; until then there are no timeouts + if (nothingToDo && timeoutsMutex !is null) + synchronized (timeoutsMutex) + nothingToDo = timeouts.length == 0; + + // keep the fixed cadence unless we really blocked: waitForInput + // returns false when raw bytes are already buffered (a message + // arriving in pieces), and skipping the sleep there would spin + if (!nothingToDo + || !rpc.waitForInput(serverConfig.idleWaitMsecs.msecs)) + Thread.sleep(loopIterationDelay); + } + else + Thread.sleep(loopIterationDelay); + synchronized (fibersMutex) fibers.call(); From 72aadca80370f2deb3c2d7a353371af5849b6110 Mon Sep 17 00:00:00 2001 From: Baruch Even Date: Fri, 31 Jul 2026 14:54:58 +0300 Subject: [PATCH 3/3] serverbase: minimize on the going-quiet collect Release memory if we can once things have settled; an idle server no longer collects often enough to reach gcMinimizeTimes on its own. (cherry picked from commit 55abe52af1bb6440f1f602a2376fffa24e810f25) --- lsp/source/served/lsp/filereader.d | 1 - serverbase/source/served/serverbase.d | 11 +++++++---- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/lsp/source/served/lsp/filereader.d b/lsp/source/served/lsp/filereader.d index 5727e7d..01d9392 100644 --- a/lsp/source/served/lsp/filereader.d +++ b/lsp/source/served/lsp/filereader.d @@ -474,7 +474,6 @@ unittest assert(code == "module served.ls"); } - unittest { import core.time : msecs, seconds; diff --git a/serverbase/source/served/serverbase.d b/serverbase/source/served/serverbase.d index aa2db66..8b9998f 100644 --- a/serverbase/source/served/serverbase.d +++ b/serverbase/source/served/serverbase.d @@ -498,7 +498,7 @@ mixin template LanguageServerRouter(alias ExtensionModule, LanguageServerConfig // fiber can block forever on a client reply, which is idle, not busy. bool activityThisInterval, activitySinceCollect; - void collectGC() + void collectGC(bool forceMinimize = false) { import core.memory : GC; @@ -512,7 +512,10 @@ mixin template LanguageServerRouter(alias ExtensionModule, LanguageServerConfig static if (serverConfig.gcMinimizeTimes > 0) { gcCollects++; - if (gcCollects >= serverConfig.gcMinimizeTimes) + // release memory if we can once things have settled; an idle + // server no longer collects often enough to reach + // gcMinimizeTimes on its own + if (forceMinimize || gcCollects >= serverConfig.gcMinimizeTimes) { GC.minimize(); gcCollects = 0; @@ -626,9 +629,9 @@ mixin template LanguageServerRouter(alias ExtensionModule, LanguageServerConfig if (GC.stats().usedSize >= gcUsedAtLastCollect + serverConfig.gcCollectMinAllocated) collectGC(); - // gone quiet: collect once to release what the work dropped + // gone quiet: collect once and release memory if we can else if (activitySinceCollect && !busy) - collectGC(); + collectGC(true); else gcInterval.reset(); }