From 35dcfe6e4d11f7c10fa62acb6ac3f5ec548b6e4b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= <6334612+rbuergi@users.noreply.github.com> Date: Tue, 22 Sep 2026 22:39:03 +0200 Subject: [PATCH 1/2] fix(compiler): bound compile CPU on a dedicated-thread IIoPool lane MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to #5327, which moved the NodeType emit off the ThreadPool onto CompileThread: off the pool, but one new thread per DISTINCT NodeType compiling at once — hundreds on a roll. And the language service's diagnostics binds still ran as Compile-pool leaves on ThreadPool workers (the script workspace with Roslyn's default ConcurrentBuild on top). - IoPool gains dedicatedThreads: InvokeBlocking leaves drain on threads the pool's LimitedConcurrencyLevelTaskScheduler starts itself (mw-cpu-lane), same queue and cap, never a ThreadPool worker. - IoPoolNames.CompileCpu (cap IoPoolOptions.CompileCpu = processor count), built by the registry with dedicatedThreads. Separate from Compile, which also runs kernel scripts that may wait on a compile — sharing its cap is the nested-gate deadlock. A lane leaf only computes, so it cannot nest. - EmitCompiled and the failure-path DiagnoseInputs run on the lane. - Every language-service diagnostics bind (workspace, speculative, script) splits into an IO half on the Compile pool and a bind on the lane; SpeculativeCompilation exposes CreateCompilationAsync + Diagnose. The script workspace binds with ConcurrentBuild off. - Tests with negative controls that fail on the unfixed routing: CompileCpuLaneIsBoundedAndOffThePoolTest, LanguageServiceDiagnosticsRunOn TheCpuLaneTest, and ServiceCompileStaysOffTheThreadPoolTest (now asserts the lane thread). - Docs: CompileOffTheThreadPool (the lane), ControlledIoPooling (pool row). Co-Authored-By: Claude Opus 5.5 (1M context) --- .../MeshNodeCompilationService.cs | 42 ++++--- .../MeshNodeLanguageService.cs | 65 +++++++--- .../SpeculativeCompilation.cs | 23 +++- .../Architecture/CompileOffTheThreadPool.md | 56 +++++++-- .../Data/Architecture/ControlledIoPooling.md | 1 + .../Threading/IoPool.cs | 19 ++- .../Threading/IoPoolOptions.cs | 23 ++++ .../Threading/IoPoolRegistry.cs | 4 +- .../LimitedConcurrencyLevelTaskScheduler.cs | 38 +++++- ...geServiceDiagnosticsRunOnTheCpuLaneTest.cs | 117 ++++++++++++++++++ ...ServiceCompileStaysOffTheThreadPoolTest.cs | 9 +- ...ompileCpuLaneIsBoundedAndOffThePoolTest.cs | 100 +++++++++++++++ 12 files changed, 450 insertions(+), 47 deletions(-) create mode 100644 test/MeshWeaver.Compiler.Pipeline.Test/LanguageServiceDiagnosticsRunOnTheCpuLaneTest.cs create mode 100644 test/MeshWeaver.Hosting.Test/CompileCpuLaneIsBoundedAndOffThePoolTest.cs diff --git a/src/MeshWeaver.Compiler.Pipeline/MeshNodeCompilationService.cs b/src/MeshWeaver.Compiler.Pipeline/MeshNodeCompilationService.cs index 9d56ae82dd..ea761eba75 100644 --- a/src/MeshWeaver.Compiler.Pipeline/MeshNodeCompilationService.cs +++ b/src/MeshWeaver.Compiler.Pipeline/MeshNodeCompilationService.cs @@ -52,6 +52,12 @@ internal class MeshNodeCompilationService( hub.ServiceProvider.GetService()?.Get(IoPoolNames.Compile) ?? IoPool.Unbounded; + // The CPU lane (IoPoolNames.CompileCpu): the Roslyn half of every compile, on a bounded set of + // DEDICATED threads. Unbounded fallback only when no registry is wired (DI-less tests). + private readonly IIoPool _cpuLane = + hub.ServiceProvider.GetService()?.Get(IoPoolNames.CompileCpu) + ?? IoPool.Unbounded; + // 🚨 Offload the compile to the ThreadPool via Task.Run — NOT the IoPool. Two reasons the // IoPool ("the correct abstraction") fails for this leaf: // 1. Its SemaphoreSlim gate serialises the compile against itself (activity-driven + @@ -1224,7 +1230,8 @@ private IObservable ValidateCellSurfaceSingleHome( .Take(1) .SelectMany(inputs => inputs is null ? Observable.Return>(Array.Empty()) - : OnThreadPool(() => CompileDiagnostics.DiagnoseInputs(inputs))) + // Pure Roslyn compute -> the bounded CPU lane, off the ThreadPool (CompileOffTheThreadPool). + : _cpuLane.InvokeBlocking(_ => CompileDiagnostics.DiagnoseInputs(inputs))) // 🚨 BOUNDED with the same clock as the emit leg — this runs on the FAILURE path, // where a hang is worst: the compile has already failed and this is what stands // between that failure and its terminal Error write. Unbounded, a stalled @@ -2048,24 +2055,31 @@ private async Task CompileAsyncCore( logger.LogInformation("Compiling assembly for {NodeName} ({Mode}, {NuGetRefs} NuGet refs)", nodeName, cacheService.IsDiskCacheEnabled ? "disk" : "in-memory", nugetRefs.Length); - // 🚨 THE CPU-BOUND HALF RUNS ON A DEDICATED THREAD — never on the ThreadPool this async - // method happens to be on. Everything above is IO (NuGet restore, the debug source write) and - // may resume on any pool thread; everything in EmitCompiled is synchronous Roslyn work - // (parse, source generators, bind, emit) that holds its thread for the whole compile. This - // leaf is reached through OnThreadPool(Func>) — Task.Run — and the debug source write - // above is ON by default (EnableSourceDebugging), so before this hop EVERY production emit ran - // on a ThreadPool worker even though CompileThread's own doc said the compile leaf did not: - // the sync overload that uses CompileThread is not the one `() => CompileAsync(...)` binds to. - // With ConcurrentBuild also off (EmitPipeline.CreateRunCompilationOptions) a compile now costs - // one dedicated thread and zero pool workers. Doc/Architecture/CompileOffTheThreadPool. - return await CompileThread.Run(() => EmitCompiled( - node, nodeName, source, references, nugetAssemblyPaths, sourcePath, ct)) + // 🚨 THE CPU-BOUND HALF RUNS ON THE CPU LANE — a dedicated thread, BOUNDED. Everything above + // is IO (NuGet restore, the debug source write) and may resume on any pool thread; everything + // in EmitCompiled is synchronous Roslyn work (generators, bind, emit) that holds its thread + // for the whole compile. Before #5327 it ran on whatever ThreadPool worker the debug write's + // await resumed on; #5327 moved it to CompileThread, which is off the pool but UNBOUNDED — one + // new thread per distinct NodeType compiling at once, hundreds on a roll. The CompileCpu lane + // keeps it off the pool AND caps it at IoPoolOptions.CompileCpu (processor count); the rest + // queue. EmitCompiled does nothing but compute and write its own artifact — no pool call, no + // mesh read — so it can never wait on another lane leaf (the nested-gate deadlock the Compile + // pool hit). Doc/Architecture/CompileOffTheThreadPool. + return await _cpuLane + .InvokeBlocking(laneCt => + { + // The lane's token (pool drain) AND the compile's own bound (BoundLeg) both stop Roslyn. + using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, laneCt); + return EmitCompiled( + node, nodeName, source, references, nugetAssemblyPaths, sourcePath, linked.Token); + }) + .Await(ct) .ConfigureAwait(false); } /// /// The synchronous, CPU-bound half of : parse, source generators, - /// emit and the disk/in-memory publish. Called ONLY on a thread. + /// emit and the disk/in-memory publish. Called ONLY on a CPU-lane thread (). /// private CompileEmit EmitCompiled( MeshNode node, diff --git a/src/MeshWeaver.Compiler.Pipeline/MeshNodeLanguageService.cs b/src/MeshWeaver.Compiler.Pipeline/MeshNodeLanguageService.cs index 5e3672c303..b73ac9a7fd 100644 --- a/src/MeshWeaver.Compiler.Pipeline/MeshNodeLanguageService.cs +++ b/src/MeshWeaver.Compiler.Pipeline/MeshNodeLanguageService.cs @@ -62,7 +62,7 @@ internal sealed class MeshNodeLanguageService : IMeshLanguageService // inline, inside this leaf's Compile-pool gate, with no cancellation participation at all, // so a silo/mesh teardown racing the first caller could not reclaim that gate permit within // the drain budget. Every caller now races its OWN `ct` against this shared Task via - // Task.WaitAsync (see GetScriptCompletionsAsync/GetScriptDiagnosticsAsync) — the FIRST + // Task.WaitAsync (see GetScriptCompletionsAsync/ScriptCompilationAsync) — the FIRST // caller's own token never cancels the build (it is shared by every request against this // mesh instance), it only governs how long THAT caller waits. private readonly Lazy> _scriptWorkspace; @@ -86,6 +86,11 @@ internal sealed class MeshNodeLanguageService : IMeshLanguageService // with ConfigureAwait(false) behind a gate. Falls back to the unbounded pool // when no registry is wired (e.g. tests constructing the service outside DI). private readonly IIoPool _ioPool; + private readonly IIoPool _cpuLane; + + /// Test seam (InternalsVisibleTo): runs on the thread that binds a diagnostics compilation. + /// Null in production. + internal Action? OnDiagnoseStarting { get; set; } public MeshNodeLanguageService( MeshNodeCompilationService compilationService, @@ -104,6 +109,11 @@ public MeshNodeLanguageService( memoryStore = new CompletionMemoryStore(hub, logger); _ioPool = hub.ServiceProvider.GetService()?.Get(IoPoolNames.Compile) ?? IoPool.Unbounded; + // The CPU lane: every GetDiagnostics (Roslyn's bind — the expensive part of a squiggle + // request) runs on a bounded set of DEDICATED threads, never a ThreadPool worker + // (Doc/Architecture/CompileOffTheThreadPool). The async workspace legs stay on _ioPool. + _cpuLane = hub.ServiceProvider.GetService()?.Get(IoPoolNames.CompileCpu) + ?? IoPool.Unbounded; // 🚨 The two helpers own subscriptions that outlive the request that started them — the // usage index's mesh-wide corpus scan (its own 30 s Timeout, a 1 s Throttle) and the memory @@ -140,7 +150,8 @@ public IObservable GetDiagnostics(string nodeTypePath) // no diagnostics, which is outside #1592's failure (the mandated sweep // enumerates nodeType:NodeType) but is not what this branch means. ? Observable.Return(NodeDiagnosticsOutcome.NotCompilable) - : _ioPool.Run(ct => GetDiagnosticsAsync(cached, ct)) + : _ioPool.Run(ct => WorkspaceCompilationAsync(cached, ct)) + .SelectMany(compilation => OnCpuLane(compilation, DiagnoseWorkspace)) .Select(NodeDiagnosticsOutcome.Compiled)), // Genuinely not there, or its delete is in flight — either way nothing was checked. @@ -280,13 +291,15 @@ private IObservable SpeculativeFor( // this says "nothing was compiled" rather than inventing a clean bill. ? Observable.Return(NodeDiagnosticsOutcome.NotCompilable) : _ioPool.Run(ct => - speculativeCompilation.GetDiagnosticsAsync(inputs, sourcePath, proposedCode, ct)) + speculativeCompilation.CreateCompilationAsync(inputs, sourcePath, proposedCode, ct)) + .SelectMany(compilation => OnCpuLane(compilation, SpeculativeCompilation.Diagnose)) .Select(NodeDiagnosticsOutcome.Compiled)), // A non-NodeType owner that EXISTS → a standalone script Code node: diagnose in // the script environment so lesson cells get live squiggles too (this used to // fall through to a compilation that parses a cell as REGULAR C#, reporting the // spurious "top-level statements must be in an executable" on every cell). - _ => _ioPool.Run(ct => GetScriptDiagnosticsAsync(proposedCode, ct)) + _ => _ioPool.Run(ct => ScriptCompilationAsync(proposedCode, ct)) + .SelectMany(compilation => OnCpuLane(compilation, DiagnoseScript)) .Select(NodeDiagnosticsOutcome.Compiled), }; @@ -536,14 +549,32 @@ private static bool VersionsEqual( return true; } - private static async Task> GetDiagnosticsAsync( - CachedWorkspace cached, CancellationToken ct) + /// + /// Hands to the CPU lane for its bind. A null compilation (no + /// project / document) answers empty without taking a lane slot. + /// + private IObservable> OnCpuLane( + TCompilation? compilation, + Func> diagnose) + where TCompilation : class + => compilation is null + ? Observable.Return>(Array.Empty()) + : _cpuLane.InvokeBlocking(ct => + { + OnDiagnoseStarting?.Invoke(); + return diagnose(compilation, ct); + }); + + /// The IO half of a node workspace's diagnostics: the project's compilation (not yet bound). + private static async Task WorkspaceCompilationAsync(CachedWorkspace cached, CancellationToken ct) { var project = cached.Workspace.CurrentSolution.GetProject(cached.ProjectId); - if (project is null) return Array.Empty(); + return project is null ? null : await project.GetCompilationAsync(ct).ConfigureAwait(false); + } - var compilation = await project.GetCompilationAsync(ct).ConfigureAwait(false); - if (compilation is null) return Array.Empty(); + /// The CPU half: bind and report — a CPU-lane leaf (no IO, no pool call). + private static IReadOnlyList DiagnoseWorkspace(Compilation compilation, CancellationToken ct) + { var diags = compilation.GetDiagnostics(ct); if (diags.IsDefaultOrEmpty) return Array.Empty(); @@ -645,19 +676,20 @@ private async Task> GetScriptCompletionsAsync( /// Diagnostics for a standalone SCRIPT Code node — the script-environment sibling of /// , so lesson cells get live squiggles. /// - private async Task> GetScriptDiagnosticsAsync( - string proposedCode, CancellationToken ct) + private async Task ScriptCompilationAsync(string proposedCode, CancellationToken ct) { var script = await _scriptWorkspace.Value.WaitAsync(ct).ConfigureAwait(false); var document = script.Workspace.CurrentSolution .WithDocumentText(script.DocumentId, SourceText.From(StripKernelDirectives(proposedCode))) .GetDocument(script.DocumentId); - var compilation = document is null + return document is null ? null : await document.Project.GetCompilationAsync(ct).ConfigureAwait(false); - if (compilation is null) return Array.Empty(); - + } + /// The CPU half of a script's diagnostics — a CPU-lane leaf. + private static IReadOnlyList DiagnoseScript(Compilation compilation, CancellationToken ct) + { var diags = compilation.GetDiagnostics(ct); if (diags.IsDefaultOrEmpty) return Array.Empty(); var result = new List(); @@ -968,7 +1000,10 @@ private static async Task BuildScriptWorkspaceAsync(IServicePro language: LanguageNames.CSharp, compilationOptions: new CSharpCompilationOptions( OutputKind.DynamicallyLinkedLibrary, - metadataReferenceResolver: Kernel.Hub.MeshScriptEnvironment.MetadataResolver) + metadataReferenceResolver: Kernel.Hub.MeshScriptEnvironment.MetadataResolver, + // Roslyn's default fans every bind out onto the ThreadPool — see + // EmitPipeline.CreateRunCompilationOptions / CompileOffTheThreadPool. + concurrentBuild: false) .WithUsings(Kernel.Hub.MeshScriptEnvironment.Imports), parseOptions: new CSharpParseOptions(kind: SourceCodeKind.Script), metadataReferences: references, diff --git a/src/MeshWeaver.Compiler/SpeculativeCompilation.cs b/src/MeshWeaver.Compiler/SpeculativeCompilation.cs index d4dfaf1201..d02256e15e 100644 --- a/src/MeshWeaver.Compiler/SpeculativeCompilation.cs +++ b/src/MeshWeaver.Compiler/SpeculativeCompilation.cs @@ -41,6 +41,19 @@ public async Task> GetDiagnosticsAsync( string sourcePath, string proposedCode, CancellationToken ct) + => Diagnose(await CreateCompilationAsync(inputs, sourcePath, proposedCode, ct).ConfigureAwait(false), ct); + + /// + /// The IO half of : resolves any proposed NuGet references and + /// builds the speculative compilation (parse only — nothing is bound yet). Split from + /// so a caller can run the CPU-bound half on the dedicated CPU lane + /// (IoPoolNames.CompileCpu, Doc/Architecture/CompileOffTheThreadPool). + /// + public async Task CreateCompilationAsync( + CompilationInputs inputs, + string sourcePath, + string proposedCode, + CancellationToken ct) { // Strip #r from the proposed source so Roslyn doesn't reject it with CS7011. // Resolve any new package refs and append them to the cached reference set — @@ -106,12 +119,20 @@ public async Task> GetDiagnosticsAsync( SourceText.From(code), inputs.ParseOptions, path: path)); } - var compilation = CSharpCompilation.Create( + return CSharpCompilation.Create( inputs.AssemblyName, syntaxTrees: trees, references: effectiveReferences, options: inputs.CompilationOptions); + } + /// + /// The CPU-bound half of : binds + /// and reports its user-actionable diagnostics (framework-generated trees filtered out). + /// Synchronous and pure — no IO, no pool call — so it is a valid CPU-lane leaf. + /// + public static IReadOnlyList Diagnose(CSharpCompilation compilation, CancellationToken ct) + { var diags = compilation.GetDiagnostics(ct); if (diags.IsDefaultOrEmpty) return Array.Empty(); diff --git a/src/MeshWeaver.Documentation/Data/Architecture/CompileOffTheThreadPool.md b/src/MeshWeaver.Documentation/Data/Architecture/CompileOffTheThreadPool.md index 2ddfe34870..1f76fc6645 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/CompileOffTheThreadPool.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/CompileOffTheThreadPool.md @@ -61,10 +61,47 @@ silo's grain-turn threads to finish it. - The canonical factory is **unchanged**, and the content key (`EmitPipeline.OptionsFingerprint`) still renders it — `ConcurrentBuild` is a scheduling choice that does not change the emitted bytes, and changing the fingerprinted factory would re-key every NodeType in the fleet. -- `CompileAsyncCore` keeps its IO prefix and hands the synchronous half (`EmitCompiled`) to - `CompileThread.Run`. - -`RoslynEmitStaysOffTheThreadPoolTest` pins it deterministically: an `AsyncLocal` change handler counts +- `CompileAsyncCore` keeps its IO prefix and hands the synchronous half (`EmitCompiled`) to the + **CPU lane** (below). The first cut of this fix handed it to `CompileThread.Run` — off the pool, but + one new thread per distinct NodeType compiling at once, which a roll turns into hundreds. + +### The CPU lane (`IoPoolNames.CompileCpu`) + +Every purely CPU-bound Roslyn leaf goes through ONE bounded lane: an `IIoPool` whose +`InvokeBlocking` leaves run on at most `IoPoolOptions.CompileCpu` (default: processor count) threads +the lane starts itself — named `mw-cpu-lane`, never ThreadPool workers. The bound is the pool's +ordinary blocking scheduler (`LimitedConcurrencyLevelTaskScheduler`) with `dedicatedThreads: true`: +the same queue and the same cap, only the drain loops run on their own threads. A burst of hundreds of +distinct-type compiles now QUEUES behind the cap instead of becoming hundreds of threads, and the OS +time-slices the lane's threads fairly against the ThreadPool the grain turns run on. + +Its leaves: + +- a NodeType's `EmitCompiled` (generators, bind, emit, artifact write) and the failure-path + `CompileDiagnostics.DiagnoseInputs`; +- every language-service diagnostics bind — the node workspace (`GetDiagnostics`), a speculative + proposal (`SpeculativeCompilation.Diagnose`) and a script cell. The IO half of each (NuGet refs, + `Project.GetCompilationAsync`) stays on the `Compile` pool; only the bind moves. The script + workspace now also binds with `ConcurrentBuild` off. + +🚨 **The lane is separate from the `Compile` pool on purpose.** `Compile` also runs kernel SCRIPTS — user +code that may activate a NodeType and wait for its compile — and cell-surface assembly loads. An emit +queued behind the same cap is the nested-gate deadlock the compile leaf already hit once (every slot +held by a script waiting for a compile that needs a slot). A lane leaf does nothing but compute: no +pool call, no mesh read, no wait on another leaf, so it cannot nest by construction. Keep it that way — +a leaf that needs IO splits it off, as the three diagnostics arms do. + +Tests, each with a negative control that fails on the unfixed code: + +- `CompileCpuLaneIsBoundedAndOffThePoolTest` (Hosting.Test) — six parked leaves on a cap-2 lane: at + most 2 run at once, all 6 run, none on a pool worker; the control, the ordinary `Compile` pool with + the same cap, runs all 6 on pool workers. With the registry building the lane without dedicated + threads the first test fails. +- `ServiceCompileStaysOffTheThreadPoolTest` — a real compile through the service reaches Roslyn on an + `mw-cpu-lane` thread with `ConcurrentBuild` off. Fails when the emit is routed through the `Compile` pool. +- `LanguageServiceDiagnosticsRunOnTheCpuLaneTest` — all three diagnostics arms bind on `mw-cpu-lane` + threads. Fails when the binds are routed through the `Compile` pool. +- `RoslynEmitStaysOffTheThreadPoolTest` pins the emit itself deterministically: an `AsyncLocal` change handler counts every ThreadPool worker that starts running under the emit's `ExecutionContext` (Roslyn's fan-out tasks flow it). With the run options the count is 0; the negative control — the same emit with `ConcurrentBuild` on — counts more than 0. A third case pins that the content key did not move. @@ -90,12 +127,11 @@ roll" is the expected state of most samples, not a rare one. What was NOT established: the portal's own ThreadPool metrics for those windows (the control instance was unavailable, so no `Logs`/Prometheus read was taken), and therefore how much of each window's starvation the compiles account for. Other CPU on a rolling survivor — re-activating the grains of the -pods being replaced — competes for the same cores and is not addressed here. The Compile `IIoPool`'s -blocking leaves (the language service) still run on pool workers by that pool's design. And nothing -bounds how many DISTINCT NodeTypes compile at once — the service single-flights per type only — so a -burst of hundreds now costs hundreds of dedicated threads (it cost as many pool-injected threads -before); a bound needs an `IIoPool` lane over dedicated threads, since the Compile pool's gate -deadlocked the compile against itself and its leaves run on the pool. +pods being replaced — competes for the same cores and is not addressed here. Still on the ThreadPool +by design: the language service's hover and completion requests (Roslyn's async `QuickInfoService` / +`CompletionService`, which schedule their own work with `ConfigureAwait(false)` and cannot be pinned +to a thread), the IO halves of the diagnostics requests, and `CompileResultFromAssembly` (assembly load +and user type initializers — bounded by its own timeout, on `CompileThread`, not a CPU compile). ## Related diff --git a/src/MeshWeaver.Documentation/Data/Architecture/ControlledIoPooling.md b/src/MeshWeaver.Documentation/Data/Architecture/ControlledIoPooling.md index 0b9c745f7a..84a66258ca 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/ControlledIoPooling.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/ControlledIoPooling.md @@ -209,6 +209,7 @@ one, which is often *not* "how much parallelism can this resource take". | `Layout` | 256 | Drain hook, as `Query` (a page renders many nested areas at once) | | `Routing` | 256 | Isolation boundary — see below | | `Compile` | `Environment.ProcessorCount` | CPU-bound | +| `CompileCpu` | `Environment.ProcessorCount` | The CPU LANE: its `InvokeBlocking` leaves run on DEDICATED threads, never ThreadPool workers — Roslyn binds and emits only, and a leaf may not call a pool or read the mesh. Separate from `Compile` so an emit can never queue behind a kernel script waiting for it. See [Compiling Off the ThreadPool](../CompileOffTheThreadPool) | | `Process` | 4 | Heavy external processes | | `pg:{provider}` / `sf:{provider}` (writes) | **1** | Half a connection BUDGET, not a mirror of one connection — see the pairing note below | | `pg-read:{provider}` / `sf-read:{provider}` | 16 | The other half: keeps read fan-out below the shared connection pool's `MaxPoolSize` so reads can't starve writes | diff --git a/src/MeshWeaver.Mesh.Contract/Threading/IoPool.cs b/src/MeshWeaver.Mesh.Contract/Threading/IoPool.cs index ea283a50a0..5bb7d5b281 100644 --- a/src/MeshWeaver.Mesh.Contract/Threading/IoPool.cs +++ b/src/MeshWeaver.Mesh.Contract/Threading/IoPool.cs @@ -161,6 +161,23 @@ public IoPool(int maxConcurrency, TimeSpan drainTimeout) : this(maxConcurrency, /// overload exists so a test whose leaf can only end by cancellation need not spend the grace. /// public IoPool(int maxConcurrency, TimeSpan drainTimeout, TimeSpan drainGrace) + : this(maxConcurrency, drainTimeout, drainGrace, dedicatedThreads: false) { } + + /// + /// As , choosing where + /// leaves run. + /// + /// Maximum number of operations allowed to run concurrently; must be at least 1. + /// How long waits at each of its joins AFTER cancelling. + /// How long waits for in-flight work BEFORE cancelling. + /// + /// true: leaves run on at most + /// threads this pool starts itself, never on ThreadPool workers — for CPU-bound leaves, whose + /// whole cost is the time they HOLD a thread (). Only the + /// blocking entry point changes; the async entry points still start on the ThreadPool, so a + /// dedicated-thread pool is meant for alone. + /// + public IoPool(int maxConcurrency, TimeSpan drainTimeout, TimeSpan drainGrace, bool dedicatedThreads) { if (maxConcurrency < 1) throw new ArgumentOutOfRangeException(nameof(maxConcurrency)); @@ -176,7 +193,7 @@ public IoPool(int maxConcurrency, TimeSpan drainTimeout, TimeSpan drainGrace) CancellationToken.None, TaskCreationOptions.DenyChildAttach, TaskContinuationOptions.None, - new LimitedConcurrencyLevelTaskScheduler(maxConcurrency)); + new LimitedConcurrencyLevelTaskScheduler(maxConcurrency, dedicatedThreads)); // 🚨 STARTED HERE, NOT AT TEARDOWN — see StartCanceller. Every field the thread touches is // assigned above; it parks on _cancelRequestedLatch and does nothing until Drain()/Dispose() // raises it. diff --git a/src/MeshWeaver.Mesh.Contract/Threading/IoPoolOptions.cs b/src/MeshWeaver.Mesh.Contract/Threading/IoPoolOptions.cs index 949220a28e..179fc29c27 100644 --- a/src/MeshWeaver.Mesh.Contract/Threading/IoPoolOptions.cs +++ b/src/MeshWeaver.Mesh.Contract/Threading/IoPoolOptions.cs @@ -83,6 +83,19 @@ public static class IoPoolNames /// CPU-bound compilation (Roslyn compile/script). Wave 3. public const string Compile = "Compile"; + /// + /// The CPU lane: purely CPU-bound Roslyn leaves — a NodeType's generators + bind + emit, and the + /// language service's GetDiagnostics — run through + /// on at most DEDICATED threads, never ThreadPool workers. + /// 🚨 Separate from on purpose. That pool also runs kernel SCRIPTS (user + /// code that may activate a NodeType and wait for its compile) and assembly loads; putting the + /// emit behind the same gate is the nested-gate deadlock the compile leaf already hit once. A + /// leaf on this lane must do NOTHING but compute: no pool call, no mesh read, no wait on another + /// leaf — which makes nesting impossible by construction. See + /// Doc/Architecture/CompileOffTheThreadPool. + /// + public const string CompileCpu = "CompileCpu"; + /// External process execution (Process.Start). Wave 3. public const string Process = "Process"; @@ -274,6 +287,15 @@ public sealed record IoPoolOptions /// Concurrent compilations. CPU-bound; defaults to the processor count. public int Compile { get; init; } = Environment.ProcessorCount; + /// + /// Concurrent CPU-lane leaves (): Roslyn emits and + /// diagnostics, each on a dedicated thread. Defaults to the processor count — the most such + /// leaves that can make progress at once — so a burst of hundreds of distinct NodeType compiles + /// queues here instead of becoming hundreds of threads, while the OS time-slices the lane + /// fairly against the ThreadPool the grain turns run on. + /// + public int CompileCpu { get; init; } = Environment.ProcessorCount; + /// Concurrent external processes. Heavy; defaults to 4. public int Process { get; init; } = 4; @@ -335,6 +357,7 @@ public int MaxConcurrencyFor(string name) => IoPoolNames.AgentStore => AgentStore, IoPoolNames.Routing => Routing, IoPoolNames.Compile => Compile, + IoPoolNames.CompileCpu => CompileCpu, IoPoolNames.Process => Process, _ => Default, }; diff --git a/src/MeshWeaver.Mesh.Contract/Threading/IoPoolRegistry.cs b/src/MeshWeaver.Mesh.Contract/Threading/IoPoolRegistry.cs index 8eedbb5936..a17d3da205 100644 --- a/src/MeshWeaver.Mesh.Contract/Threading/IoPoolRegistry.cs +++ b/src/MeshWeaver.Mesh.Contract/Threading/IoPoolRegistry.cs @@ -84,7 +84,9 @@ public IIoPool Get(string name) // and anything that did not win the race is disposed, which wakes its canceller and lets it // exit. Same reason the refusal path below disposes `raced`. IoPool? candidate = null; - var pool = _pools.GetOrAdd(name, n => candidate = new IoPool(_options.MaxConcurrencyFor(n), _options.DrainTimeout, _options.DrainGrace)); + var pool = _pools.GetOrAdd(name, n => candidate = new IoPool(_options.MaxConcurrencyFor(n), _options.DrainTimeout, _options.DrainGrace, + // The CPU lane holds its threads for pure computation — never borrow them from the pool. + dedicatedThreads: n == IoPoolNames.CompileCpu)); if (candidate is not null && !ReferenceEquals(pool, candidate)) candidate.Dispose(); diff --git a/src/MeshWeaver.Mesh.Contract/Threading/LimitedConcurrencyLevelTaskScheduler.cs b/src/MeshWeaver.Mesh.Contract/Threading/LimitedConcurrencyLevelTaskScheduler.cs index 1d4d45f958..61f4aec693 100644 --- a/src/MeshWeaver.Mesh.Contract/Threading/LimitedConcurrencyLevelTaskScheduler.cs +++ b/src/MeshWeaver.Mesh.Contract/Threading/LimitedConcurrencyLevelTaskScheduler.cs @@ -19,6 +19,14 @@ namespace MeshWeaver.Mesh.Threading; /// on. This is the "compatible with how Orleans wants us to pool" property: /// reuse the framework's pool, just govern it. /// +/// 🚨 Except for CPU-bound work, where borrowing is the defect. A capped borrower still +/// HOLDS up to maxDegreeOfParallelism pool workers for as long as each leaf computes, and the +/// pool's minimum is ProcessorCount — so a CPU pool capped at the processor count occupies +/// every worker the grain turns need. With dedicatedThreads the same cap is enforced over +/// threads this scheduler starts itself (one per concurrent drain loop, exiting when the queue is +/// empty): the bound is identical, and the ThreadPool is not touched at all +/// (IoPoolNames.CompileCpu, Doc/Architecture/CompileOffTheThreadPool). +/// /// The _tasks queue is an INSTANCE field guarded by lock(_tasks) /// — not static, so it dies with the owning and never bleeds /// across meshes/tests. A mutable is required because @@ -36,11 +44,18 @@ internal sealed class LimitedConcurrencyLevelTaskScheduler : TaskScheduler // Number of ThreadPool work items currently dispatched (running or queued to run). private int _delegatesQueuedOrRunning; - public LimitedConcurrencyLevelTaskScheduler(int maxDegreeOfParallelism) + // True: each drain loop runs on a thread started here, never a ThreadPool worker. + private readonly bool _dedicatedThreads; + + /// Name of the threads a dedicated-thread scheduler starts — visible in dumps. + internal const string DedicatedThreadName = "mw-cpu-lane"; + + public LimitedConcurrencyLevelTaskScheduler(int maxDegreeOfParallelism, bool dedicatedThreads = false) { if (maxDegreeOfParallelism < 1) throw new ArgumentOutOfRangeException(nameof(maxDegreeOfParallelism)); _maxDegreeOfParallelism = maxDegreeOfParallelism; + _dedicatedThreads = dedicatedThreads; } protected override void QueueTask(Task task) @@ -60,7 +75,24 @@ protected override void QueueTask(Task task) private void NotifyThreadPoolOfPendingWork() { - ThreadPool.UnsafeQueueUserWorkItem(_ => + if (_dedicatedThreads) + { + // A drain loop of its own: the same loop, on a thread the pool never lends out. Background + // so an idle-but-draining loop can never hold the process open; UnsafeStart because each + // task carries its own captured ExecutionContext into TryExecuteTask. + new Thread(_ => DrainQueue(), maxStackSize: 0) + { + IsBackground = true, + Name = DedicatedThreadName, + }.UnsafeStart(); + return; + } + + ThreadPool.UnsafeQueueUserWorkItem(_ => DrainQueue(), null); + } + + private void DrainQueue() + { { // Process tasks until the queue drains, then relinquish this slot. while (true) @@ -80,7 +112,7 @@ private void NotifyThreadPoolOfPendingWork() TryExecuteTask(item); } - }, null); + } } // Never inline. Inlining would run the (blocking) task on whatever thread diff --git a/test/MeshWeaver.Compiler.Pipeline.Test/LanguageServiceDiagnosticsRunOnTheCpuLaneTest.cs b/test/MeshWeaver.Compiler.Pipeline.Test/LanguageServiceDiagnosticsRunOnTheCpuLaneTest.cs new file mode 100644 index 0000000000..4afb0478f4 --- /dev/null +++ b/test/MeshWeaver.Compiler.Pipeline.Test/LanguageServiceDiagnosticsRunOnTheCpuLaneTest.cs @@ -0,0 +1,117 @@ +using System; +using System.Collections.Concurrent; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using MeshWeaver.Fixture; +using MeshWeaver.Graph.Configuration; +using MeshWeaver.Hosting.Monolith.TestBase; +using MeshWeaver.Mesh; +using MeshWeaver.Mesh.Services; +using MeshWeaver.Mesh.Services.LanguageServer; +using Microsoft.Extensions.DependencyInjection; +using Xunit; + +namespace MeshWeaver.Compiler.Pipeline.Test; + +/// +/// Every diagnostics request of the language service binds its compilation on the CPU lane +/// (IoPoolNames.CompileCpu): a dedicated, bounded thread — never a ThreadPool worker +/// (Doc/Architecture/CompileOffTheThreadPool). +/// +/// Before the lane these ran as _ioPool.Run leaves on the Compile pool, i.e. ON the +/// ThreadPool: the pool caps them at the processor count, which is exactly the number of workers +/// the grain turns have before injection starts — and the script workspace additionally bound with +/// Roslyn's default ConcurrentBuild, fanning out further. All three arms are covered — the +/// node workspace (GetDiagnostics), a speculative proposal against a NodeType, and a script +/// cell — because each had its own GetDiagnostics call. +/// +/// The observation is the service's own seam, on the binding thread, at the moment it binds — +/// the thread's identity is a fact, not a timing. +/// +public class LanguageServiceDiagnosticsRunOnTheCpuLaneTest(ITestOutputHelper output) : MonolithMeshTestBase(output) +{ + private IMeshService MeshService => Mesh.ServiceProvider.GetRequiredService(); + + private MeshNodeLanguageService LanguageService => + (MeshNodeLanguageService)Mesh.ServiceProvider.GetRequiredService(); + + private sealed record Binding(bool OnPool, string? ThreadName); + + [Fact(Timeout = 300_000)] + public async Task EveryDiagnosticsArm_BindsOnTheCpuLane_NeverOnTheThreadPool() + { + var bindings = new ConcurrentQueue(); + var service = LanguageService; + service.OnDiagnoseStarting = () => bindings.Enqueue( + new Binding(Thread.CurrentThread.IsThreadPoolThread, Thread.CurrentThread.Name)); + try + { + var (typePath, sourcePath, typeName) = await ANodeTypeWithOneSource(); + + // Arm 1 — a speculative proposal against a NodeType. + await service.CheckSpeculativeOutcome(typePath, sourcePath, + $"public record {typeName} {{ public string Title {{ get; init; }} = string.Empty; }}") + .Should().Within(TestTimeouts.Convergence).Emit(cancellationToken: TestContext.Current.CancellationToken); + var afterSpeculative = bindings.Count; + + // Arm 2 — the node workspace. + await service.GetDiagnostics(typePath) + .Should().Within(TestTimeouts.Convergence).Emit(cancellationToken: TestContext.Current.CancellationToken); + var afterWorkspace = bindings.Count; + + // Arm 3 — a script cell (an owner that is not a NodeType). + var scriptId = $"ScriptOwner{Guid.NewGuid():N}"; + await MeshService.CreateNode(new MeshNode(scriptId, "type") + { + Name = "a script owner", + Content = new CodeConfiguration { Code = "not code", Language = "markdown" }, + State = MeshNodeState.Active, + }).Should().Within(TestTimeouts.Convergence).Emit(cancellationToken: TestContext.Current.CancellationToken); + await service.CheckSpeculativeOutcome($"type/{scriptId}", $"type/{scriptId}/Source/Probe.cs", "var probe = 1 + 1;") + .Should().Within(TestTimeouts.Convergence).Emit(cancellationToken: TestContext.Current.CancellationToken); + + afterSpeculative.Should().BeGreaterThan(0, "the speculative arm must bind through the seam"); + afterWorkspace.Should().BeGreaterThan(afterSpeculative, "the workspace arm must bind through the seam"); + bindings.Count.Should().BeGreaterThan(afterWorkspace, "the script arm must bind through the seam"); + } + finally + { + service.OnDiagnoseStarting = null; + } + + bindings.Where(b => b.OnPool).Should().BeEmpty( + "a diagnostics bind is pure Roslyn CPU; on a ThreadPool worker it holds a thread the silo's " + + "grain turns and routing legs need"); + bindings.Should().OnlyContain(b => b.ThreadName == "mw-cpu-lane", + "the bind must run on the bounded CPU lane, not on some other dedicated thread"); + } + + private async Task<(string TypePath, string SourcePath, string TypeName)> ANodeTypeWithOneSource() + { + var id = $"LaneType{Guid.NewGuid():N}"; + var typePath = $"type/{id}"; + + await MeshService.CreateNode(MeshNode.FromPath(typePath) with + { + Name = id, + NodeType = MeshNode.NodeTypePath, + Content = new NodeTypeDefinition { Configuration = $"config => config.WithContentType<{id}>()" }, + State = MeshNodeState.Active, + }).Should().Within(TestTimeouts.Convergence).Emit(cancellationToken: TestContext.Current.CancellationToken); + + await MeshService.CreateNode(new MeshNode($"{id}.cs", $"{typePath}/Source") + { + NodeType = "Code", + Name = $"{id}.cs", + Content = new CodeConfiguration + { + Code = $"public record {id} {{ public string Id {{ get; init; }} = string.Empty; }}", + Language = "csharp", + }, + State = MeshNodeState.Active, + }).Should().Within(TestTimeouts.Convergence).Emit(cancellationToken: TestContext.Current.CancellationToken); + + return (typePath, $"{typePath}/Source/{id}.cs", id); + } +} diff --git a/test/MeshWeaver.Compiler.Pipeline.Test/ServiceCompileStaysOffTheThreadPoolTest.cs b/test/MeshWeaver.Compiler.Pipeline.Test/ServiceCompileStaysOffTheThreadPoolTest.cs index 5cf112a40e..df03ca6dba 100644 --- a/test/MeshWeaver.Compiler.Pipeline.Test/ServiceCompileStaysOffTheThreadPoolTest.cs +++ b/test/MeshWeaver.Compiler.Pipeline.Test/ServiceCompileStaysOffTheThreadPoolTest.cs @@ -68,10 +68,12 @@ public async Task TheServicesEmit_RunsOnADedicatedThread_WithConcurrentBuildOff( var observed = 0; var onPoolThread = true; var concurrentBuild = true; + string? threadName = null; service.OnEmitStarting = compilation => { onPoolThread = Thread.CurrentThread.IsThreadPoolThread; concurrentBuild = compilation.Options.ConcurrentBuild; + threadName = Thread.CurrentThread.Name; Interlocked.Increment(ref observed); }; try @@ -93,8 +95,11 @@ public async Task TheServicesEmit_RunsOnADedicatedThread_WithConcurrentBuildOff( "the seam must have seen the ONE emit this compile ran — otherwise the two readings below " + "are defaults, not observations"); onPoolThread.Should().BeFalse( - "the service's Roslyn half must run on a CompileThread, never on a ThreadPool worker the " - + "silo's grain turns and routing legs need"); + "the service's Roslyn half must never run on a ThreadPool worker the silo's grain turns and " + + "routing legs need"); + threadName.Should().Be("mw-cpu-lane", + "the emit must run on the BOUNDED CPU lane (IoPoolNames.CompileCpu) — an unbounded dedicated " + + "thread per distinct NodeType turns a roll's hundreds of compiles into hundreds of threads"); concurrentBuild.Should().BeFalse( "the compilation the service emits must not fan out onto the ThreadPool either"); } diff --git a/test/MeshWeaver.Hosting.Test/CompileCpuLaneIsBoundedAndOffThePoolTest.cs b/test/MeshWeaver.Hosting.Test/CompileCpuLaneIsBoundedAndOffThePoolTest.cs new file mode 100644 index 0000000000..17a7fc468b --- /dev/null +++ b/test/MeshWeaver.Hosting.Test/CompileCpuLaneIsBoundedAndOffThePoolTest.cs @@ -0,0 +1,100 @@ +using System.Reactive.Linq; +using MeshWeaver.Mesh.Threading; +using MeshWeaver.Messaging; +using Xunit; + +namespace MeshWeaver.Hosting.Test; + +/// +/// The CPU lane () runs its blocking leaves on DEDICATED threads — +/// never ThreadPool workers — and never more of them at once than its cap +/// (Doc/Architecture/CompileOffTheThreadPool). +/// +/// Both halves matter. A NodeType emit on the ThreadPool holds a worker the grain turns need +/// (the roll-time routing starvation); on an UNBOUNDED dedicated thread (the #5327 shape) a roll's +/// hundreds of distinct-type compiles become hundreds of threads. The lane is the bound without the +/// pool. +/// +/// The leaves park on a volatile flag under a bounded +/// and are released in a finally, so the cap is observed while it is being tested — a leaf +/// that started beyond it would be counted by the leaves themselves, not inferred from timing. +/// +public class CompileCpuLaneIsBoundedAndOffThePoolTest +{ + private const int Cap = 2; + private const int Leaves = 6; + + private sealed class Probe + { + public volatile int Release; + public int Running; + public int MaxRunning; + public int OnPool; + public int Ran; + } + + private static IObservable ParkedLeaf(IIoPool pool, Probe probe) => + pool.InvokeBlocking(_ => + { + if (Thread.CurrentThread.IsThreadPoolThread) Interlocked.Increment(ref probe.OnPool); + var now = Interlocked.Increment(ref probe.Running); + int seen; + while ((seen = Volatile.Read(ref probe.MaxRunning)) < now + && Interlocked.CompareExchange(ref probe.MaxRunning, now, seen) != seen) { } + SpinWait.SpinUntil(() => probe.Release != 0, TimeSpan.FromSeconds(20)); + Interlocked.Decrement(ref probe.Running); + return Interlocked.Increment(ref probe.Ran); + }); + + private static async Task RunParked(IIoPool pool, CancellationToken ct) + { + var probe = new Probe(); + try + { + var all = Enumerable.Range(0, Leaves).Select(_ => ParkedLeaf(pool, probe).Await(ct)).ToArray(); + + SpinWait.SpinUntil(() => Volatile.Read(ref probe.Running) == Cap, TimeSpan.FromSeconds(10)) + .Should().BeTrue($"the lane must start {Cap} leaves while the rest queue"); + // Negative "nothing happened" check, bounded: with Cap leaves parked, a further leaf can + // only start if the cap is broken. + SpinWait.SpinUntil(() => Volatile.Read(ref probe.Running) > Cap, TimeSpan.FromMilliseconds(300)) + .Should().BeFalse($"no more than {Cap} leaves may run at once"); + + probe.Release = 1; + await Task.WhenAll(all); + } + finally + { + probe.Release = 1; + } + return probe; + } + + [Fact(Timeout = 60_000)] + public async Task TheCpuLane_RunsEveryLeafOffThePool_AndNeverMoreThanItsCap() + { + using var registry = new IoPoolRegistry(new IoPoolOptions { CompileCpu = Cap }); + var probe = await RunParked(registry.Get(IoPoolNames.CompileCpu), TestContext.Current.CancellationToken); + + probe.Ran.Should().Be(Leaves, "every queued leaf runs once a slot frees"); + probe.MaxRunning.Should().Be(Cap, "the cap is the bound — reached, never exceeded"); + probe.OnPool.Should().Be(0, + "a CPU-lane leaf must never occupy a ThreadPool worker the grain turns and routing legs need"); + } + + /// + /// NEGATIVE CONTROL — the ordinary pool, same cap, same leaves: + /// bounded, but ON the ThreadPool. This is what the NodeType emit and the language service's + /// diagnostics would have ridden without the lane; without the control the zero above could be an + /// instrument that counts nothing. + /// + [Fact(Timeout = 60_000)] + public async Task Control_AnOrdinaryPool_BoundsTheLeaves_ButRunsThemOnThePool() + { + using var registry = new IoPoolRegistry(new IoPoolOptions { Compile = Cap }); + var probe = await RunParked(registry.Get(IoPoolNames.Compile), TestContext.Current.CancellationToken); + + probe.MaxRunning.Should().Be(Cap); + probe.OnPool.Should().Be(Leaves, "control: a borrowing pool runs every blocking leaf on a pool worker"); + } +} From 468c6ac34284cfc77277fc6544764d525da3bc1c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= <6334612+rbuergi@users.noreply.github.com> Date: Tue, 22 Sep 2026 22:51:19 +0200 Subject: [PATCH 2/2] fix(compiler): lane diagnose honours cancellation; a failed thread start returns its slot Review of #5375: DiagnoseInputs ran generators and GetDiagnostics with CancellationToken.None, so an abandoned failure diagnose held a CompileCpu slot to the end; it now takes the lane token. And a dedicated-thread drain whose Thread.Start throws now rolls back its reservation and withdraws the task instead of leaving the scheduler at its cap with no drain loop. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../MeshNodeCompilationService.cs | 2 +- src/MeshWeaver.Compiler/CompileDiagnostics.cs | 9 ++++++--- .../LimitedConcurrencyLevelTaskScheduler.cs | 18 ++++++++++++++++-- 3 files changed, 23 insertions(+), 6 deletions(-) diff --git a/src/MeshWeaver.Compiler.Pipeline/MeshNodeCompilationService.cs b/src/MeshWeaver.Compiler.Pipeline/MeshNodeCompilationService.cs index ea761eba75..f94f795fcf 100644 --- a/src/MeshWeaver.Compiler.Pipeline/MeshNodeCompilationService.cs +++ b/src/MeshWeaver.Compiler.Pipeline/MeshNodeCompilationService.cs @@ -1231,7 +1231,7 @@ private IObservable ValidateCellSurfaceSingleHome( .SelectMany(inputs => inputs is null ? Observable.Return>(Array.Empty()) // Pure Roslyn compute -> the bounded CPU lane, off the ThreadPool (CompileOffTheThreadPool). - : _cpuLane.InvokeBlocking(_ => CompileDiagnostics.DiagnoseInputs(inputs))) + : _cpuLane.InvokeBlocking(ct => CompileDiagnostics.DiagnoseInputs(inputs, ct))) // 🚨 BOUNDED with the same clock as the emit leg — this runs on the FAILURE path, // where a hang is worst: the compile has already failed and this is what stands // between that failure and its terminal Error write. Unbounded, a stalled diff --git a/src/MeshWeaver.Compiler/CompileDiagnostics.cs b/src/MeshWeaver.Compiler/CompileDiagnostics.cs index 44d01060b5..03faaf4493 100644 --- a/src/MeshWeaver.Compiler/CompileDiagnostics.cs +++ b/src/MeshWeaver.Compiler/CompileDiagnostics.cs @@ -61,7 +61,10 @@ internal static string FormatCompileFailure(string nodePath, IEnumerable - internal static IReadOnlyList DiagnoseInputs(CompilationInputs inputs) + /// The compilation inputs to diagnose. + /// Stops the bind when the caller stops waiting — on the CPU lane an abandoned + /// diagnose must release its slot, not run to the end holding it. + internal static IReadOnlyList DiagnoseInputs(CompilationInputs inputs, CancellationToken ct = default) { var trees = new List(inputs.Sources.Length + 2) { @@ -86,9 +89,9 @@ internal static string FormatCompileFailure(string nodePath, IEnumerable(), Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance, CancellationToken.None); + Array.Empty(), Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance, ct); - var diags = compilation.GetDiagnostics(); + var diags = compilation.GetDiagnostics(ct); if (diags.IsDefaultOrEmpty) return Array.Empty(); var result = new List(diags.Length); diff --git a/src/MeshWeaver.Mesh.Contract/Threading/LimitedConcurrencyLevelTaskScheduler.cs b/src/MeshWeaver.Mesh.Contract/Threading/LimitedConcurrencyLevelTaskScheduler.cs index 61f4aec693..84ed474cb2 100644 --- a/src/MeshWeaver.Mesh.Contract/Threading/LimitedConcurrencyLevelTaskScheduler.cs +++ b/src/MeshWeaver.Mesh.Contract/Threading/LimitedConcurrencyLevelTaskScheduler.cs @@ -64,11 +64,25 @@ protected override void QueueTask(Task task) // ThreadPool work item to drain it. lock (_tasks) { - _tasks.AddLast(task); + var node = _tasks.AddLast(task); if (_delegatesQueuedOrRunning < _maxDegreeOfParallelism) { ++_delegatesQueuedOrRunning; - NotifyThreadPoolOfPendingWork(); + try + { + NotifyThreadPoolOfPendingWork(); + } + catch + { + // Nothing will drain for this reservation (thread creation / start failed, e.g. + // out of memory for a stack): give the slot back and withdraw the task, or the + // count stays at the cap with no drain loop behind it and every later QueueTask + // queues behind a loop that does not exist. The fault propagates to the caller + // (TaskFactory.StartNew), which faults the leaf. + --_delegatesQueuedOrRunning; + _tasks.Remove(node); + throw; + } } } }