diff --git a/SW.Serverless.Installer.UnitTests/LocalLimitsTests.cs b/SW.Serverless.Installer.UnitTests/LocalLimitsTests.cs new file mode 100644 index 0000000..249be6b --- /dev/null +++ b/SW.Serverless.Installer.UnitTests/LocalLimitsTests.cs @@ -0,0 +1,211 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.IO; +using System.Linq; +using System.Threading.Tasks; +using SW.Serverless.Runtimes; +using SW.Serverless.Tooling; +using SW.Serverless.Tooling.Building; +using SW.Serverless.Tooling.Conformance; + +namespace SW.Serverless.Installer.UnitTests; + +/// +/// Memory and CPU ceilings on an adapter run by LocalAdapterHost, as a tool trying drafts on a +/// shared server sets them: an adapter over one is stopped and the call says which limit, and an +/// adapter run without limits runs as it always has. +/// +[TestClass] +public class LocalLimitsTests +{ + const long Limit = 150L * 1024 * 1024; + + static string RepositoryRoot() + { + var dir = new DirectoryInfo(AppContext.BaseDirectory); + while (dir != null && !File.Exists(Path.Combine(dir.FullName, "SW.Serverless.sln"))) dir = dir.Parent; + return dir!.FullName; + } + + static string WorkFolder() + { + var folder = Path.Combine(RepositoryRoot(), ".test-work", "limits-" + Guid.NewGuid().ToString("N")); + Directory.CreateDirectory(folder); + return folder; + } + + [ClassCleanup] + public static void Cleanup() + { + try { Directory.Delete(Path.Combine(RepositoryRoot(), ".test-work"), true); } catch { } + } + + // Allocates in steps, so a watchdog sampling every second sees it climb; touched (b"x" * n), so + // it is resident and not merely reserved. Capped, so a limit that failed costs a second, not the machine. + const string PythonHog = """ + import os + import time + + import sw_serverless as sw + + + class Hog: + def __init__(self): + sw.expect("HogOnStart", "0", required=False) + + def start(self): + if sw.value_of("HogOnStart") != "0": + self.allocate(1000) + + @sw.command("Allocate") + def allocate(self, megabytes: int) -> int: + held = [] + for _ in range(megabytes // 10): + held.append(b"x" * (10 << 20)) + time.sleep(0.05) + return len(held) * 10 + + @sw.command("Spin") + def spin(self, seconds: int) -> int: + end, turns = time.monotonic() + seconds, 0 + while time.monotonic() < end: + turns += 1 + return turns + + @sw.command("Limit") + def limit(self) -> str: + return os.environ.get("SW_SERVERLESS_MEMORY_LIMIT_BYTES", "none") + + + if __name__ == "__main__": + sw.run(Hog) + """; + + // V8 heap, not Buffers: the old space is what --max-old-space-size holds. + const string NodeHog = """ + const { run } = require("@simplyworks/sw-serverless"); + + class Hog { + static commands = { Allocate: { method: "allocate", input: "json", output: "json" } }; + + async allocate(megabytes) { + const held = []; + for (let i = 0; i < megabytes; i++) { + held.push(new Array(128 * 1024).fill(i + 0.5)); + if (i % 10 === 0) await new Promise((r) => setTimeout(r, 50)); + } + return held.length; + } + } + + run(Hog); + """; + + static async Task BuildAsync(string runtime, string entry, string code) + { + var project = Path.Combine(WorkFolder(), "hog"); + Directory.CreateDirectory(project); + File.WriteAllText(Path.Combine(project, entry), code); + File.WriteAllText(Path.Combine(project, "adapter.json"), + $$"""{ "id": "test.hog.{{runtime}}", "version": "0.1.0", "runtime": "{{runtime}}", "entry": "{{entry}}" }"""); + var result = await PackageBuilder.BuildAsync(new BuildRequest { ProjectDirectory = project, IncludeSource = false }); + Assert.IsTrue(result.Succeeded, string.Join("; ", result.Problems)); + return result.PackageDirectory; + } + + static readonly Lazy> PythonPackage = new(() => BuildAsync("python", "main.py", PythonHog)); + + static Task Start(string package, LocalAdapterLimits? limits, IDictionary? settings = null) => + LocalAdapterHost.StartAsync(package, settings ?? new Dictionary(), commandTimeoutSeconds: 30, limits: limits); + + static async Task FailureOf(Func call) + { + try { await call(); } + catch (Exception ex) { return ex; } + Assert.Fail("the call succeeded"); + return null!; + } + + [TestMethod] + public async Task A_Python_adapter_over_its_memory_limit_is_stopped_and_the_call_names_the_limit() + { + await using var host = await Start(await PythonPackage.Value, new LocalAdapterLimits { MemoryLimitBytes = Limit }); + + Assert.AreEqual(Limit.ToString(), await host.CallAsync("Limit", null), "the SDK is handed the limit"); + + var watch = Stopwatch.StartNew(); + var failure = await FailureOf(() => host.CallAsync("Allocate", 1000)); + Console.WriteLine($"{failure.GetType().Name} after {watch.Elapsed}: {failure.Message}"); + + // On Linux RLIMIT_DATA stops the allocation inside the adapter (MemoryError); elsewhere the + // watchdog kills it. Either way the caller is told about the limit, not about a broken pipe. + StringAssert.Contains(failure.Message, "memory limit of 150 MB"); + Assert.IsTrue(watch.Elapsed < TimeSpan.FromSeconds(10), $"stopped after {watch.Elapsed}"); + } + + [TestMethod] + public async Task Without_limits_a_Python_adapter_runs_as_it_always_has() + { + await using var host = await Start(await PythonPackage.Value, limits: null); + + Assert.AreEqual("none", await host.CallAsync("Limit", null), "no limit reaches the adapter"); + Assert.AreEqual("200", await host.CallAsync("Allocate", 200)); + + // An empty limits object is no limits. + await using var empty = await Start(await PythonPackage.Value, new LocalAdapterLimits()); + Assert.AreEqual("none", await empty.CallAsync("Limit", null)); + } + + [TestMethod] + public async Task A_Python_adapter_over_its_CPU_limit_is_stopped_and_the_call_names_the_limit() + { + // One core flat out, against a ceiling of half a core. + var halfACore = 50.0 / Environment.ProcessorCount; + await using var host = await Start(await PythonPackage.Value, new LocalAdapterLimits + { + CpuPercentLimit = halfACore, + CpuLimitSamples = 2, + SampleInterval = TimeSpan.FromMilliseconds(500), + }); + + var watch = Stopwatch.StartNew(); + var failure = await Assert.ThrowsExceptionAsync(() => host.CallAsync("Spin", 20)); + Console.WriteLine($"after {watch.Elapsed}: {failure.Message}"); + + StringAssert.Contains(failure.Message, "CPU limit"); + Assert.IsTrue(failure.LimitExceeded); + Assert.IsTrue(watch.Elapsed < TimeSpan.FromSeconds(15), $"stopped after {watch.Elapsed}"); + } + + [TestMethod] + public async Task A_Node_adapter_over_its_memory_limit_is_stopped_and_the_call_names_the_limit() + { + var package = await BuildAsync("node", "main.js", NodeHog); + await using var host = await Start(package, new LocalAdapterLimits { MemoryLimitBytes = Limit }); + + var watch = Stopwatch.StartNew(); + var failure = await Assert.ThrowsExceptionAsync(() => host.CallAsync("Allocate", 1000)); + Console.WriteLine($"after {watch.Elapsed}: {failure.Message}"); + + StringAssert.Contains(failure.Message, "memory limit of 150 MB"); + Assert.IsTrue(failure.LimitExceeded); + } + + [TestMethod] + public async Task The_conformance_kit_runs_an_adapter_under_its_limits() + { + var report = await new ConformanceRunner().RunAsync(new ConformanceOptions + { + PackageDirectory = await PythonPackage.Value, + Settings = new Dictionary { ["HogOnStart"] = "1" }, + CommandTimeoutSeconds = 30, + Limits = new LocalAdapterLimits { MemoryLimitBytes = Limit }, + }); + Console.WriteLine(string.Join(Environment.NewLine, report.Checks.Select(c => $"{c.Outcome,-7} {c.Name} {c.Detail}"))); + + var check = report.Checks.Single(c => c.Name == "an unknown command is refused"); + Assert.AreEqual(CheckOutcome.Failed, check.Outcome); + StringAssert.Contains(check.Detail, "memory limit of 150 MB"); + } +} diff --git a/SW.Serverless.Tooling/Conformance/ConformanceOptions.cs b/SW.Serverless.Tooling/Conformance/ConformanceOptions.cs index 54a4134..588eeff 100644 --- a/SW.Serverless.Tooling/Conformance/ConformanceOptions.cs +++ b/SW.Serverless.Tooling/Conformance/ConformanceOptions.cs @@ -29,6 +29,12 @@ public class ConformanceOptions /// Where the host finds python3, node and dotnet. public Runtimes.AdapterRuntimeOptions Runtimes { get; set; } = new(); + /// + /// Memory and CPU ceilings to run the adapter under while it is checked; null (the default) + /// for none. A call that crosses one fails its check with the limit named. + /// + public LocalAdapterLimits Limits { get; set; } + /// Progress, one line at a time. public Action Log { get; set; } = _ => { }; } diff --git a/SW.Serverless.Tooling/Conformance/ConformanceRunner.cs b/SW.Serverless.Tooling/Conformance/ConformanceRunner.cs index f84bd97..1b2639e 100644 --- a/SW.Serverless.Tooling/Conformance/ConformanceRunner.cs +++ b/SW.Serverless.Tooling/Conformance/ConformanceRunner.cs @@ -78,7 +78,7 @@ async Task RunCoreAsync(ConformanceOptions options, ConformanceReport report, st try { session = await LocalAdapterHost.StartAsync(options.PackageDirectory, options.Settings, options.Runtimes, - options.CommandTimeoutSeconds, work); + options.CommandTimeoutSeconds, work, options.Limits); report.Pass("starts", manifest.IsResident ? "resident, attached over gRPC" : "classic session"); } catch (Exception ex) @@ -302,6 +302,12 @@ static async Task CheckUnknownCommandAsync(LocalAdapterHost session, Conformance report.Fail("an unknown command is refused", "it didn't answer a command it doesn't have"); return; } + catch (SW.Serverless.Resident.AdapterStoppedException ex) + { + // Not a refusal: it was stopped — at a memory or CPU limit, say — and answers nothing now. + report.Fail("an unknown command is refused", ex.Message); + return; + } catch (Exception) when (attempt == 1) { // Refused; the second attempt shows it is still answering after that. diff --git a/SW.Serverless.Tooling/LocalAdapterHost.cs b/SW.Serverless.Tooling/LocalAdapterHost.cs index 90cbb34..88eed31 100644 --- a/SW.Serverless.Tooling/LocalAdapterHost.cs +++ b/SW.Serverless.Tooling/LocalAdapterHost.cs @@ -3,6 +3,7 @@ using System.Diagnostics; using System.IO; using System.IO.Compression; +using System.Linq; using System.Security.Cryptography; using System.Threading; using System.Threading.Tasks; @@ -40,14 +41,28 @@ public sealed class LocalAdapterHost : IAsyncDisposable this.ownsWork = ownsWork; } + /// As the overload with limits, with none: kept so code built against 10.2.0 still binds. + public static Task StartAsync(string packageDirectory, IDictionary settings, + AdapterRuntimeOptions runtimes, int commandTimeoutSeconds, string workDirectory) => + StartAsync(packageDirectory, settings, runtimes, commandTimeoutSeconds, workDirectory, limits: null); + + /// + /// Installs the package in a temporary store and starts it. With , the + /// adapter runs under those memory and CPU ceilings, sampled every + /// ; a call it is running when it crosses one + /// fails with an naming the limit. Without, it runs as + /// a host runs it by default, unconstrained. + /// public static async Task StartAsync(string packageDirectory, IDictionary settings, - AdapterRuntimeOptions runtimes = null, int commandTimeoutSeconds = 60, string workDirectory = null) + AdapterRuntimeOptions runtimes = null, int commandTimeoutSeconds = 60, string workDirectory = null, + LocalAdapterLimits limits = null) { + if (limits is { IsEmpty: true }) limits = null; var manifest = AdapterManifest.Parse(File.ReadAllText(Path.Combine(packageDirectory, AdapterManifest.FileName))); var work = workDirectory ?? Path.Combine(Path.GetTempPath(), "swsl-local", Guid.NewGuid().ToString("N")); Directory.CreateDirectory(work); - var host = Build(runtimes ?? new AdapterRuntimeOptions(), commandTimeoutSeconds, work); + var host = Build(runtimes ?? new AdapterRuntimeOptions(), commandTimeoutSeconds, work, limits); await host.StartAsync(); try { @@ -59,7 +74,14 @@ await UploadAsync(host.Services.GetRequiredService(), packag ISession session = manifest.IsResident ? await ResidentSession.StartAsync(host.Services, adapterRef, settings, commandTimeoutSeconds) : await ClassicSession.StartAsync(host.Services, adapterRef, settings, commandTimeoutSeconds); - return new LocalAdapterHost(host, session, work, workDirectory == null); + return new LocalAdapterHost(host, session, work, workDirectory == null) + { + limits = limits, + adapterId = adapterRef, + // The process the calls go to, when the resident host runs it: every adapter but + // a .NET one on the classic text protocol, which no limit reaches. + instance = host.Services.GetRequiredService().List().FirstOrDefault(), + }; } catch { @@ -70,11 +92,87 @@ await UploadAsync(host.Services.GetRequiredService(), packag } } + LocalAdapterLimits limits; + string adapterId; + ResidentAdapterInstance instance; + /// Calls a command and returns its answer as text: raw for a string, JSON for anything else. - public Task CallAsync(string command, object input) => session.CallAsync(command, input); + public async Task CallAsync(string command, object input) + { + try + { + return await session.CallAsync(command, input); + } + catch (Exception ex) when (limits != null) + { + var explained = await ExplainAsync(ex); + if (explained == null) throw; + throw explained; + } + } /// Calls a command that answers nothing. - public Task CallVoidAsync(string command, object input) => session.CallVoidAsync(command, input); + public async Task CallVoidAsync(string command, object input) + { + try + { + await session.CallVoidAsync(command, input); + } + catch (Exception ex) when (limits != null) + { + var explained = await ExplainAsync(ex); + if (explained == null) throw; + throw explained; + } + } + + static readonly string[] OutOfMemorySigns = + { + "heap out of memory", "Reached heap limit", "MemoryError", "OutOfMemoryException", "Cannot allocate memory", + }; + + /// + /// A failure that a limit explains, said in those terms; null to let it through as it is. + /// The host's own stops already say which limit, and a runtime's own ceiling raises inside + /// the adapter — but an adapter that died at its runtime's ceiling (Node's heap) only closed + /// its stream, which reads as nothing in particular. + /// + async Task ExplainAsync(Exception ex) + { + if (ex is AdapterStoppedException || limits.MemoryLimitBytes <= 0) return null; + var limit = $"{limits.MemoryLimitBytes / 1024 / 1024} MB"; + + if (ex is AdapterInvocationException invocation) + { + var outOfMemory = (invocation.AdapterExceptionType ?? "").Contains("OutOfMemory") || + (invocation.AdapterExceptionType ?? "").EndsWith("MemoryError"); + if (!outOfMemory || invocation.Message.Contains("memory limit")) return null; + var message = invocation.Message[(invocation.AdapterExceptionType.Length + 2)..]; + return new AdapterInvocationException(invocation.AdapterExceptionType, + $"the adapter ran out of memory: it runs under a memory limit of {limit}" + + (string.IsNullOrWhiteSpace(message) ? "" : $" ({message})"), invocation.Detail); + } + + if (ex is not IOException || instance?.Process == null) return null; + + // Its last words are on stderr, read as they come: wait for the process to be gone. + try + { + using var wait = new CancellationTokenSource(TimeSpan.FromSeconds(3)); + await instance.Process.WaitForExitAsync(wait.Token); + } + catch { /* still running, or already disposed */ } + + if (instance.StopReason != null) + return new AdapterStoppedException(adapterId, instance.StopReason, instance.StoppedForLimit); + var sign = instance.Diagnostics.Select(line => OutOfMemorySigns.FirstOrDefault(s => + line.Contains(s, StringComparison.OrdinalIgnoreCase))) + .FirstOrDefault(s => s != null); + return sign == null + ? null + : new AdapterStoppedException(adapterId, + $"it ran out of memory under its memory limit of {limit} and exited ({sign})", limitExceeded: true); + } public async ValueTask DisposeAsync() { @@ -84,7 +182,10 @@ public async ValueTask DisposeAsync() if (ownsWork) try { Directory.Delete(work, true); } catch { } } - static IHost Build(AdapterRuntimeOptions runtimes, int commandTimeoutSeconds, string work) + /// The default host tolerates 45 seconds of missed heartbeats; sampling faster keeps that. + static readonly TimeSpan HeartbeatTolerance = TimeSpan.FromSeconds(45); + + static IHost Build(AdapterRuntimeOptions runtimes, int commandTimeoutSeconds, string work, LocalAdapterLimits limits) { var tag = Guid.NewGuid().ToString("N")[..10]; return Host.CreateDefaultBuilder() @@ -116,6 +217,21 @@ static IHost Build(AdapterRuntimeOptions runtimes, int commandTimeoutSeconds, st o.SocketPath = Path.Combine("/tmp", $"swsl-local-{tag}.sock"); o.PipeName = $"swsl-local-{tag}"; o.HandshakeTimeout = TimeSpan.FromSeconds(60); + if (limits == null) return; + + // Sampled on the heartbeat, so the heartbeat is the sample interval: at the + // default 15 seconds a short try finishes between two samples. The ping waits + // as long as the interval, so the misses allowed grow with it and an adapter + // busy in one long call is given the 45 seconds it always was. + var interval = limits.SampleInterval > TimeSpan.FromMilliseconds(250) + ? limits.SampleInterval : TimeSpan.FromMilliseconds(250); + o.HeartbeatInterval = interval; + o.MissedHeartbeatsBeforeRestart = Math.Max(3, (int)Math.Ceiling(HeartbeatTolerance / interval)); + o.HardMemoryLimitBytes = Math.Max(0, limits.MemoryLimitBytes); + o.CpuPercentLimit = Math.Max(0, limits.CpuPercentLimit); + o.CpuLimitSamples = Math.Max(1, limits.CpuLimitSamples); + // Asked to stop at the CPU ceiling, an adapter in a runaway loop never will. + o.DrainDeadline = interval * 2 > TimeSpan.FromSeconds(2) ? interval * 2 : TimeSpan.FromSeconds(2); }); }) .Build(); diff --git a/SW.Serverless.Tooling/LocalAdapterLimits.cs b/SW.Serverless.Tooling/LocalAdapterLimits.cs new file mode 100644 index 0000000..bf0034c --- /dev/null +++ b/SW.Serverless.Tooling/LocalAdapterLimits.cs @@ -0,0 +1,41 @@ +using System; + +namespace SW.Serverless.Tooling +{ + /// + /// The ceilings an adapter run by runs under — what a tool that + /// builds and tries adapters on a shared server puts on drafts it has no reason to trust yet. + /// Zero on a ceiling means none; with no limits at all an adapter runs as it always has. + /// + public sealed class LocalAdapterLimits + { + /// + /// The most memory the adapter's process may use, in bytes. It is the host's hard ceiling: the + /// process is killed when a sample finds it over, and the call it was running fails saying so. + /// The runtime enforces it too where it can — .NET's GC heap limit, Node's + /// --max-old-space-size, RLIMIT_DATA for Python on Linux — so an allocation past it usually + /// fails inside the adapter first. 0 for none. + /// + public long MemoryLimitBytes { get; set; } + + /// + /// Sustained CPU ceiling, as a percentage of the WHOLE machine, as the resident host measures + /// it: one core flat out on an eight-core machine is 12.5%. Held for + /// samples in a row, the adapter is asked to stop and, if it does + /// not within a few samples, killed; the call it was running fails saying so. 0 for none. + /// + public double CpuPercentLimit { get; set; } + + /// Samples in a row over before it trips. + public int CpuLimitSamples { get; set; } = 3; + + /// + /// How often the process is sampled. A second by default, so a short try is caught rather + /// than finishing between two of a long-running host's samples. + /// + public TimeSpan SampleInterval { get; set; } = TimeSpan.FromSeconds(1); + + /// True when neither ceiling is set. + public bool IsEmpty => MemoryLimitBytes <= 0 && CpuPercentLimit <= 0; + } +} diff --git a/SW.Serverless.Tooling/Scaffolding/Scaffolder.cs b/SW.Serverless.Tooling/Scaffolding/Scaffolder.cs index a0b3a64..117d5e2 100644 --- a/SW.Serverless.Tooling/Scaffolding/Scaffolder.cs +++ b/SW.Serverless.Tooling/Scaffolding/Scaffolder.cs @@ -48,10 +48,10 @@ public static class Scaffolder public const string SdkPackageVersion = "10.1.0"; /// The Python SDK the templates name; the build vendors the copy it carries. - public const string PythonSdkVersion = "10.2.0"; + public const string PythonSdkVersion = "10.2.2"; /// The Node SDK the templates name; the build vendors the copy it carries. - public const string NodeSdkVersion = "10.2.0"; + public const string NodeSdkVersion = "10.2.2"; public static readonly IReadOnlyList Languages = new[] { "dotnet", "python", "node", "typescript" }; diff --git a/SW.Serverless/Resident/ResidentAdapterHost.cs b/SW.Serverless/Resident/ResidentAdapterHost.cs index f38d196..2c06edd 100644 --- a/SW.Serverless/Resident/ResidentAdapterHost.cs +++ b/SW.Serverless/Resident/ResidentAdapterHost.cs @@ -717,6 +717,8 @@ async Task HeartbeatAsync(Supervised supervised) if (supervised.MissedHeartbeats >= options.MissedHeartbeatsBeforeRestart) { supervised.MissedHeartbeats = 0; + instance.MarkStopping( + $"it missed {options.MissedHeartbeatsBeforeRestart} heartbeats in a row and was restarted", limit: false); TryKill(instance.Process); // Exited handler restarts it } } @@ -772,6 +774,9 @@ void SampleProcess(Supervised supervised, ResidentAdapterInstance instance) logger.LogWarning( "Adapter {AdapterId}/{InstanceKey} held {Cpu}% CPU across {Samples} samples (limit {Limit}%); asking it to drain.", instance.AdapterId, instance.InstanceKey, supervised.CpuPercent, cpuSamples, cpuLimit); + instance.MarkStopping( + $"it held {supervised.CpuPercent}% CPU for {cpuSamples} samples in a row, over its CPU limit of {Math.Round(cpuLimit, 1)}%, and was asked to stop", + limit: true); instance.RequestShutdown("sustained cpu limit", drain: true); return; } @@ -784,18 +789,23 @@ void SampleProcess(Supervised supervised, ResidentAdapterInstance instance) supervised.DrainRequestedOn = DateTimeOffset.UtcNow; logger.LogWarning("Adapter {AdapterId}/{InstanceKey} at {Rss} MB crossed the soft limit; asking it to drain.", instance.AdapterId, instance.InstanceKey, rss / 1024 / 1024); + instance.MarkStopping( + $"it was using {Mb(rss)}, over its soft memory limit of {Mb(soft)}, and was asked to stop", limit: true); instance.RequestShutdown("soft memory limit", drain: true); } else if (hard > 0 && rss > hard) { logger.LogError("Adapter {AdapterId}/{InstanceKey} at {Rss} MB crossed the hard limit; killing.", instance.AdapterId, instance.InstanceKey, rss / 1024 / 1024); + instance.MarkStopping($"it was using {Mb(rss)}, over its memory limit of {Mb(hard)}, and was killed", limit: true); TryKill(process); } } catch { /* sampling must never throw */ } } + static string Mb(long bytes) => $"{Math.Round(bytes / 1024.0 / 1024.0)} MB"; + static int SafeExitCode(Process p) { try { return p?.ExitCode ?? -1; } catch { return -1; } } static void TryKill(Process p) { try { if (p != null && !p.HasExited) p.Kill(entireProcessTree: true); } catch { } } diff --git a/SW.Serverless/Resident/ResidentAdapterInstance.cs b/SW.Serverless/Resident/ResidentAdapterInstance.cs index ae9849d..e30725d 100644 --- a/SW.Serverless/Resident/ResidentAdapterInstance.cs +++ b/SW.Serverless/Resident/ResidentAdapterInstance.cs @@ -129,6 +129,32 @@ internal ResidentAdapterInstance(string adapterId, string instanceKey, string to public IReadOnlyCollection Diagnostics => diagnostics.ToArray(); + /// + /// Why the host stopped this process, when it was the host's doing rather than the adapter's: + /// a memory or CPU ceiling it crossed, or heartbeats it stopped answering. Null otherwise. + /// Calls cut short by that stop fail with an that says so. + /// + public string StopReason { get; private set; } + + /// True when is a resource ceiling. + public bool StoppedForLimit { get; private set; } + + /// Records why the supervisor is about to stop it; the first reason stands. + internal void MarkStopping(string reason, bool limit) + { + if (StopReason != null) return; + StoppedForLimit = limit; + StopReason = reason; + } + + /// What a call that can no longer be answered fails with. + Exception StreamClosed() => StopReason == null + ? new IOException("Adapter stream closed.") + : new AdapterStoppedException(AdapterId, StopReason, StoppedForLimit); + + string NotReady() => $"Adapter {AdapterId} is {State}, not Ready." + + (StopReason == null ? "" : $" It stopped: {StopReason}."); + /// Completes when the adapter has attached and been made Ready. public Task Attached => attached.Task; @@ -211,7 +237,7 @@ internal async Task RunStreamAsync(IAsyncStreamReader input, { State = InstanceState.Stopped; stateRequests.Writer.TryComplete(); - FailAllPending(new IOException("Adapter stream closed.")); + FailAllPending(StreamClosed()); } } @@ -399,7 +425,7 @@ public async Task InvokeAsync(string command, byte[] payload = null, string sessionId = null, IDictionary properties = null) { if (State != InstanceState.Ready) - throw new InvalidOperationException($"Adapter {AdapterId} is {State}, not Ready."); + throw new InvalidOperationException(NotReady()); var id = Interlocked.Increment(ref nextId); var call = new PendingCall @@ -413,7 +439,7 @@ public async Task InvokeAsync(string command, byte[] payload = null, // FailAllPending has already run, so nothing would ever complete this call and it // waited out the whole timeout. if (State == InstanceState.Stopped && pending.TryRemove(id, out _)) - throw new IOException("Adapter stream closed."); + throw StreamClosed(); var timeout = timeoutSeconds > 0 ? TimeSpan.FromSeconds(timeoutSeconds) @@ -520,7 +546,7 @@ public async Task ResetAsync(string sessionId, TimeSpan? timeout = null, CancellationToken cancellationToken = default) { if (State != InstanceState.Ready) - throw new InvalidOperationException($"Adapter {AdapterId} is {State}, not Ready."); + throw new InvalidOperationException(NotReady()); var id = Interlocked.Increment(ref nextId); var call = new PendingCall @@ -614,6 +640,27 @@ sealed class PendingCall } } + /// + /// A call cut short because the host stopped the adapter on purpose: it crossed a memory or CPU + /// ceiling, or stopped answering heartbeats. An , as a closed stream + /// always was, so code that caught that still does. + /// + public class AdapterStoppedException : IOException + { + public AdapterStoppedException(string adapterId, string reason, bool limitExceeded) + : base($"Adapter '{adapterId}' stopped: {reason}.") + { + Reason = reason; + LimitExceeded = limitExceeded; + } + + /// Why, in words a person can act on. + public string Reason { get; } + + /// True when it crossed a memory or CPU ceiling. + public bool LimitExceeded { get; } + } + public class AdapterInvocationException : Exception { public AdapterInvocationException(string type, string message, string detail) diff --git a/SW.Serverless/Runtimes/AdapterRuntimes.cs b/SW.Serverless/Runtimes/AdapterRuntimes.cs index bdbec37..75ccb89 100644 --- a/SW.Serverless/Runtimes/AdapterRuntimes.cs +++ b/SW.Serverless/Runtimes/AdapterRuntimes.cs @@ -193,6 +193,12 @@ public ProcessStartInfo StartInfo(string entryPath, RuntimeLaunch launch) } } + /// + /// The memory ceiling, in bytes, handed to a Python adapter, whose SDK applies it as a + /// resource limit where the platform enforces one (Linux). Absent when there is no ceiling. + /// + public const string MemoryLimitVariable = "SW_SERVERLESS_MEMORY_LIMIT_BYTES"; + sealed class PythonRuntime(string executable) : IAdapterRuntime { public string Name => AdapterManifest.PythonRuntime; @@ -208,8 +214,14 @@ public ProcessStartInfo StartInfo(string entryPath, RuntimeLaunch launch) foreach (var argument in launch.Arguments) psi.ArgumentList.Add(argument); psi.Environment["PYTHONUNBUFFERED"] = "1"; psi.Environment["PYTHONDONTWRITEBYTECODE"] = "1"; - // Python has no heap ceiling of its own; the watchdog and, where the container - // allows them, cgroups are what hold it. + // Python has no heap ceiling of its own. The SDK (10.2.2 and later) reads this one + // and, on Linux, sets RLIMIT_DATA to it, so an allocation past it raises MemoryError + // inside the adapter; the watchdog and, where the container allows them, cgroups + // hold it everywhere else. Never inherited from the host: no limit means none. + if (launch.HardMemoryLimitBytes > 0) + psi.Environment[MemoryLimitVariable] = launch.HardMemoryLimitBytes.ToString(System.Globalization.CultureInfo.InvariantCulture); + else + psi.Environment.Remove(MemoryLimitVariable); return psi; } } diff --git a/docs/cli.md b/docs/cli.md index fa2bc3d..51ded43 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -175,7 +175,7 @@ checks: ``` PASS manifest -PASS describe — python SDK 10.2.0, 2 commands +PASS describe — python SDK 10.2.2, 2 commands PASS settings match the manifest — 2 settings PASS starts — classic session SKIP contracts — it declares no contract, so only what every adapter must do is checked diff --git a/docs/compatibility.md b/docs/compatibility.md index 671d19a..5e2640f 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -20,7 +20,7 @@ file formats are versioned so that it stays that way. |---|---|---| | Host, `SimplyWorks.Serverless` | 10.x | `HostInfo.Version` is the host's version, compared with an adapter's `minHostVersion`. | | .NET SDK, `SimplyWorks.Serverless.Sdk` | 10.x | 10.1.0 added `--describe` (needed by `sw-serverless build`) and reading settings from standard input. | -| Python SDK `sw-serverless`, Node SDK `@simplyworks/sw-serverless` | 10.2.0 | Copied into each package by `sw-serverless build`. | +| Python SDK `sw-serverless`, Node SDK `@simplyworks/sw-serverless` | 10.2.2 | Copied into each package by `sw-serverless build`. | | Protocol | 1 and 2 | 1 is the .NET classic text protocol; 2 is gRPC. Hosts speak 2 to 2. | | Manifest (`adapter.json`) | `manifestVersion` 1 | | | Catalog entry | `catalogVersion` 1 | | diff --git a/docs/contracts.md b/docs/contracts.md index eb60a68..a02e4dc 100644 --- a/docs/contracts.md +++ b/docs/contracts.md @@ -259,7 +259,7 @@ For the orders processor above: ``` PASS manifest -PASS describe — python SDK 10.2.0, 1 commands +PASS describe — python SDK 10.2.2, 1 commands PASS settings match the manifest — 1 settings PASS starts — classic session PASS orders processor: methods — Process diff --git a/docs/extending.md b/docs/extending.md index de6b210..6a85093 100644 --- a/docs/extending.md +++ b/docs/extending.md @@ -262,8 +262,38 @@ var report = await new ConformanceRunner().RunAsync(new ConformanceOptions }); ``` -`ConformanceOptions` also takes `Contracts` (used before registered ones), `WorkDirectory` and -`Runtimes`. +`ConformanceOptions` also takes `Contracts` (used before registered ones), `WorkDirectory`, +`Runtimes` and `Limits`. + +### Limits on what runs locally + +A tool that builds and tries adapters on a shared server can cap them (10.2.2 and later). Pass +`LocalAdapterLimits` to `LocalAdapterHost.StartAsync` (`limits:`) or set +`ConformanceOptions.Limits`; leave it null, the default, for no limits. + +```csharp +await using var adapter = await LocalAdapterHost.StartAsync(packageDirectory, settings, + limits: new LocalAdapterLimits + { + MemoryLimitBytes = 256L * 1024 * 1024, // hard: the process is killed above it + CpuPercentLimit = 100.0 / Environment.ProcessorCount, // one core, as a share of the machine + }); +``` + +| Property | Default | Meaning | +|---|---|---| +| `MemoryLimitBytes` | `0` (none) | The host's hard memory limit. Also given to the runtime (see [Hosting](hosting.md#limits-and-supervision)), so on .NET, Node and Python on Linux an allocation past it usually fails inside the adapter first. | +| `CpuPercentLimit` | `0` (none) | Sustained CPU as a share of the whole machine. Over it for `CpuLimitSamples` samples in a row, the adapter is asked to stop, and killed if it hasn't within two samples (at least 2 seconds). | +| `CpuLimitSamples` | `3` | Samples in a row over `CpuPercentLimit` before it trips. | +| `SampleInterval` | 1 second | How often the process is sampled: the host's heartbeat. The missed heartbeats allowed grow to keep the default host's 45 seconds of tolerance. | + +A call the adapter is running when it crosses a limit fails saying so, never with a bare closed +stream: an `AdapterStoppedException` (`LimitExceeded` is true), such as +`Adapter 'acme.orders/0.0.0' stopped: it was using 167 MB, over its memory limit of 150 MB, and +was killed.`, or the adapter's own out-of-memory error, such as +`MemoryError: the adapter ran out of memory: it runs under a memory limit of 150 MB`. Limits reach +every adapter the resident host runs, which is all of them except a .NET adapter on the protocol 1 +text protocol. ## Publishing: PackagePublisher and AdapterRepository diff --git a/docs/hosting.md b/docs/hosting.md index 61bcfa8..ddc8457 100644 --- a/docs/hosting.md +++ b/docs/hosting.md @@ -420,7 +420,14 @@ Notes: - The CPU figure is a share of the whole machine. On a 16-core machine one busy core is about 6%. - The hard memory limit is also given to the runtime: `DOTNET_GCHeapHardLimit` for .NET and `--max-old-space-size` for Node, so an allocation past it fails inside the adapter. Python has no - such setting; only the host's kill applies. + such setting; the host passes it as `SW_SERVERLESS_MEMORY_LIMIT_BYTES`, and from SDK 10.2.2 the + Python SDK sets it as `RLIMIT_DATA` on Linux, so an allocation past it raises `MemoryError`. + Elsewhere (macOS, Windows), and for older SDKs, only the host's kill applies. Without a hard + limit the variable is not set. +- A call cut short because the host stopped the adapter (a memory or CPU limit, or missed + heartbeats) fails with `AdapterStoppedException`, an `IOException` whose message says which + limit, e.g. `Adapter 'acme.orders/1.0.0' stopped: it was using 167 MB, over its memory limit of + 150 MB, and was killed.` `ResidentAdapterInstance.StopReason` keeps the reason. - On Linux the host sets each adapter's `oom_score_adj` to 500, so under memory pressure the kernel kills an adapter before the host. - SW-Serverless does not set operating-system limits such as cgroups. For a hard guarantee, run the diff --git a/docs/manifest.md b/docs/manifest.md index c60b8c4..7ad883d 100644 --- a/docs/manifest.md +++ b/docs/manifest.md @@ -72,7 +72,7 @@ What the build writes into the package (abridged), given an adapter that declare }, "lifecycle": "classic", "protocol": { "min": 2, "max": 2 }, - "sdkVersion": "10.2.0", + "sdkVersion": "10.2.2", "compatibility": { "minHostVersion": "10.1.0", "applications": { "orders-app": "3.2.0" } }, "properties": [ { "name": "Endpoint", "displayName": "API URL", "description": "Where orders go.", "type": "text", "required": true, "secret": false, "group": "Connection" }, diff --git a/docs/protocol.md b/docs/protocol.md index de324d0..4e6f713 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -271,7 +271,7 @@ with standard input closed, and give up after 30 seconds. { "describeVersion": 1, "sdkLanguage": "python", - "sdkVersion": "10.2.0", + "sdkVersion": "10.2.2", "lifecycle": "classic", "protocol": { "min": 2, "max": 2 }, "settings": [ diff --git a/sdk/node/package.json b/sdk/node/package.json index 614a562..610008b 100644 --- a/sdk/node/package.json +++ b/sdk/node/package.json @@ -1,6 +1,6 @@ { "name": "@simplyworks/sw-serverless", - "version": "10.2.0", + "version": "10.2.2", "description": "Write SW-Serverless adapters in JavaScript or TypeScript: settings, commands, and the host protocol, with no dependencies.", "main": "src/index.js", "types": "src/index.d.ts", diff --git a/sdk/node/src/index.js b/sdk/node/src/index.js index c3d179d..40842c0 100644 --- a/sdk/node/src/index.js +++ b/sdk/node/src/index.js @@ -26,7 +26,7 @@ const readline = require("node:readline"); const { AsyncLocalStorage } = require("node:async_hooks"); const wire = require("./wire"); -const SDK_VERSION = "10.2.0"; +const SDK_VERSION = "10.2.2"; const SDK_LANGUAGE = "node"; const PROTOCOL = 2; const ATTACH = "/sw.serverless.v1.AdapterHost/Attach"; diff --git a/sdk/python/pyproject.toml b/sdk/python/pyproject.toml index e189548..b38cb6a 100644 --- a/sdk/python/pyproject.toml +++ b/sdk/python/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "sw-serverless" -version = "10.2.0" +version = "10.2.2" description = "Write SW-Serverless adapters in Python: settings, commands, and the host protocol, with no dependencies." readme = "README.md" requires-python = ">=3.12" diff --git a/sdk/python/src/sw_serverless/_runner.py b/sdk/python/src/sw_serverless/_runner.py index fad14b7..8dbc268 100644 --- a/sdk/python/src/sw_serverless/_runner.py +++ b/sdk/python/src/sw_serverless/_runner.py @@ -19,12 +19,16 @@ from . import _adapter, _types, _wire from ._http2 import GrpcStream -SDK_VERSION = "10.2.0" +SDK_VERSION = "10.2.2" SDK_LANGUAGE = "python" PROTOCOL = 2 ATTACH = "/sw.serverless.v1.AdapterHost/Attach" DESCRIBE_FLAG = "--describe" DESCRIBE_VERSION = 1 +MEMORY_LIMIT_VARIABLE = "SW_SERVERLESS_MEMORY_LIMIT_BYTES" + +# The memory ceiling this process runs under once apply_memory_limit has set it; None otherwise. +_memory_limit = None # ILogger levels, which the host's log pipeline reads: Trace 0 .. Critical 5. _LEVELS = ((logging.CRITICAL, 5), (logging.ERROR, 4), (logging.WARNING, 3), (logging.INFO, 2), (logging.DEBUG, 1)) @@ -314,6 +318,12 @@ async def _invoke(self, frame_id, invoke, cancelled): await self.send({"id": frame_id, "invoke_result": {"payload": payload}}) except asyncio.CancelledError: await self._fail(frame_id, "OperationCanceledException", "the call was cancelled", "") + except MemoryError as ex: + message = str(ex) + if _memory_limit: + message = (f"the adapter ran out of memory: it runs under a memory limit of " + f"{_memory_limit // (1024 * 1024)} MB" + (f" ({message})" if message else "")) + await self._fail(frame_id, _qualified(type(ex)), message, traceback.format_exc()) except Exception as ex: await self._fail(frame_id, getattr(ex, "type", None) or _qualified(type(ex)), str(ex), getattr(ex, "detail", None) or traceback.format_exc()) @@ -479,6 +489,46 @@ def describe(adapter): } +def apply_memory_limit(environ=None, platform=None, resource_module=None): + """Applies the memory ceiling the host hands a Python adapter in SW_SERVERLESS_MEMORY_LIMIT_BYTES. + + Python has no heap ceiling of its own, so on Linux it becomes RLIMIT_DATA, soft and hard: an + allocation past it raises MemoryError inside the adapter rather than growing until the host's + watchdog or the kernel kills it. RLIMIT_DATA rather than RLIMIT_AS: address space counts what + is only reserved — 64 MB per malloc arena, a stack per thread — so a few worker threads put an + adapter using 30 MB at over 600 MB of it. Elsewhere (macOS, Windows) nothing is set, and the + host's watchdog is the ceiling. With no variable, nothing is ever set. + + Returns the ceiling applied in bytes, or None. + """ + global _memory_limit + environ = os.environ if environ is None else environ + platform = sys.platform if platform is None else platform + raw = (environ.get(MEMORY_LIMIT_VARIABLE) or "").strip() + if not raw: + return None + try: + limit = int(raw) + except ValueError: + print(f"{MEMORY_LIMIT_VARIABLE}={raw!r} is not a number of bytes; no memory limit applied", file=sys.stderr) + return None + if limit <= 0 or not platform.startswith("linux"): + return None + try: + if resource_module is None: + import resource as resource_module + _, hard = resource_module.getrlimit(resource_module.RLIMIT_DATA) + # A process can lower its hard limit but not raise it. + if hard != resource_module.RLIM_INFINITY and 0 <= hard < limit: + limit = hard + resource_module.setrlimit(resource_module.RLIMIT_DATA, (limit, limit)) + except Exception as ex: + print(f"the memory limit of {limit} bytes could not be applied: {ex}", file=sys.stderr) + return None + _memory_limit = limit + return limit + + def _check_kinds(adapter): check = getattr(adapter, "__sw_check__", None) if callable(check): @@ -495,6 +545,7 @@ def run(adapter): sys.stdout.flush() return + apply_memory_limit() instance = adapter() if isinstance(adapter, type) else adapter _check_kinds(instance) diff --git a/sdk/python/tests/test_memory_limit.py b/sdk/python/tests/test_memory_limit.py new file mode 100644 index 0000000..92d60e5 --- /dev/null +++ b/sdk/python/tests/test_memory_limit.py @@ -0,0 +1,79 @@ +import os +import subprocess +import sys +import unittest + +from sw_serverless import _runner + +VARIABLE = _runner.MEMORY_LIMIT_VARIABLE +SRC = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "src") + + +class FakeResource: + RLIMIT_DATA = 2 + RLIM_INFINITY = -1 + + def __init__(self, hard=-1): + self.hard = hard + self.set = [] + + def getrlimit(self, which): + return (self.hard, self.hard) + + def setrlimit(self, which, limits): + self.set.append((which, limits)) + + +class MemoryLimitTests(unittest.TestCase): + def tearDown(self): + _runner._memory_limit = None + + def test_on_linux_the_limit_becomes_rlimit_data_soft_and_hard(self): + resource = FakeResource() + applied = _runner.apply_memory_limit({VARIABLE: str(256 << 20)}, "linux", resource) + self.assertEqual(256 << 20, applied) + self.assertEqual([(FakeResource.RLIMIT_DATA, (256 << 20, 256 << 20))], resource.set) + self.assertEqual(256 << 20, _runner._memory_limit) + + def test_a_lower_hard_limit_already_in_place_is_kept(self): + resource = FakeResource(hard=100 << 20) + self.assertEqual(100 << 20, _runner.apply_memory_limit({VARIABLE: str(256 << 20)}, "linux", resource)) + self.assertEqual([(FakeResource.RLIMIT_DATA, (100 << 20, 100 << 20))], resource.set) + + def test_without_a_limit_nothing_is_set(self): + for environ in ({}, {VARIABLE: ""}, {VARIABLE: "0"}, {VARIABLE: "lots"}): + resource = FakeResource() + self.assertIsNone(_runner.apply_memory_limit(environ, "linux", resource), environ) + self.assertEqual([], resource.set) + self.assertIsNone(_runner._memory_limit) + + def test_off_linux_nothing_is_set(self): + # macOS doesn't enforce RLIMIT_DATA (nor RLIMIT_AS, reliably); the host's watchdog holds it there. + for platform in ("darwin", "win32"): + resource = FakeResource() + self.assertIsNone(_runner.apply_memory_limit({VARIABLE: str(256 << 20)}, platform, resource)) + self.assertEqual([], resource.set) + + @unittest.skipUnless(sys.platform.startswith("linux"), "RLIMIT_DATA is enforced on Linux only") + def test_on_linux_an_allocation_past_the_limit_raises_memory_error(self): + script = ( + "from sw_serverless import _runner\n" + "print(_runner.apply_memory_limit())\n" + "try:\n" + " hog = bytearray(512 << 20)\n" + " print('allocated')\n" + "except MemoryError:\n" + " print('MemoryError')\n" + ) + environ = dict(os.environ, PYTHONPATH=SRC) + limited = subprocess.run([sys.executable, "-c", script], env=dict(environ, **{VARIABLE: str(128 << 20)}), + capture_output=True, text=True, timeout=60) + self.assertEqual(["134217728", "MemoryError"], limited.stdout.split(), limited.stderr) + + environ.pop(VARIABLE, None) + unlimited = subprocess.run([sys.executable, "-c", script], env=environ, capture_output=True, text=True, timeout=60) + self.assertEqual(["None", "allocated"], unlimited.stdout.split(), unlimited.stderr) + + +if __name__ == "__main__": + unittest.main()