serverbase: only collect when something was allocated

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();
+}

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)
 	{

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,16 +586,54 @@ 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();
 
 			static if (serverConfig.gcCollectSeconds > 0)
 			{
 				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();
 				}
 			}
 		}