diff --git a/lsp/source/served/lsp/filereader.d b/lsp/source/served/lsp/filereader.d index 31af8e8..01d9392 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,52 @@ 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 a6cc8bd..8b9998f 100644 --- a/serverbase/source/served/serverbase.d +++ b/serverbase/source/served/serverbase.d @@ -38,6 +38,14 @@ 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; + + /// 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 @@ -483,8 +491,14 @@ mixin template LanguageServerRouter(alias ExtensionModule, LanguageServerConfig int gcCollects, totalGcCollects; StopWatch gcInterval; gcInterval.start(); - - void collectGC() + // 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(bool forceMinimize = false) { import core.memory : GC; @@ -498,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; @@ -507,6 +524,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, @@ -551,6 +570,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) @@ -562,8 +586,33 @@ 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); + // 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(); @@ -571,7 +620,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 and release memory if we can + else if (activitySinceCollect && !busy) + collectGC(true); + else + gcInterval.reset(); } } }