From e81ee75bdad800c9e4840e710a20fa289a7fcaee Mon Sep 17 00:00:00 2001 From: Ivan Murzak Date: Thu, 29 Oct 2026 15:09:49 -0700 Subject: [PATCH 1/4] fix: a snippet blocked in a wait that never ends stops at its budget (Python + C#) Python: a snippet that joined a thread it started which never ends (or waited on Future.result, Event.wait, queue.get, a long time.sleep) hung the host's thread for good: the wait blocks in a C lock acquire, where the watchdog's asynchronous SnippetTimeout is never delivered, and the snippet's threads were only stopped at run end, which could not come. - At the deadline the watchdog now stops the snippet's threads at once (the existing stop-on-agent-code mechanism), so join() / result() return. - While a snippet runs, Thread.join, Condition.wait (Event, Semaphore, Barrier, Queue, Future) and time.sleep wait in 50 ms slices on the snippet's threads: between two the snippet's thread raises its timeout (the watchdog sends no asynchronous stop into a sliced wait, so the wait's lock is held again) and a thread it started raises SnippetThreadStopped once the run is stopping. time.sleep(60) with a 1 s budget now stops after about 1 s. - A SnippetThreadStopped handed to the snippet by stopped work is the Timeout. C#: work the snippet starts already shared the budget, but a wait nothing would ever end (SemaphoreSlim/ManualResetEventSlim/CountdownEvent.Wait, Monitor.Wait, WaitHandle waits, Task.Wait/Result/GetResult of a task that never completes, Thread.Sleep, a join on blocked work) hung the host. GuardWriter now rewrites those waits in the agent's code into guard stand-ins of the same parameters that wait in 50 ms slices with a guard check between two. Child-process tests for every repro (Python 3.10-3.14, C# on the Roslyn matrix) check Timeout within budget + slack and that the process stays usable. Co-Authored-By: Claude Opus 5.5 --- dotnet/CodeExec.Roslyn/src/GuardWriter.cs | 134 +++++++++- dotnet/CodeExec.Tests/ChildProcess.cs | 24 +- dotnet/CodeExec.Tests/GuardTests.cs | 59 +++++ dotnet/CodeExec/ExecutionGuard.cs | 126 +++++++++ python/code_exec/runner.py | 300 ++++++++++++++++++---- python/tests/test_code_exec.py | 107 ++++++++ spec/contract.md | 51 +++- 7 files changed, 725 insertions(+), 76 deletions(-) diff --git a/dotnet/CodeExec.Roslyn/src/GuardWriter.cs b/dotnet/CodeExec.Roslyn/src/GuardWriter.cs index 9b572a1..afeb98c 100644 --- a/dotnet/CodeExec.Roslyn/src/GuardWriter.cs +++ b/dotnet/CodeExec.Roslyn/src/GuardWriter.cs @@ -219,9 +219,135 @@ static void Collect(SyntaxNode node, List edits, Semantics? semantics) case ImplicitObjectCreationExpressionSyntax creation when semantics != null: Workers(creation, creation.ArgumentList, semantics, edits); break; + case InvocationExpressionSyntax call when semantics != null && WaitNames.Contains(RightmostName(call.Expression)): + Wait(call, semantics, edits); + break; + case MemberAccessExpressionSyntax access when semantics != null && access.Name.Identifier.ValueText == "Result": + Result(access, semantics, edits); + break; + } + } + + /// The blocking waits rewrites, by name (what the semantic model resolves decides). + static readonly HashSet WaitNames = new HashSet { "Join", "Sleep", "Wait", "WaitAll", "WaitAny", "WaitOne", "GetResult" }; + + const string Ms = "int", Span = "System.TimeSpan", Token = "System.Threading.CancellationToken"; + const string TaskArray = "System.Threading.Tasks.Task[]", HandleArray = "System.Threading.WaitHandle[]"; + + /// + /// Each blocking wait the guard has a sliced stand-in for (Type.Method(parameter types) → guard method): the + /// stand-in takes the receiver first, then the same parameters. + /// + static readonly Dictionary WaitTargets = BuildWaitTargets(); + + static Dictionary BuildWaitTargets() + { + var targets = new Dictionary(); + void Add(string type, string method, string guard, params string[][] overloads) + { + foreach (var parameters in overloads) + targets[type + "." + method + "(" + string.Join(",", parameters) + ")"] = guard; + } + var none = new string[0]; + Add("System.Threading.Thread", "Join", "Join", none, new[] { Ms }, new[] { Span }); + Add("System.Threading.Thread", "Sleep", "Sleep", new[] { Ms }, new[] { Span }); + Add("System.Threading.Tasks.Task", "Wait", "TaskWait", none, new[] { Ms }, new[] { Span }, new[] { Token }, new[] { Ms, Token }); + Add("System.Threading.Tasks.Task", "WaitAll", "TaskWaitAll", new[] { TaskArray }, new[] { TaskArray, Ms }, new[] { TaskArray, Span }, + new[] { TaskArray, Token }, new[] { TaskArray, Ms, Token }, + new[] { "System.Collections.Generic.IEnumerable" }, + new[] { "System.ReadOnlySpan" }); // .NET 9+ `params ReadOnlySpan`: see Wait + Add("System.Threading.Tasks.Task", "WaitAny", "TaskWaitAny", new[] { TaskArray }, new[] { TaskArray, Ms }, new[] { TaskArray, Span }, + new[] { TaskArray, Token }, new[] { TaskArray, Ms, Token }); + foreach (var pair in new[] { new[] { "SemaphoreSlim", "SemaphoreWait" }, new[] { "ManualResetEventSlim", "EventWait" }, new[] { "CountdownEvent", "CountdownWait" } }) + Add("System.Threading." + pair[0], "Wait", pair[1], none, new[] { Ms }, new[] { Span }, new[] { Token }, new[] { Ms, Token }, new[] { Span, Token }); + Add("System.Threading.Monitor", "Wait", "MonitorWait", new[] { "object" }, new[] { "object", Ms }, new[] { "object", Span }); + Add("System.Threading.WaitHandle", "WaitOne", "WaitOne", none, new[] { Ms }, new[] { Span }); + Add("System.Threading.WaitHandle", "WaitAll", "WaitAll", new[] { HandleArray }, new[] { HandleArray, Ms }, new[] { HandleArray, Span }); + Add("System.Threading.WaitHandle", "WaitAny", "WaitAny", new[] { HandleArray }, new[] { HandleArray, Ms }, new[] { HandleArray, Span }); + return targets; + } + + static string Signature(IMethodSymbol method) => + method.ContainingType.OriginalDefinition.ToDisplayString() + "." + method.Name + "(" + + string.Join(",", method.Parameters.Select(p => p.Type.ToDisplayString())) + ")"; + + /// + /// A blocking wait () becomes the guard's sliced stand-in, which checks the guard between + /// slices: t.Join() → Join(t), Thread.Sleep(n) → Sleep(n); x.GetAwaiter().GetResult() + /// of a task becomes Ready(x).GetAwaiter().GetResult(). Arguments stay as written (none named), so they bind + /// the same way. Not inside an expression tree or a query. + /// + static void Wait(InvocationExpressionSyntax call, Semantics semantics, List edits) + { + if (semantics.HasErrors || !(semantics.Model.GetSymbolInfo(call).Symbol is IMethodSymbol method) + || call.ArgumentList.Arguments.Any(a => a.NameColon != null || !a.RefKindKeyword.IsKind(SyntaxKind.None)) + || InExpressionTree(call, semantics)) + return; + if (method.Name == "GetResult") + { + // x.GetAwaiter().GetResult() of a Task / Task. + if (call.Expression is MemberAccessExpressionSyntax outer && outer.Expression is InvocationExpressionSyntax awaiter + && awaiter.ArgumentList.Arguments.Count == 0 && awaiter.Expression is MemberAccessExpressionSyntax getAwaiter + && getAwaiter.Name.Identifier.ValueText == "GetAwaiter" && !(getAwaiter.Expression is BaseExpressionSyntax) + && semantics.Model.GetSymbolInfo(awaiter).Symbol is IMethodSymbol get && !get.IsStatic && IsTask(get.ContainingType)) + Ready(getAwaiter.Expression, edits); + return; } + var signature = Signature(method); + if (!WaitTargets.TryGetValue(signature, out var guard)) + return; + var arguments = call.ArgumentList.Arguments; + if (signature.IndexOf("ReadOnlySpan", System.StringComparison.Ordinal) >= 0 + && arguments.Any(a => !IsTask(semantics.Model.GetTypeInfo(a.Expression).Type))) + return; // only the expanded form (`WaitAll(a, b)`): its tasks bind to the stand-in's `params Task[]` + var target = ExecutionGuard.QualifiedName + "." + guard; + if (method.IsStatic) + { + edits.Add(new Edit(call.Expression.SpanStart, target, call.Expression.SpanStart, edits.Count, remove: call.Expression.Span.Length)); + return; + } + if (!(call.Expression is MemberAccessExpressionSyntax access) || access.Expression is BaseExpressionSyntax) + return; + // `x.Wait(args)` → `Guard.TaskWait(x, args)`: `.Wait(` gives way to `, ` (to nothing without arguments). + var receiver = access.Expression; + var open = call.ArgumentList.OpenParenToken.Span.End; + edits.Add(new Edit(receiver.SpanStart, target + "(", receiver.SpanStart, edits.Count)); + edits.Add(new Edit(receiver.Span.End, arguments.Count > 0 ? ", " : "", open, edits.Count, closer: true, remove: open - receiver.Span.End)); + } + + /// x.Result of a Task<T> becomes Ready(x).Result: a sliced wait, then the result. + static void Result(MemberAccessExpressionSyntax access, Semantics semantics, List edits) + { + if (semantics.HasErrors || access.Expression is BaseExpressionSyntax + || !(semantics.Model.GetSymbolInfo(access).Symbol is IPropertySymbol property) || property.IsStatic + || property.ContainingType.OriginalDefinition.ToDisplayString() != "System.Threading.Tasks.Task" + || access.Ancestors().Any(a => a is InvocationExpressionSyntax nameOf && nameOf.Expression is IdentifierNameSyntax name + && name.Identifier.ValueText == "nameof") + || InExpressionTree(access, semantics)) + return; + Ready(access.Expression, edits); + } + + static void Ready(ExpressionSyntax task, List edits) + { + edits.Add(new Edit(task.SpanStart, ExecutionGuard.QualifiedName + ".Ready(", task.SpanStart, edits.Count)); + edits.Add(new Edit(task.Span.End, ")", task.Span.End, edits.Count, closer: true)); } + static bool IsTask(ITypeSymbol? type) + { + for (; type != null; type = type.BaseType) + { + if (type.ToDisplayString() == "System.Threading.Tasks.Task") + return true; + } + return false; + } + + /// Inside an expression tree (a lambda that is not a delegate) or a query, which may be one. + static bool InExpressionTree(SyntaxNode node, Semantics semantics) => + node.Ancestors().Any(outer => outer is QueryExpressionSyntax || (outer is AnonymousFunctionExpressionSyntax function && !semantics.IsDelegate(function))); + /// The methods that start work on another thread or as a task (with the constructors of Thread and Task). static readonly HashSet WorkStartNames = new HashSet { "Run", "StartNew", "QueueUserWorkItem", "UnsafeQueueUserWorkItem" }; @@ -242,13 +368,9 @@ static void Collect(SyntaxNode node, List edits, Semantics? semantics) /// static void Workers(ExpressionSyntax call, ArgumentListSyntax arguments, Semantics semantics, List edits) { - if (semantics.HasErrors || !(semantics.Model.GetSymbolInfo(call).Symbol is IMethodSymbol method) || !(StartsWork(method) is bool task)) + if (semantics.HasErrors || !(semantics.Model.GetSymbolInfo(call).Symbol is IMethodSymbol method) || !(StartsWork(method) is bool task) + || InExpressionTree(call, semantics)) return; - foreach (var outer in call.Ancestors()) - { - if (outer is QueryExpressionSyntax || (outer is AnonymousFunctionExpressionSyntax function && !semantics.IsDelegate(function))) - return; - } var entry = task ? ExecutionGuard.TaskWorker : ExecutionGuard.Worker; foreach (var argument in arguments.Arguments) { diff --git a/dotnet/CodeExec.Tests/ChildProcess.cs b/dotnet/CodeExec.Tests/ChildProcess.cs index 5667e9f..3cf183f 100644 --- a/dotnet/CodeExec.Tests/ChildProcess.cs +++ b/dotnet/CodeExec.Tests/ChildProcess.cs @@ -22,8 +22,12 @@ public static int Main(string[] args) return 2; } var code = Encoding.UTF8.GetString(Convert.FromBase64String(args[1])); - var result = Fixtures.Runner().Execute(new ExecuteRequest { Code = code, TimeoutMs = int.Parse(args[2]) }); - Console.Out.Write(result.ErrorKind + "\n" + result.Error); + var runner = Fixtures.Runner(); + var result = runner.Execute(new ExecuteRequest { Code = code, TimeoutMs = int.Parse(args[2]) }); + // The process stays usable: another snippet (with a wait and a loop of its own) runs normally afterwards. + var after = runner.Execute(new ExecuteRequest { Code = "System.Threading.Thread.Sleep(100);\nvar s = 0;\nfor (var i = 0; i < 3; i++) s += i;\ns", TimeoutMs = 10_000 }); + var afterText = after.Success ? Convert.ToString(after.Value) : after.ErrorKind + ": " + after.Error; + Console.Out.Write(result.ErrorKind + "\n" + result.ExecuteMs + "\n" + afterText?.Replace("\n", " ") + "\n" + result.Error); Console.Out.Flush(); System.Threading.Thread.Sleep(int.Parse(args[3])); // what the snippet left running gets the time to end the process return 0; @@ -34,6 +38,16 @@ public static int Main(string[] args) /// snippet left running ends it within or keeps it from exiting. /// public static (string ErrorKind, string Error) Execute(string code, int timeoutMs = 0, int lingerMs = 0) + { + var run = Run(code, timeoutMs, lingerMs); + return (run.ErrorKind, run.Error); + } + + /// + /// Like , plus how long the snippet ran () and what a + /// snippet run after it in the same process returned (3 when the process is still usable). + /// + public static (string ErrorKind, string Error, long ExecuteMs, string After) Run(string code, int timeoutMs = 0, int lingerMs = 0) { // The build output, not Assembly.Location: the .NET Framework test host runs a shadow copy without its dependencies. var assembly = Fixtures.Path(typeof(ChildProcess).Assembly.GetName().Name + ".dll"); @@ -60,10 +74,10 @@ public static (string ErrorKind, string Error) Execute(string code, int timeoutM throw new TimeoutException("the child process did not end within 120 s"); } var stdout = output.Result; - var split = stdout.IndexOf('\n'); - if (process.ExitCode != 0 || split < 0) + var lines = stdout.Split(new[] { '\n' }, 4); + if (process.ExitCode != 0 || lines.Length < 4) throw new InvalidOperationException($"the child process ended with exit code {process.ExitCode}: {stdout} {stderr.Result}"); - return (stdout.Substring(0, split), stdout.Substring(split + 1)); + return (lines[0], lines[3], long.Parse(lines[1]), lines[2]); } } } diff --git a/dotnet/CodeExec.Tests/GuardTests.cs b/dotnet/CodeExec.Tests/GuardTests.cs index 728a3ae..8a6c038 100644 --- a/dotnet/CodeExec.Tests/GuardTests.cs +++ b/dotnet/CodeExec.Tests/GuardTests.cs @@ -307,6 +307,48 @@ public void StoppedThreadsNeverEndTheProcess(string code, string kind) Assert.True(errorKind == kind, errorKind + ": " + error); } + // A blocking wait the agent's code writes waits in slices with a guard check between two (spec §3.6): one that never + // ends stops at the budget, and the process stays usable. In a child process: unguarded, each of these hangs it. + [Theory] + [InlineData("var t = new System.Threading.Thread(() => { while (true) { } });\nt.Start();\nt.Join();")] + [InlineData("Task.Run(() => { while (true) { } }).Wait();")] + [InlineData("return Task.Run(() => { while (true) { } return 1; }).Result;")] + [InlineData("Task.WaitAll(Task.Run(() => { while (true) { } }));")] + [InlineData("new System.Threading.SemaphoreSlim(0).Wait();")] + [InlineData("new System.Threading.ManualResetEventSlim().Wait();")] + [InlineData("new System.Threading.CountdownEvent(1).Wait(System.Threading.CancellationToken.None);")] + [InlineData("var o = new object();\nlock (o) System.Threading.Monitor.Wait(o);")] + [InlineData("System.Threading.Thread.Sleep(60_000);")] + [InlineData("System.Threading.Thread.Sleep(System.Threading.Timeout.Infinite);")] + [InlineData("new System.Threading.ManualResetEvent(false).WaitOne();")] + [InlineData("System.Threading.WaitHandle.WaitAny(new System.Threading.WaitHandle[] { new System.Threading.ManualResetEvent(false) });")] + [InlineData("new TaskCompletionSource().Task.Wait(60_000);")] + [InlineData("return new TaskCompletionSource().Task.Result;")] + [InlineData("return new TaskCompletionSource().Task.GetAwaiter().GetResult();")] + [InlineData("var a = new TaskCompletionSource().Task;\nTask.WaitAll(a, a);")] + [InlineData("Task.WaitAny(new TaskCompletionSource().Task);")] + [InlineData("System.Threading.Thread.CurrentThread.Join();")] + // Work the snippet waits for that is itself blocked in a wait: it stops in its wait too. + [InlineData("var e = new System.Threading.ManualResetEventSlim();\nvar t = new System.Threading.Thread(() => e.Wait());\nt.Start();\nt.Join();")] + [InlineData("var e = new System.Threading.ManualResetEventSlim();\nTask.Run(() => e.Wait()).Wait();")] + public void BlockingWaitsStopAtTheBudget(string code) + { + var run = ChildProcess.Run(code, timeoutMs: 1000, lingerMs: 300); + Assert.True(run.ErrorKind == nameof(ErrorKind.Timeout), run.ErrorKind + ": " + run.Error); + Assert.StartsWith("System.TimeoutException: The snippet ran longer than its time budget of 1000 ms (timeoutMs)", run.Error); + Assert.True(run.ExecuteMs < 2500, $"stopped after {run.ExecuteMs} ms"); + Assert.Equal("3", run.After); + } + + [Fact] + public void ATimedOutWaitShowsTheAgentsLine() + { + var result = Fixtures.Runner().Execute(new ExecuteRequest { Code = "var s = new System.Threading.SemaphoreSlim(0);\ns\n .Wait();", TimeoutMs = 200 }); + Assert.Equal(ErrorKind.Timeout, result.ErrorKind); + Assert.Contains("snippet.cs:line 2", result.Error); + Assert.DoesNotContain(ExecutionGuard.TypeName, result.Error); + } + [Fact] public void WorkACallbackStartsAfterTheRunIsItsOwn() { @@ -374,6 +416,23 @@ public void GuardsCanBeTurnedOff() [InlineData("Func f = null;\ntry { Task.Run(f); } catch (ArgumentNullException) { return 13; }\nreturn 0;", "13")] [InlineData("try { Task.Run(() => { throw new InvalidOperationException(\"w\"); }).Wait(); } catch (AggregateException e) { return e.InnerException!.Message; }\nreturn \"\";", "w")] [InlineData("System.Linq.Expressions.Expression>> e = () => Task.Run(() => 1);\ne.Body.ToString().Contains(\"Worker\")", "False")] + // Blocking waits become sliced stand-ins: the same values, timeouts, exceptions and wake-ups. + [InlineData("var t = new System.Threading.Thread(() => System.Threading.Thread.Sleep(150));\nt.Start();\n$\"{t.Join(10)} {t.Join(5000)} {t.Join(TimeSpan.FromSeconds(1))}\"", "False True True")] + [InlineData("var s = new System.Threading.SemaphoreSlim(1);\n$\"{s.Wait(10)} {s.Wait(120)} {s.Wait(TimeSpan.FromMilliseconds(80), System.Threading.CancellationToken.None)}\"", "True False False")] + [InlineData("var s = new System.Threading.SemaphoreSlim(0);\ns.Wait(millisecondsTimeout: 100)", "False")] + [InlineData("var w = System.Diagnostics.Stopwatch.StartNew();\nSystem.Threading.Thread.Sleep(300);\nSystem.Threading.Thread.Sleep(TimeSpan.FromMilliseconds(100));\nw.ElapsedMilliseconds >= 390", "True")] + [InlineData("Task.WaitAny(Task.Delay(10_000), Task.FromResult(1))", "1")] + [InlineData("Task.WaitAny(new[] { Task.Delay(10_000) }, 120)", "-1")] + [InlineData("Task.WaitAll(new[] { Task.Delay(10_000) }, 120)", "False")] + [InlineData("System.Threading.WaitHandle.WaitAny(new System.Threading.WaitHandle[] { new System.Threading.ManualResetEvent(false) }, 120)", "258")] + [InlineData("var o = new object();\nvar t = new System.Threading.Thread(() => { System.Threading.Thread.Sleep(200); lock (o) System.Threading.Monitor.Pulse(o); });\nbool got;\nlock (o) { t.Start(); got = System.Threading.Monitor.Wait(o); }\nt.Join();\ngot", "True")] + [InlineData("var e = new System.Threading.ManualResetEventSlim();\nTask.Run(() => { System.Threading.Thread.Sleep(150); e.Set(); });\ne.Wait();\ne.IsSet", "True")] + [InlineData("Task.Run(async () => { await Task.Delay(150); return 4; }).GetAwaiter().GetResult()", "4")] + [InlineData("try { Task.Run(() => { System.Threading.Thread.Sleep(100); throw new InvalidOperationException(\"r\"); return 1; }).GetAwaiter().GetResult(); } catch (InvalidOperationException e) { return e.Message; }\nreturn \"\";", "r")] + [InlineData("try { return Task.Run(() => { System.Threading.Thread.Sleep(100); throw new InvalidOperationException(\"q\"); return 1; }).Result.ToString(); } catch (AggregateException e) { return e.InnerException!.Message; }", "q")] + [InlineData("var cts = new System.Threading.CancellationTokenSource(150);\ntry { new System.Threading.SemaphoreSlim(0).Wait(cts.Token); } catch (OperationCanceledException) { return \"canceled\"; }\nreturn \"\";", "canceled")] + [InlineData("var t = Task.FromResult(1);\nnameof(t.Result)", "Result")] + [InlineData("System.Linq.Expressions.Expression> e = () => new System.Threading.SemaphoreSlim(1).Wait(10);\ne.Body.ToString().Contains(\"SemaphoreWait\")", "False")] public async Task GuardedCodeKeepsItsMeaning(string code, string expected) { var result = await Fixtures.Runner().ExecuteAsync(new ExecuteRequest { Code = code }); diff --git a/dotnet/CodeExec/ExecutionGuard.cs b/dotnet/CodeExec/ExecutionGuard.cs index 7caf693..e29137d 100644 --- a/dotnet/CodeExec/ExecutionGuard.cs +++ b/dotnet/CodeExec/ExecutionGuard.cs @@ -21,6 +21,15 @@ public static class ExecutionGuard /// The guard type in the snippet's assembly (global namespace). public const string TypeName = "__CodeExecGuard"; + /// + /// The guard type as inserted code names it. A blocking wait the agent's code writes (Thread.Join, + /// Thread.Sleep, Task.Wait / WaitAll / WaitAny / Result / + /// GetAwaiter().GetResult(), SemaphoreSlim / ManualResetEventSlim / CountdownEvent.Wait, + /// Monitor.Wait, WaitHandle.WaitOne / WaitAll / WaitAny) becomes a call of the guard's + /// stand-in of the same name and parameters (the receiver first), which waits in slices with a check between two. + /// + public const string QualifiedName = "global::" + TypeName; + /// The file name the guard source compiles under (never shown to the agent). public const string FileName = ""; @@ -257,6 +266,123 @@ static TResult Get(Run run, global::System.Func work) public static global::System.Func TaskWorker(global::System.Func work) => Spawn(work) is Run run ? state => Get(run, () => work(state)) : work; + // Blocking waits the agent's code writes (Join, Wait, Sleep, ...) wait in slices of at most SliceMs, with a clock + // check between two: a wait that never ends stops at the budget, and work of an ended run stops in it too. + const int SliceMs = 50; + + static bool Sliced(int timeout, global::System.Func wait) + { + if (timeout < -1 || (timeout >= 0 && timeout <= SliceMs)) + return wait(timeout); // short, or invalid: the framework's own wait (and its exception) + var start = global::System.Environment.TickCount; + for (;;) + { + if (timeout > 0) + { + var left = timeout - unchecked(global::System.Environment.TickCount - start); + if (left <= SliceMs) + return wait(left > 0 ? left : 0); + } + var done = wait(SliceMs); + Check(); // also when the wait is over: what it waited for may have been stopped by the budget + if (done) + return true; + } + } + + /// A TimeSpan timeout in ms; -2 when out of range (the framework's own wait throws for it). + static int Ms(global::System.TimeSpan timeout) + { + var ms = (long)timeout.TotalMilliseconds; + return ms >= -1 && ms <= int.MaxValue ? (int)ms : -2; + } + + public static void Join(global::System.Threading.Thread thread) => Sliced(-1, ms => thread.Join(ms)); + public static bool Join(global::System.Threading.Thread thread, int millisecondsTimeout) => Sliced(millisecondsTimeout, ms => thread.Join(ms)); + public static bool Join(global::System.Threading.Thread thread, global::System.TimeSpan timeout) => Sliced(Ms(timeout), ms => thread.Join(ms)); + + public static void Sleep(int millisecondsTimeout) => Sliced(millisecondsTimeout, ms => { global::System.Threading.Thread.Sleep(ms); return false; }); + public static void Sleep(global::System.TimeSpan timeout) => Sleep(Ms(timeout)); + + public static void TaskWait(global::System.Threading.Tasks.Task task) => Sliced(-1, ms => task.Wait(ms)); + public static bool TaskWait(global::System.Threading.Tasks.Task task, int millisecondsTimeout) => Sliced(millisecondsTimeout, ms => task.Wait(ms)); + public static bool TaskWait(global::System.Threading.Tasks.Task task, global::System.TimeSpan timeout) => Sliced(Ms(timeout), ms => task.Wait(ms)); + public static void TaskWait(global::System.Threading.Tasks.Task task, global::System.Threading.CancellationToken cancellationToken) => Sliced(-1, ms => task.Wait(ms, cancellationToken)); + public static bool TaskWait(global::System.Threading.Tasks.Task task, int millisecondsTimeout, global::System.Threading.CancellationToken cancellationToken) => Sliced(millisecondsTimeout, ms => task.Wait(ms, cancellationToken)); + + public static void TaskWaitAll(params global::System.Threading.Tasks.Task[] tasks) => Sliced(-1, ms => global::System.Threading.Tasks.Task.WaitAll(tasks, ms)); + public static void TaskWaitAll(global::System.Collections.Generic.IEnumerable tasks) => TaskWaitAll(new global::System.Collections.Generic.List(tasks).ToArray()); + public static bool TaskWaitAll(global::System.Threading.Tasks.Task[] tasks, int millisecondsTimeout) => Sliced(millisecondsTimeout, ms => global::System.Threading.Tasks.Task.WaitAll(tasks, ms)); + public static bool TaskWaitAll(global::System.Threading.Tasks.Task[] tasks, global::System.TimeSpan timeout) => Sliced(Ms(timeout), ms => global::System.Threading.Tasks.Task.WaitAll(tasks, ms)); + public static void TaskWaitAll(global::System.Threading.Tasks.Task[] tasks, global::System.Threading.CancellationToken cancellationToken) => Sliced(-1, ms => global::System.Threading.Tasks.Task.WaitAll(tasks, ms, cancellationToken)); + public static bool TaskWaitAll(global::System.Threading.Tasks.Task[] tasks, int millisecondsTimeout, global::System.Threading.CancellationToken cancellationToken) => Sliced(millisecondsTimeout, ms => global::System.Threading.Tasks.Task.WaitAll(tasks, ms, cancellationToken)); + + public static int TaskWaitAny(params global::System.Threading.Tasks.Task[] tasks) => TaskWaitAny(tasks, -1, default); + public static int TaskWaitAny(global::System.Threading.Tasks.Task[] tasks, int millisecondsTimeout) => TaskWaitAny(tasks, millisecondsTimeout, default); + public static int TaskWaitAny(global::System.Threading.Tasks.Task[] tasks, global::System.TimeSpan timeout) => TaskWaitAny(tasks, Ms(timeout), default); + public static int TaskWaitAny(global::System.Threading.Tasks.Task[] tasks, global::System.Threading.CancellationToken cancellationToken) => TaskWaitAny(tasks, -1, cancellationToken); + public static int TaskWaitAny(global::System.Threading.Tasks.Task[] tasks, int millisecondsTimeout, global::System.Threading.CancellationToken cancellationToken) + { + var index = -1; + Sliced(millisecondsTimeout, ms => (index = global::System.Threading.Tasks.Task.WaitAny(tasks, ms, cancellationToken)) >= 0); + return index; + } + + /// The task once it has completed (.Result, .GetAwaiter().GetResult() then return at once). + public static T Ready(T task) where T : global::System.Threading.Tasks.Task + { + if (task != null) + Sliced(-1, ms => + { + try { return task.Wait(ms); } + catch (global::System.AggregateException) { return true; } // faulted or canceled: the caller's own call throws + }); + return task; + } + + public static void SemaphoreWait(global::System.Threading.SemaphoreSlim semaphore) => Sliced(-1, ms => semaphore.Wait(ms)); + public static bool SemaphoreWait(global::System.Threading.SemaphoreSlim semaphore, int millisecondsTimeout) => Sliced(millisecondsTimeout, ms => semaphore.Wait(ms)); + public static bool SemaphoreWait(global::System.Threading.SemaphoreSlim semaphore, global::System.TimeSpan timeout) => Sliced(Ms(timeout), ms => semaphore.Wait(ms)); + public static void SemaphoreWait(global::System.Threading.SemaphoreSlim semaphore, global::System.Threading.CancellationToken cancellationToken) => Sliced(-1, ms => semaphore.Wait(ms, cancellationToken)); + public static bool SemaphoreWait(global::System.Threading.SemaphoreSlim semaphore, int millisecondsTimeout, global::System.Threading.CancellationToken cancellationToken) => Sliced(millisecondsTimeout, ms => semaphore.Wait(ms, cancellationToken)); + public static bool SemaphoreWait(global::System.Threading.SemaphoreSlim semaphore, global::System.TimeSpan timeout, global::System.Threading.CancellationToken cancellationToken) => Sliced(Ms(timeout), ms => semaphore.Wait(ms, cancellationToken)); + + public static void EventWait(global::System.Threading.ManualResetEventSlim waitable) => Sliced(-1, ms => waitable.Wait(ms)); + public static bool EventWait(global::System.Threading.ManualResetEventSlim waitable, int millisecondsTimeout) => Sliced(millisecondsTimeout, ms => waitable.Wait(ms)); + public static bool EventWait(global::System.Threading.ManualResetEventSlim waitable, global::System.TimeSpan timeout) => Sliced(Ms(timeout), ms => waitable.Wait(ms)); + public static void EventWait(global::System.Threading.ManualResetEventSlim waitable, global::System.Threading.CancellationToken cancellationToken) => Sliced(-1, ms => waitable.Wait(ms, cancellationToken)); + public static bool EventWait(global::System.Threading.ManualResetEventSlim waitable, int millisecondsTimeout, global::System.Threading.CancellationToken cancellationToken) => Sliced(millisecondsTimeout, ms => waitable.Wait(ms, cancellationToken)); + public static bool EventWait(global::System.Threading.ManualResetEventSlim waitable, global::System.TimeSpan timeout, global::System.Threading.CancellationToken cancellationToken) => Sliced(Ms(timeout), ms => waitable.Wait(ms, cancellationToken)); + + public static void CountdownWait(global::System.Threading.CountdownEvent waitable) => Sliced(-1, ms => waitable.Wait(ms)); + public static bool CountdownWait(global::System.Threading.CountdownEvent waitable, int millisecondsTimeout) => Sliced(millisecondsTimeout, ms => waitable.Wait(ms)); + public static bool CountdownWait(global::System.Threading.CountdownEvent waitable, global::System.TimeSpan timeout) => Sliced(Ms(timeout), ms => waitable.Wait(ms)); + public static void CountdownWait(global::System.Threading.CountdownEvent waitable, global::System.Threading.CancellationToken cancellationToken) => Sliced(-1, ms => waitable.Wait(ms, cancellationToken)); + public static bool CountdownWait(global::System.Threading.CountdownEvent waitable, int millisecondsTimeout, global::System.Threading.CancellationToken cancellationToken) => Sliced(millisecondsTimeout, ms => waitable.Wait(ms, cancellationToken)); + public static bool CountdownWait(global::System.Threading.CountdownEvent waitable, global::System.TimeSpan timeout, global::System.Threading.CancellationToken cancellationToken) => Sliced(Ms(timeout), ms => waitable.Wait(ms, cancellationToken)); + + // Waiting again after a slice misses no Pulse: the lock is held again between two slices. + public static bool MonitorWait(object obj) => Sliced(-1, ms => global::System.Threading.Monitor.Wait(obj, ms)); + public static bool MonitorWait(object obj, int millisecondsTimeout) => Sliced(millisecondsTimeout, ms => global::System.Threading.Monitor.Wait(obj, ms)); + public static bool MonitorWait(object obj, global::System.TimeSpan timeout) => Sliced(Ms(timeout), ms => global::System.Threading.Monitor.Wait(obj, ms)); + + public static bool WaitOne(global::System.Threading.WaitHandle handle) => Sliced(-1, ms => handle.WaitOne(ms)); + public static bool WaitOne(global::System.Threading.WaitHandle handle, int millisecondsTimeout) => Sliced(millisecondsTimeout, ms => handle.WaitOne(ms)); + public static bool WaitOne(global::System.Threading.WaitHandle handle, global::System.TimeSpan timeout) => Sliced(Ms(timeout), ms => handle.WaitOne(ms)); + + public static bool WaitAll(global::System.Threading.WaitHandle[] waitHandles) => Sliced(-1, ms => global::System.Threading.WaitHandle.WaitAll(waitHandles, ms)); + public static bool WaitAll(global::System.Threading.WaitHandle[] waitHandles, int millisecondsTimeout) => Sliced(millisecondsTimeout, ms => global::System.Threading.WaitHandle.WaitAll(waitHandles, ms)); + public static bool WaitAll(global::System.Threading.WaitHandle[] waitHandles, global::System.TimeSpan timeout) => Sliced(Ms(timeout), ms => global::System.Threading.WaitHandle.WaitAll(waitHandles, ms)); + + public static int WaitAny(global::System.Threading.WaitHandle[] waitHandles) => WaitAny(waitHandles, -1); + public static int WaitAny(global::System.Threading.WaitHandle[] waitHandles, global::System.TimeSpan timeout) => WaitAny(waitHandles, Ms(timeout)); + public static int WaitAny(global::System.Threading.WaitHandle[] waitHandles, int millisecondsTimeout) + { + var index = global::System.Threading.WaitHandle.WaitTimeout; + Sliced(millisecondsTimeout, ms => (index = global::System.Threading.WaitHandle.WaitAny(waitHandles, ms)) != global::System.Threading.WaitHandle.WaitTimeout); + return index; + } + public static unsafe void Call() { Calls = 8; diff --git a/python/code_exec/runner.py b/python/code_exec/runner.py index 16fe137..29da349 100644 --- a/python/code_exec/runner.py +++ b/python/code_exec/runner.py @@ -40,6 +40,7 @@ _max_timeout_ms = 0 # the highest budget a request may ask for (set_max_timeout_ms); 0 = no cap _REFIRE_SECONDS = 0.1 # a snippet that catches the stop is stopped again this often _POLL_SECONDS = 0.02 # how often the watchdog looks whether the snippet ended +_SLICE_SECONDS = 0.05 # a blocking wait of the snippet's threads (join, Event.wait, time.sleep, ...) looks at the budget this often _MARGIN_SECONDS = 0.05 # a snippet ending this long before its deadline was never stopped: no need to wait for the watchdog # Python < 3.12 has no C stack check: a snippet that raises the recursion limit and then recurses in C (repr / json.dumps @@ -50,6 +51,8 @@ _CAPPED = sys.version_info < (3, 12) _max_recursion_limit = MAX_RECURSION_LIMIT _SET_RECURSION_LIMIT = sys.setrecursionlimit # bound once: the runner restores limits with it, whatever a snippet replaced +# Bound once: the runner's own threads sleep and look at the clock with these, whatever a snippet (or the runner) replaced. +_sleep, _monotonic, _is_alive = time.sleep, time.monotonic, threading.Thread.is_alive # JSON request field -> run() keyword argument (integers). _INT_FIELDS = { @@ -186,9 +189,9 @@ def _execute(result, logs, code, include, include_root, imports, include_outside out, err = _LogStream(logs, "info"), _LogStream(logs, "error") prefixes = [_prefix(folder) for folder in folders] saved = _HostState(code, sources) - watchdog = _Watchdog(timeout_ms) agent_files = {SNIPPET_FILE} | {path for path, _ in files} threads = _SnippetThreads(lambda filename: filename in agent_files or _under(filename, prefixes)) + watchdog = _Watchdog(timeout_ms, threads) try: sys.path[:0] = folders if folders: @@ -214,6 +217,9 @@ def _execute(result, logs, code, include, include_root, imports, include_outside except BaseException as error: # SystemExit, KeyboardInterrupt, ...: a snippet must never take the host down _make_room(saved) watchdog.stop() + if isinstance(error, SnippetThreadStopped) and threads.stop_type is not None and threads.stopping: + # A thread stopped at the deadline handed its stop on (Future.result): the run ran out of time. + error = threads.stop_type().with_traceback(error.__traceback__) result["errorKind"] = "Timeout" if isinstance(error, SnippetTimeout) else "Runtime" try: result["error"] = _format_error(error, root, prefixes, {path for path, _ in files}) @@ -272,14 +278,18 @@ def __str__(self): class _Watchdog: """Stops a snippet that outruns its time budget: a timer thread raises :class:`SnippetTimeout` in the snippet's thread (``PyThreadState_SetAsyncExc``), again every ``_REFIRE_SECONDS`` while the snippet catches it. Python delivers it - between bytecodes, so pure Python loops stop; a call into C (``time.sleep``, an engine call) finishes first. + between bytecodes, so pure Python loops stop; a call into C (an engine call) finishes first. At the deadline the + threads the snippet started are stopped too (:meth:`_SnippetThreads.deadline_passed`): a snippet waiting for one of + them (``join``, ``Future.result``) gets it back. A snippet's thread inside a sliced wait (:func:`_sliced`) gets no + asynchronous stop: the wait raises it itself between two slices, where the lock it waits with is held again. No lock is ever taken in the snippet's thread (a stop delivered right after acquiring one would leave it held): :meth:`stop` clears a flag, waits for the timer thread to end if it may have raised, then drops a stop raised but not delivered yet. None reaches the host after the run.""" - def __init__(self, timeout_ms): + def __init__(self, timeout_ms, threads): self._timeout_ms = timeout_ms + self._threads = threads self._thread_id = None self._deadline = 0.0 self._armed = self._fired = self._exited = False @@ -289,7 +299,8 @@ def start(self): return stop = type("SnippetTimeout", (SnippetTimeout,), {"budget_ms": self._timeout_ms}) self._thread_id = threading.get_ident() - self._deadline = time.monotonic() + self._timeout_ms / 1000.0 + self._deadline = _monotonic() + self._timeout_ms / 1000.0 + self._threads.arm(self._deadline, stop) self._armed = True _start_untracked(threading.Thread(target=self._watch, args=(stop,), name="code_exec watchdog", daemon=True)) @@ -297,12 +308,15 @@ def _watch(self, stop): try: deadline = self._deadline while self._armed: - now = time.monotonic() + now = _monotonic() if now < deadline: - time.sleep(min(_POLL_SECONDS, deadline - now)) + _sleep(min(_POLL_SECONDS, deadline - now)) elif self._armed: - self._fired = True - _async_exc(self._thread_id, stop) + first, self._fired = not self._fired, True + if not self._threads.main_waiting: + _async_exc(self._thread_id, stop) + if first: + self._threads.deadline_passed() deadline = now + _REFIRE_SECONDS finally: self._exited = True @@ -313,9 +327,9 @@ def stop(self): self._armed = False # Before the deadline (and never fired) the timer cannot raise any more: it looks at the flag before it # would. Otherwise wait until it is gone, then drop what it may have raised. - if self._fired or time.monotonic() >= self._deadline - _MARGIN_SECONDS: + if self._fired or _monotonic() >= self._deadline - _MARGIN_SECONDS: while not self._exited: - time.sleep(0.001) + _sleep(0.001) _async_exc(self._thread_id, None) self._thread_id = None except SnippetTimeout: # delivered while disarming: disarm again @@ -323,31 +337,131 @@ def stop(self): class SnippetThreadStopped(BaseException): - """Raised in a thread the snippet started, once the snippet has ended (spec section 4): threads may not outlive it.""" + """Raised in a thread the snippet started once the snippet has ended or ran out of its time budget (spec section 4): + threads may not outlive it.""" def __str__(self): - return "The snippet has ended, so this thread it started was stopped: join threads before the snippet returns." + return ("The snippet has ended (or ran out of its time budget), so this thread it started was stopped: join " + "threads before the snippet returns.") -# Runs whose snippet is running, and the threading.Thread.start the dispatcher stands in for while there is one. A host -# that replaced threading.Thread.start meanwhile keeps the dispatcher in its chain: it is then never installed again. +# Runs whose snippet is running. While there is one, the runner stands in for threading.Thread.start (tracking) and for +# the blocking waits (sliced): each entry is (owner, attribute, stand-in). A host that replaced one of them meanwhile +# keeps the stand-in in its chain: that one is then never installed again (_kept), its original stays in _originals. _thread_runs = [] -_saved_start = None -_dispatcher_kept = False +_originals = {} +_kept = set() +_thread_run = {} # thread ident -> the innermost run it belongs to (its snippet's thread, or a thread the run started) + + +def _original(name): + """What the stand-in ``name`` stands in for (the attribute itself while it is not installed).""" + original = _originals.get(name) + if original is None: + owner, attribute, _ = _PATCHES[name] + original = getattr(owner, attribute) + return original + + +def _install(): + for name, (owner, attribute, stand_in) in _PATCHES.items(): + if name not in _kept: + _originals[name] = getattr(owner, attribute) + setattr(owner, attribute, stand_in) + + +def _uninstall(): + for name, (owner, attribute, stand_in) in _PATCHES.items(): + if name in _kept: + continue + if getattr(owner, attribute) is stand_in: + setattr(owner, attribute, _originals.pop(name)) + else: + _kept.add(name) def _dispatch_start(thread, *args, **kwargs): """``threading.Thread.start`` while snippets run: a thread started from a snippet's thread (or one of its threads) is that run's.""" - for runs in list(_thread_runs): - if threading.get_ident() in runs.owners: - return runs.start_tracked(thread, args, kwargs) - return _saved_start(thread, *args, **kwargs) + run = _thread_run.get(threading.get_ident()) + if run is not None and run.active: + return run.start_tracked(thread, args, kwargs) + return _original("start")(thread, *args, **kwargs) def _start_untracked(thread): """Starts one of the runner's own threads: never a snippet's.""" - (_saved_start if _dispatcher_kept or threading.Thread.start is _dispatch_start else threading.Thread.start)(thread) + _original("start")(thread) + + +def _long(timeout): + """Whether a wait of ``timeout`` seconds (None = no bound) is long enough to slice. Anything else (a short or + negative one, NaN, infinity, a type Python's own rejects) goes to Python's own wait unchanged.""" + if timeout is None: + return True + return isinstance(timeout, (int, float)) and _SLICE_SECONDS < timeout < 1e9 + + +def _sliced(timeout, wait): + """A blocking wait of a snippet's thread (``wait(seconds)`` returns whether it is over), in slices of at most + ``_SLICE_SECONDS``: between two slices the thread is stopped when its run is (:meth:`_SnippetThreads.check`), so a + wait that never ends cannot outlast the time budget. Returns what the last slice returned, or ``None`` when the + thread belongs to no run (the caller then waits as Python does).""" + ident = threading.get_ident() + run = _thread_run.get(ident) + if run is None: + return None + end = None if timeout is None else _monotonic() + timeout + while True: + left = _SLICE_SECONDS if end is None else max(0.0, min(_SLICE_SECONDS, end - _monotonic())) + if run.ended and not run.stops(ident): + # Its run is over and nothing of the agent's is left on its stack: wait the rest as Python does. + return wait(None if end is None else max(0.0, end - _monotonic())) + done = run.wait_slice(ident, wait, left) + if done or (end is not None and _monotonic() >= end): + return done + + +def _sliced_sleep(seconds): + """``time.sleep`` while snippets run (spec section 4).""" + if not (_thread_run and _long(seconds) and seconds is not None): + return _original("sleep")(seconds) + sleep = _original("sleep") + if _sliced(seconds, lambda left: sleep(left) or False) is None: + sleep(seconds) + + +def _sliced_join(thread, timeout=None): + """``threading.Thread.join`` while snippets run (spec section 4).""" + join = _original("join") + if not (_thread_run and _long(timeout)): + return join(thread, timeout) + + def once(left): + join(thread, left) # raises as Python's own does: not started, the current thread + return not _is_alive(thread) + + if _sliced(timeout, once) is None: + join(thread, timeout) + + +def _sliced_wait(condition, timeout=None): + """``threading.Condition.wait`` while snippets run (spec section 4): ``Event.wait``, ``Semaphore.acquire``, + ``queue.Queue.get``, ``concurrent.futures.Future.result`` and ``Barrier.wait`` wait through it. Waiting again after + a slice misses no ``notify``: the condition's lock is held between two slices.""" + wait = _original("wait") + if not (_thread_run and _long(timeout)): + return wait(condition, timeout) + done = _sliced(timeout, lambda left: wait(condition, left)) + return wait(condition, timeout) if done is None else done + + +_PATCHES = { + "start": (threading.Thread, "start", _dispatch_start), + "join": (threading.Thread, "join", _sliced_join), + "wait": (threading.Condition, "wait", _sliced_wait), + "sleep": (time, "sleep", _sliced_sleep), +} class _Tracked: @@ -359,38 +473,100 @@ def __init__(self, thread): class _SnippetThreads: - """The threads a snippet starts: ``threading.Thread.start`` called from the snippet's thread, or from one of those - threads, while the snippet runs. When it ends, a ``threading.Timer`` not fired yet is cancelled and every other one - still alive gets :class:`SnippetThreadStopped` (``PyThreadState_SetAsyncExc``) whenever the agent's own code - (``is_agent_file``) is the frame it runs, again every ``_REFIRE_SECONDS`` while that code is on its stack. A thread with - none of it on its stack (a pool worker waiting for work) is left alone from then on. The thread swallows the stop: - nothing is printed.""" + """One run's threads: ``threading.Thread.start`` called from the snippet's thread, or from one of those threads, + while the snippet runs. At the deadline and when it ends, a ``threading.Timer`` not fired yet is cancelled (at the + end) and every other one still alive gets :class:`SnippetThreadStopped` (``PyThreadState_SetAsyncExc``) whenever the + agent's own code (``is_agent_file``) is the frame it runs, again every ``_REFIRE_SECONDS`` while that code is on its + stack; one inside a sliced wait raises it there itself. A thread with none of it on its stack (a pool worker waiting + for work) is left alone once the run has ended. The thread swallows the stop: nothing is printed.""" def __init__(self, is_agent_file): self._is_agent_file = is_agent_file self._tracked = {} # id(thread) -> _Tracked; only ever added to, so the stopper thread can read it any time - self.owners = set() - self._active = False + self._previous = {} # thread ident -> the run it belonged to before this one (nested runs) + self._stopper = None + self.main_ident = None + self.active = self.ended = self.stopping = False + self.main_waiting = False # the snippet's thread is inside a sliced wait: it raises the timeout itself + self.deadline, self.stop_type = 0.0, None def start(self): - global _saved_start if _SET_ASYNC_EXC is None: return - self.owners.add(threading.get_ident()) - if not _thread_runs and not _dispatcher_kept: - _saved_start, threading.Thread.start = threading.Thread.start, _dispatch_start + if not _thread_runs: + _install() _thread_runs.append(self) - self._active = True + self.main_ident = threading.get_ident() + self._enter(self.main_ident) + self.active = True + + def arm(self, deadline, stop_type): + self.deadline, self.stop_type = deadline, stop_type + + def _enter(self, ident): + self._previous[ident] = _thread_run.get(ident) + _thread_run[ident] = self + + def _leave(self, ident): + previous = self._previous.pop(ident, None) + if _thread_run.get(ident) is self: + if previous is None: + _thread_run.pop(ident, None) + else: + _thread_run[ident] = previous + + def _chain(self, ident): + """This run and the runs ``ident`` belonged to before it (nested runs), innermost first.""" + run = self + while run is not None: + yield run + run = run._previous.get(ident) + + def wait_slice(self, ident, wait, seconds): + """One slice of a sliced wait of thread ``ident``, then the stop it is due, if any.""" + mains = [run for run in self._chain(ident) if run.main_ident == ident] + for run in mains: + run.main_waiting = True + try: + done = wait(seconds) + finally: + for run in mains: + run.main_waiting = False + for run in self._chain(ident): # also when the wait is over: what it waited for may have been stopped + run.check(ident) + return done + + def check(self, ident): + """Raises the stop thread ``ident`` is due between two slices of a wait: the run's timeout in its snippet's thread + once the deadline has passed, :class:`SnippetThreadStopped` in a thread it started once it is stopping.""" + if ident == self.main_ident: + if self.stop_type is not None and _monotonic() >= self.deadline: + raise self.stop_type() + elif self.stops(ident): + raise SnippetThreadStopped() + + def stops(self, ident): + """Whether thread ``ident`` (one the run started) is to stop: the run is stopping and the agent's code is on the + calling thread's stack.""" + if not self.stopping or ident == self.main_ident: + return False + frame = sys._getframe(1) + while frame is not None: + if self._is_agent_file(frame.f_code.co_filename): + return True + frame = frame.f_back + return False def start_tracked(self, thread, args, kwargs): + start = _original("start") if id(thread) in self._tracked: # started twice: the second start raises as Python's own does - return _saved_start(thread, *args, **kwargs) + return start(thread, *args, **kwargs) entry = self._tracked[id(thread)] = _Tracked(thread) - run, owners = thread.run, self.owners + run = thread.run def tracked(): ident = threading.get_ident() - owners.add(ident) + self._enter(ident) try: try: run() @@ -398,7 +574,7 @@ def tracked(): while True: # a stop raised while leaving must not outlive the thread either try: entry.done = True - owners.discard(ident) + self._leave(ident) _async_exc(ident, None) break except SnippetThreadStopped: @@ -409,24 +585,31 @@ def tracked(): thread.run = tracked entry.starting = True try: - return _saved_start(thread, *args, **kwargs) + return start(thread, *args, **kwargs) finally: entry.starting = False + def deadline_passed(self): + """The watchdog's thread, at the deadline: the threads the snippet started stop now, not only once it has ended + (the snippet may be waiting for one of them). Never raises.""" + try: + self.stopping = True + if self._stop_running(): + self._watch_on() + except BaseException: + pass + def stop(self, logs): """Never raises: the host's state is restored after it, whatever the snippet's threads do.""" - global _dispatcher_kept - if not self._active: + if not self.active: return - self._active = False + self.active = False try: + self.ended = self.stopping = True _thread_runs.remove(self) - if not _thread_runs and not _dispatcher_kept: - if threading.Thread.start is _dispatch_start: - threading.Thread.start = _saved_start - else: - _dispatcher_kept = True - self.owners.discard(threading.get_ident()) + if not _thread_runs: + _uninstall() + self._leave(self.main_ident) entries = list(self._tracked.values()) for entry in entries: timer = entry.thread @@ -434,15 +617,21 @@ def stop(self, logs): threading.Timer.cancel(timer) entry.cancelled = True alive = [entry for entry in entries - if not entry.cancelled and (entry.starting or threading.Thread.is_alive(entry.thread))] + if not entry.cancelled and (entry.starting or _is_alive(entry.thread))] if alive: logs.add("warning", "%d thread(s) the snippet started were still running when it ended: each is stopped " "while it runs the snippet's code. Join threads before the snippet returns." % len(alive)) if self._stop_running(): - _start_untracked(threading.Thread(target=self._watch, name="code_exec thread stopper", daemon=True)) + self._watch_on() except BaseException: # a snippet's Thread subclass that raises: the rest of the run still ends cleanly pass + def _watch_on(self): + """Starts the stopper thread (once): it raises the stop again every ``_REFIRE_SECONDS``.""" + if self._stopper is None: + self._stopper = threading.Thread(target=self._watch, name="code_exec thread stopper", daemon=True) + _start_untracked(self._stopper) + def _stop_running(self): """Raises the stop in the tracked threads running the agent's code; returns whether any is still to watch.""" frames = sys._current_frames() @@ -457,8 +646,11 @@ def _stop_running(self): top = frame = frames.get(thread.ident) while frame is not None and not self._is_agent_file(frame.f_code.co_filename): frame = frame.f_back - if frame is None or not threading.Thread.is_alive(thread): - entry.dismissed = True # ended, never started, or none of the agent's code left on its stack + if frame is None or not _is_alive(thread): + if self.ended: # ended, never started, or none of the agent's code left on its stack + entry.dismissed = True + else: + watching = True # between two tasks, or not running the agent's code yet: look again continue watching = True if self._is_agent_file(top.f_code.co_filename) and not entry.done: @@ -468,8 +660,8 @@ def _stop_running(self): def _watch(self): try: while True: - time.sleep(_REFIRE_SECONDS) - if not self._stop_running(): + _sleep(_REFIRE_SECONDS) + if not self._stop_running() and self.ended: break except BaseException: # interpreter shutdown pass diff --git a/python/tests/test_code_exec.py b/python/tests/test_code_exec.py index f068ca4..cb05190 100644 --- a/python/tests/test_code_exec.py +++ b/python/tests/test_code_exec.py @@ -537,6 +537,113 @@ def host(): self.assertEqual([], result["logs"]) +_CHILD = """import json, sys, threading, time +sys.path.insert(0, %r) +import code_exec +started = time.monotonic() +result = code_exec.run(%r, timeout_ms=%d) +elapsed = time.monotonic() - started +after = code_exec.run("import threading, time\\ntime.sleep(0.2)\\nthreading.Event().wait(0.1)\\n1 + 1", timeout_ms=5000) +print(json.dumps({"result": result, "elapsed": elapsed, "after": after})) +sys.stdout.flush() +""" + + +@unittest.skipIf(runner._SET_ASYNC_EXC is None, "needs ctypes") +class BlockingWaitTests(unittest.TestCase): + """Spec section 4: a snippet blocked in a wait that never ends (a join on a thread that never ends, Future.result, + Event.wait, queue.get, time.sleep) still stops with Timeout at its budget. Each case runs in a child process: before + the fix it hung the interpreter for good.""" + + SPIN = "import threading\ndef spin():\n while True:\n pass\n" + + def run_child(self, code, timeout_ms=1000): + script = _CHILD % (os.path.join(ROOT, "python"), code, timeout_ms) + try: + # A daemon-less thread the snippet leaves blocked keeps a child from exiting: the output is read first. + child = subprocess.run([sys.executable, "-B", "-c", script], capture_output=True, text=True, timeout=60) + except subprocess.TimeoutExpired as error: + self.fail("the snippet hung the interpreter: %r" % (error.stdout,)) + self.assertEqual(0, child.returncode, child.stderr[-2000:]) + report = json.loads(child.stdout) + self.assertTrue(report["after"]["success"], report["after"]) # the interpreter is usable afterwards + self.assertEqual("2", report["after"]["value"]) + return report["result"], report["elapsed"] + + def assert_times_out(self, code, timeout_ms=1000): + result, elapsed = self.run_child(code, timeout_ms) + self.assertEqual("Timeout", result["errorKind"], result) + self.assertTrue(result["error"].startswith( + "SnippetTimeout: The snippet ran longer than its time budget of %d ms (timeoutMs)" % timeout_ms), result["error"]) + self.assertIn('snippet.py", line', result["error"]) + self.assertLess(elapsed, timeout_ms / 1000.0 + 1.5) + return result + + def test_join_on_a_thread_that_never_ends(self): + self.assert_times_out(self.SPIN + "t = threading.Thread(target=spin)\nt.start()\nt.join()") + + def test_future_result_of_work_that_never_ends(self): + self.assert_times_out("from concurrent.futures import ThreadPoolExecutor\n" + "ThreadPoolExecutor(2).submit(lambda: [0 for _ in iter(int, 1)]).result()") + + def test_pool_shutdown_waiting_for_work_that_never_ends(self): + self.assert_times_out("from concurrent.futures import ThreadPoolExecutor\nwith ThreadPoolExecutor(2) as pool:\n" + " pool.submit(lambda: [0 for _ in iter(int, 1)])") + + def test_waits_nothing_ever_ends(self): + codes = ["import threading\nthreading.Event().wait()", + "import queue\nqueue.Queue().get()", + "import threading\nthreading.Semaphore(0).acquire()", + "import threading\nc = threading.Condition()\nwith c:\n c.wait()", + "import threading\nthreading.Event().wait(60)", + "import time\ntime.sleep(60)"] + for code in codes: + with self.subTest(code): + self.assert_times_out(code) + + def test_join_on_a_thread_blocked_in_a_wait(self): + result = self.assert_times_out("import threading\ne = threading.Event()\ndef f():\n e.wait()\n" + "t = threading.Thread(target=f, daemon=True)\nt.start()\nt.join()") + self.assertEqual([], result["logs"]) # stopped in its wait at the deadline, gone by the end + + def test_result_of_work_blocked_in_a_wait(self): + self.assert_times_out("import threading\nfrom concurrent.futures import ThreadPoolExecutor\n" + "e = threading.Event()\nThreadPoolExecutor(1).submit(lambda: e.wait()).result()") + + def test_waits_that_end_keep_their_results(self): + code = ("import threading, time, queue\ne = threading.Event()\nq = queue.Queue()\n" + "def later():\n time.sleep(0.3)\n q.put(5)\n e.set()\n" + "t = threading.Thread(target=later)\nt.start()\nstarted = time.monotonic()\n" + "got = (e.wait(), q.get(), threading.Event().wait(0.2), t.join(), threading.Condition().acquire())\n" + "time.sleep(0.25)\n(got, round(time.monotonic() - started, 1))") + result, _ = self.run_child(code, timeout_ms=5000) + self.assertTrue(result["success"], result) + self.assertEqual("[[true, 5, false, null, true], 0.8]", result["value"]) + + def test_stand_ins_are_restored_after_the_run(self): + originals = (threading.Thread.start, threading.Thread.join, threading.Condition.wait, time.sleep) + result = code_exec.run("import threading, time\n(time.sleep is threading.Thread.join) or 1", timeout_ms=5000) + self.assertTrue(result["success"], result) + self.assertEqual(originals, (threading.Thread.start, threading.Thread.join, threading.Condition.wait, time.sleep)) + self.assertEqual({}, runner._thread_run) + + def test_host_threads_wait_as_python_does(self): + # A host thread waiting while a snippet runs is the host's own: its wait is not sliced, nor stopped. + done = threading.Event() + waited = [] + + def host(): + waited.append(threading.Event().wait(0.5)) + done.set() + + thread = threading.Thread(target=host) + thread.start() + result = code_exec.run("import threading\nthreading.Event().wait()", timeout_ms=200) + self.assertEqual("Timeout", result["errorKind"], result) + thread.join() + self.assertEqual([False], waited) + + CAPPED = runner._CAPPED # Python 3.12+ raise RecursionError before the C stack runs out: no cap there diff --git a/spec/contract.md b/spec/contract.md index fca730b..b188c4c 100644 --- a/spec/contract.md +++ b/spec/contract.md @@ -147,8 +147,19 @@ the host turns them off (`CodeRunnerOptions.ExecutionGuards`): - **Time.** Every loop iteration (`while`, `do`, `for`, `foreach`) and every `goto` counts down and, every few iterations, reads the clock (every iteration while iterations are slow); so do the calls above. Once `timeoutMs` is spent, the run stops with `Timeout`: `System.TimeoutException: The snippet ran longer than its time budget of N ms - (timeoutMs) and was stopped.`, with the frame where it was (`snippet.cs:line N`). Code that blocks inside a host or - framework call (`Thread.Sleep`, a lock, an engine call that never returns) is stopped only when the call returns. + (timeoutMs) and was stopped.`, with the frame where it was (`snippet.cs:line N`). The work the snippet started + (below) shares the budget: at the deadline it stops at its next check too, so a snippet waiting for it gets it back. +- **Blocking waits.** A blocking wait written in the agent's code waits in slices of at most 50 ms with a check between + two, so it stops at the budget (and in work of an ended run) even when nothing would ever end it: `Thread.Join`, + `Thread.Sleep`, `Task.Wait` / `Task.WaitAll` / `Task.WaitAny`, `Task.Result`, `task.GetAwaiter().GetResult()`, + `SemaphoreSlim.Wait`, `ManualResetEventSlim.Wait`, `CountdownEvent.Wait`, `Monitor.Wait`, `WaitHandle.WaitOne` / + `WaitAll` / `WaitAny`, in their forms without a timeout and with one (`int`, `TimeSpan`, with or without a + `CancellationToken`). Each becomes the guard's stand-in of the same parameters; values, exceptions, timeouts and + wake-ups stay the same, and a wait of at most 50 ms is the framework's own. Not covered (stopped only when the call + returns): such a wait with named arguments, through `?.`, `base.`, a method group or a delegate; `.NET 9` + `Task.WaitAll` given a span; `lock` / `Monitor.Enter` and other waits (`BlockingCollection.Take`, `Barrier`, + `SpinWait`, `Process.WaitForExit`, I/O); and every wait inside host, engine or framework code the snippet calls (an + engine call that never returns). A native thread cannot be stopped at all. - **Once stopped**, every further check throws again, and the agent's `catch` clauses that could catch the stop (no type, `Exception`, `SystemException` or the stop's own type) get a filter that lets it pass, so it reaches the runtime; `finally` blocks still run. That filter makes a few invalid programs compile (a later `catch` that an earlier @@ -166,8 +177,8 @@ the host turns them off (`CodeRunnerOptions.ExecutionGuards`): snippet started had not finished when it ended: they are stopped at their next loop iteration or call. Join or await work the snippet starts.` (work created but never started counts too). Not covered, so it may outlive the run: timers (`System.Threading.Timer`, `System.Timers.Timer`, engine timers) and other callbacks; an `async void` delegate's code - after its first `await` (`new Thread(async () => ...)`); work blocked in a wait or a host call (it stops once that - returns); threads started by native code or by host code the snippet calls; a start written through an alias of + after its first `await` (`new Thread(async () => ...)`); work blocked in a wait the guards do not slice or in a host + call (it stops once that returns); threads started by native code or by host code the snippet calls; a start written through an alias of `Thread` / `Task`, inside a query expression, or whose delegate type cannot be named (an anonymous type); a delegate stored first and started later through another API; and work that a callback left behind starts after the run (it is the callback's own). Caveat: a timer or other callback created inside asynchronous task work inherits the run @@ -199,8 +210,22 @@ the host turns them off (`CodeRunnerOptions.ExecutionGuards`): - **Time budget.** When `timeoutMs` is spent, a watchdog thread raises `SnippetTimeout` (a `BaseException`) in the snippet's thread, and again every 100 ms while the snippet catches it: `Timeout`, `SnippetTimeout: The snippet ran longer than its time budget of N ms (timeoutMs) and was stopped.` plus the snippet's frames. Python delivers it - between bytecodes: pure Python loops stop; a call into C (`time.sleep`, an engine call) finishes first, and a snippet - that catches `BaseException` in its loop is not stopped. Python 3.10 never delivers it, nor releases the GIL, in a loop + between bytecodes: pure Python loops stop; a call into C (an engine call) finishes first, and a snippet that catches + `BaseException` in its loop is not stopped. At the deadline the threads the snippet started are stopped too (see + Threads), so a snippet waiting for one of them (`join`, `Future.result`) gets it back. A `SnippetThreadStopped` that + reaches the snippet that way (`Future.result` of stopped work) is reported as the `Timeout`. +- **Blocking waits.** While a snippet runs, `threading.Thread.join`, `threading.Condition.wait` (and so `Event.wait`, + `Semaphore.acquire`, `Barrier.wait`, `queue.Queue.get` / `put`, `concurrent.futures.Future.result`, pool shutdown) + and `time.sleep` are replaced: called from the snippet's thread or a thread it started, a wait longer than 50 ms waits + in slices of at most 50 ms. Between two, the snippet's thread raises its `SnippetTimeout` once the deadline has + passed (where the wait's lock is held again: the watchdog sends no asynchronous stop into a sliced wait), and a + thread the snippet started raises `SnippetThreadStopped` once the run is stopping while the agent's code is on its + stack. So `time.sleep(60)` with a 1 s budget stops after about 1 s; values, timeouts and wake-ups stay as Python's + own (`notify` cannot fall between two slices: the lock is held there). Host threads wait as before; the originals + are restored when no snippet runs (a host that replaced one meanwhile keeps the stand-in in its chain, which then + waits as Python does outside runs). Not covered (stopped only when the call returns): `Lock.acquire` / + `RLock.acquire` (C), `queue.SimpleQueue.get`, `select` / sockets / `subprocess` waits, `asyncio`'s event loop waiting + for I/O, waits inside C extensions or the engine, and `from time import sleep` bound before the run. Python 3.10 never delivers it, nor releases the GIL, in a loop whose body begins with `try` or `with` (`while True: try: ...`): such a run hangs the interpreter there (3.11+ stop it). It needs `ctypes` (`PyThreadState_SetAsyncExc`); without it the budget is not enforced. The stop is disarmed before the runner returns: none reaches the host afterwards. @@ -208,12 +233,13 @@ the host turns them off (`CodeRunnerOptions.ExecutionGuards`): snippet's thread or from one of those threads while it runs) may not outlive it. When it ends (finished or timed out), a `threading.Timer` that has not fired is cancelled, and every other such thread still alive gets `code_exec.SnippetThreadStopped` (a `BaseException`, the same mechanism as the time budget) whenever the snippet's or - an include's code is what it runs, checked every 100 ms while such code is on its stack. The thread ends quietly + an include's code is what it runs, checked every 100 ms while such code is on its stack; this starts already at the + deadline of a run that times out. One inside a sliced wait (above) raises it there itself. The thread ends quietly (nothing is printed). One `warning` log entry counts the threads still running (cancelled timers aside): `N thread(s) the snippet started were still running when it ended: each is stopped while it runs the snippet's code. Join threads - before the snippet returns.` A thread inside library or host code (`queue.get`) gets no stop until it returns to the - snippet's code; one with none of it left on its stack (a `concurrent.futures` worker waiting for work) is left alone - from then on; one that catches `BaseException` every time is not stopped. `threading.Thread.start` is replaced only + before the snippet returns.` A thread inside library or host code gets no stop until it returns to the snippet's code + or reaches a sliced wait with that code on its stack; one with none of it left on its stack (a `concurrent.futures` + worker waiting for work, a thread whose target is `event.wait` itself) is left alone once the run has ended; one that catches `BaseException` every time is not stopped. `threading.Thread.start` is replaced only while a snippet runs. Not covered: `_thread.start_new_thread`, threads started by C code, and threads started by the snippet's threads after it has ended. Threads that host code starts while the snippet calls it count as the snippet's. Needs `ctypes`, like the time budget. @@ -266,7 +292,10 @@ Lengths are in characters: UTF-16 code units in C#, code points in Python. A lim guarded too (§3.6). Python < 3.12: a snippet's `sys.setrecursionlimit` is capped (default 3000) with a warning (§4). Threads and tasks a snippet starts stop when its run ends, with a `warning` when some had not finished (C# §3.6, Python §4). Hosts can cap `timeoutMs` (§1). C#: a using declaration (`using var x = ...;`) is a statement also as the - first line, and `global using` goes before the runtime's usings (§3.1). + first line, and `global using` goes before the runtime's usings (§3.1). A snippet blocked in a wait that never ends + (a join on a thread that never ends, `Future.result`, `Event.wait`, `SemaphoreSlim.Wait`, `Task.Result`, a long + sleep) stops at its budget instead of hanging the host: blocking waits are sliced (C# §3.6, Python §4), and Python + stops the snippet's threads already at the deadline. - **v0.3.1** — C#: `static readonly` fields of immutable types are not static state (§3.5). Python is unchanged. - **v0.3** — output limits (§6): request fields `maxValueChars`, `maxLogChars`, `maxLogLineChars`, `maxErrorChars` From 7fae625119333afe92cc33b8821f3a2c62dcea6f Mon Sep 17 00:00:00 2001 From: Ivan Murzak Date: Thu, 29 Oct 2026 15:22:55 -0700 Subject: [PATCH 2/4] simplify: shared agent-frame walk, one slicing entry point, forwarding wait overloads - Python: _sliced decides itself whether to slice (no None sentinel repeated by each stand-in); the snippet's thread of a run without a budget is not sliced (nothing could stop it); one _agent_frame helper for both stack walks. - C#: shorter wait overloads forward to the fullest one; WaitNames derives from the target table; cheap syntactic checks before the symbol lookup; the ReadOnlySpan overload is matched by its exact key. Skipped: Thread.Interrupt instead of call-site rewriting (it would throw into host/engine waits on the host's thread and can leave a pending interrupt for the host); per-slice Task.Wait allocations (~20/s while blocked, acceptable); QualifiedName in the older constants (cosmetic); sharing the old recursion child-process test helper. Co-Authored-By: Claude Opus 5.5 --- dotnet/CodeExec.Roslyn/src/GuardWriter.cs | 21 ++++--- dotnet/CodeExec/ExecutionGuard.cs | 48 ++++++++-------- python/code_exec/runner.py | 68 +++++++++++------------ 3 files changed, 68 insertions(+), 69 deletions(-) diff --git a/dotnet/CodeExec.Roslyn/src/GuardWriter.cs b/dotnet/CodeExec.Roslyn/src/GuardWriter.cs index afeb98c..f11a525 100644 --- a/dotnet/CodeExec.Roslyn/src/GuardWriter.cs +++ b/dotnet/CodeExec.Roslyn/src/GuardWriter.cs @@ -228,18 +228,22 @@ static void Collect(SyntaxNode node, List edits, Semantics? semantics) } } - /// The blocking waits rewrites, by name (what the semantic model resolves decides). - static readonly HashSet WaitNames = new HashSet { "Join", "Sleep", "Wait", "WaitAll", "WaitAny", "WaitOne", "GetResult" }; - const string Ms = "int", Span = "System.TimeSpan", Token = "System.Threading.CancellationToken"; const string TaskArray = "System.Threading.Tasks.Task[]", HandleArray = "System.Threading.WaitHandle[]"; + /// .NET 9+ Task.WaitAll(params ReadOnlySpan<Task>): rewritten only in its expanded form (see ). + const string WaitAllSpan = "System.Threading.Tasks.Task.WaitAll(System.ReadOnlySpan)"; + /// /// Each blocking wait the guard has a sliced stand-in for (Type.Method(parameter types) → guard method): the /// stand-in takes the receiver first, then the same parameters. /// static readonly Dictionary WaitTargets = BuildWaitTargets(); + /// The method names of , and GetResult (x.GetAwaiter().GetResult()). + static readonly HashSet WaitNames = new HashSet( + WaitTargets.Keys.Select(k => k.Substring(0, k.IndexOf('(')).Substring(k.LastIndexOf('.', k.IndexOf('(')) + 1)).Concat(new[] { "GetResult" })); + static Dictionary BuildWaitTargets() { var targets = new Dictionary(); @@ -254,8 +258,8 @@ void Add(string type, string method, string guard, params string[][] overloads) Add("System.Threading.Tasks.Task", "Wait", "TaskWait", none, new[] { Ms }, new[] { Span }, new[] { Token }, new[] { Ms, Token }); Add("System.Threading.Tasks.Task", "WaitAll", "TaskWaitAll", new[] { TaskArray }, new[] { TaskArray, Ms }, new[] { TaskArray, Span }, new[] { TaskArray, Token }, new[] { TaskArray, Ms, Token }, - new[] { "System.Collections.Generic.IEnumerable" }, - new[] { "System.ReadOnlySpan" }); // .NET 9+ `params ReadOnlySpan`: see Wait + new[] { "System.Collections.Generic.IEnumerable" }); + targets[WaitAllSpan] = "TaskWaitAll"; Add("System.Threading.Tasks.Task", "WaitAny", "TaskWaitAny", new[] { TaskArray }, new[] { TaskArray, Ms }, new[] { TaskArray, Span }, new[] { TaskArray, Token }, new[] { TaskArray, Ms, Token }); foreach (var pair in new[] { new[] { "SemaphoreSlim", "SemaphoreWait" }, new[] { "ManualResetEventSlim", "EventWait" }, new[] { "CountdownEvent", "CountdownWait" } }) @@ -279,9 +283,8 @@ static string Signature(IMethodSymbol method) => /// static void Wait(InvocationExpressionSyntax call, Semantics semantics, List edits) { - if (semantics.HasErrors || !(semantics.Model.GetSymbolInfo(call).Symbol is IMethodSymbol method) - || call.ArgumentList.Arguments.Any(a => a.NameColon != null || !a.RefKindKeyword.IsKind(SyntaxKind.None)) - || InExpressionTree(call, semantics)) + if (call.ArgumentList.Arguments.Any(a => a.NameColon != null || !a.RefKindKeyword.IsKind(SyntaxKind.None)) + || semantics.HasErrors || !(semantics.Model.GetSymbolInfo(call).Symbol is IMethodSymbol method) || InExpressionTree(call, semantics)) return; if (method.Name == "GetResult") { @@ -297,7 +300,7 @@ static void Wait(InvocationExpressionSyntax call, Semantics semantics, List= 0 + if (signature == WaitAllSpan && arguments.Any(a => !IsTask(semantics.Model.GetTypeInfo(a.Expression).Type))) return; // only the expanded form (`WaitAll(a, b)`): its tasks bind to the stand-in's `params Task[]` var target = ExecutionGuard.QualifiedName + "." + guard; diff --git a/dotnet/CodeExec/ExecutionGuard.cs b/dotnet/CodeExec/ExecutionGuard.cs index e29137d..94491bc 100644 --- a/dotnet/CodeExec/ExecutionGuard.cs +++ b/dotnet/CodeExec/ExecutionGuard.cs @@ -304,17 +304,17 @@ static int Ms(global::System.TimeSpan timeout) public static void Sleep(int millisecondsTimeout) => Sliced(millisecondsTimeout, ms => { global::System.Threading.Thread.Sleep(ms); return false; }); public static void Sleep(global::System.TimeSpan timeout) => Sleep(Ms(timeout)); - public static void TaskWait(global::System.Threading.Tasks.Task task) => Sliced(-1, ms => task.Wait(ms)); - public static bool TaskWait(global::System.Threading.Tasks.Task task, int millisecondsTimeout) => Sliced(millisecondsTimeout, ms => task.Wait(ms)); - public static bool TaskWait(global::System.Threading.Tasks.Task task, global::System.TimeSpan timeout) => Sliced(Ms(timeout), ms => task.Wait(ms)); - public static void TaskWait(global::System.Threading.Tasks.Task task, global::System.Threading.CancellationToken cancellationToken) => Sliced(-1, ms => task.Wait(ms, cancellationToken)); + public static void TaskWait(global::System.Threading.Tasks.Task task) => TaskWait(task, -1, default); + public static bool TaskWait(global::System.Threading.Tasks.Task task, int millisecondsTimeout) => TaskWait(task, millisecondsTimeout, default); + public static bool TaskWait(global::System.Threading.Tasks.Task task, global::System.TimeSpan timeout) => TaskWait(task, Ms(timeout), default); + public static void TaskWait(global::System.Threading.Tasks.Task task, global::System.Threading.CancellationToken cancellationToken) => TaskWait(task, -1, cancellationToken); public static bool TaskWait(global::System.Threading.Tasks.Task task, int millisecondsTimeout, global::System.Threading.CancellationToken cancellationToken) => Sliced(millisecondsTimeout, ms => task.Wait(ms, cancellationToken)); - public static void TaskWaitAll(params global::System.Threading.Tasks.Task[] tasks) => Sliced(-1, ms => global::System.Threading.Tasks.Task.WaitAll(tasks, ms)); - public static void TaskWaitAll(global::System.Collections.Generic.IEnumerable tasks) => TaskWaitAll(new global::System.Collections.Generic.List(tasks).ToArray()); - public static bool TaskWaitAll(global::System.Threading.Tasks.Task[] tasks, int millisecondsTimeout) => Sliced(millisecondsTimeout, ms => global::System.Threading.Tasks.Task.WaitAll(tasks, ms)); - public static bool TaskWaitAll(global::System.Threading.Tasks.Task[] tasks, global::System.TimeSpan timeout) => Sliced(Ms(timeout), ms => global::System.Threading.Tasks.Task.WaitAll(tasks, ms)); - public static void TaskWaitAll(global::System.Threading.Tasks.Task[] tasks, global::System.Threading.CancellationToken cancellationToken) => Sliced(-1, ms => global::System.Threading.Tasks.Task.WaitAll(tasks, ms, cancellationToken)); + public static void TaskWaitAll(params global::System.Threading.Tasks.Task[] tasks) => TaskWaitAll(tasks, -1, default); + public static void TaskWaitAll(global::System.Collections.Generic.IEnumerable tasks) => TaskWaitAll(new global::System.Collections.Generic.List(tasks).ToArray(), -1, default); + public static bool TaskWaitAll(global::System.Threading.Tasks.Task[] tasks, int millisecondsTimeout) => TaskWaitAll(tasks, millisecondsTimeout, default); + public static bool TaskWaitAll(global::System.Threading.Tasks.Task[] tasks, global::System.TimeSpan timeout) => TaskWaitAll(tasks, Ms(timeout), default); + public static void TaskWaitAll(global::System.Threading.Tasks.Task[] tasks, global::System.Threading.CancellationToken cancellationToken) => TaskWaitAll(tasks, -1, cancellationToken); public static bool TaskWaitAll(global::System.Threading.Tasks.Task[] tasks, int millisecondsTimeout, global::System.Threading.CancellationToken cancellationToken) => Sliced(millisecondsTimeout, ms => global::System.Threading.Tasks.Task.WaitAll(tasks, ms, cancellationToken)); public static int TaskWaitAny(params global::System.Threading.Tasks.Task[] tasks) => TaskWaitAny(tasks, -1, default); @@ -340,26 +340,26 @@ public static T Ready(T task) where T : global::System.Threading.Tasks.Task return task; } - public static void SemaphoreWait(global::System.Threading.SemaphoreSlim semaphore) => Sliced(-1, ms => semaphore.Wait(ms)); - public static bool SemaphoreWait(global::System.Threading.SemaphoreSlim semaphore, int millisecondsTimeout) => Sliced(millisecondsTimeout, ms => semaphore.Wait(ms)); - public static bool SemaphoreWait(global::System.Threading.SemaphoreSlim semaphore, global::System.TimeSpan timeout) => Sliced(Ms(timeout), ms => semaphore.Wait(ms)); - public static void SemaphoreWait(global::System.Threading.SemaphoreSlim semaphore, global::System.Threading.CancellationToken cancellationToken) => Sliced(-1, ms => semaphore.Wait(ms, cancellationToken)); + public static void SemaphoreWait(global::System.Threading.SemaphoreSlim semaphore) => SemaphoreWait(semaphore, -1, default); + public static bool SemaphoreWait(global::System.Threading.SemaphoreSlim semaphore, int millisecondsTimeout) => SemaphoreWait(semaphore, millisecondsTimeout, default); + public static bool SemaphoreWait(global::System.Threading.SemaphoreSlim semaphore, global::System.TimeSpan timeout) => SemaphoreWait(semaphore, Ms(timeout), default); + public static void SemaphoreWait(global::System.Threading.SemaphoreSlim semaphore, global::System.Threading.CancellationToken cancellationToken) => SemaphoreWait(semaphore, -1, cancellationToken); + public static bool SemaphoreWait(global::System.Threading.SemaphoreSlim semaphore, global::System.TimeSpan timeout, global::System.Threading.CancellationToken cancellationToken) => SemaphoreWait(semaphore, Ms(timeout), cancellationToken); public static bool SemaphoreWait(global::System.Threading.SemaphoreSlim semaphore, int millisecondsTimeout, global::System.Threading.CancellationToken cancellationToken) => Sliced(millisecondsTimeout, ms => semaphore.Wait(ms, cancellationToken)); - public static bool SemaphoreWait(global::System.Threading.SemaphoreSlim semaphore, global::System.TimeSpan timeout, global::System.Threading.CancellationToken cancellationToken) => Sliced(Ms(timeout), ms => semaphore.Wait(ms, cancellationToken)); - public static void EventWait(global::System.Threading.ManualResetEventSlim waitable) => Sliced(-1, ms => waitable.Wait(ms)); - public static bool EventWait(global::System.Threading.ManualResetEventSlim waitable, int millisecondsTimeout) => Sliced(millisecondsTimeout, ms => waitable.Wait(ms)); - public static bool EventWait(global::System.Threading.ManualResetEventSlim waitable, global::System.TimeSpan timeout) => Sliced(Ms(timeout), ms => waitable.Wait(ms)); - public static void EventWait(global::System.Threading.ManualResetEventSlim waitable, global::System.Threading.CancellationToken cancellationToken) => Sliced(-1, ms => waitable.Wait(ms, cancellationToken)); + public static void EventWait(global::System.Threading.ManualResetEventSlim waitable) => EventWait(waitable, -1, default); + public static bool EventWait(global::System.Threading.ManualResetEventSlim waitable, int millisecondsTimeout) => EventWait(waitable, millisecondsTimeout, default); + public static bool EventWait(global::System.Threading.ManualResetEventSlim waitable, global::System.TimeSpan timeout) => EventWait(waitable, Ms(timeout), default); + public static void EventWait(global::System.Threading.ManualResetEventSlim waitable, global::System.Threading.CancellationToken cancellationToken) => EventWait(waitable, -1, cancellationToken); + public static bool EventWait(global::System.Threading.ManualResetEventSlim waitable, global::System.TimeSpan timeout, global::System.Threading.CancellationToken cancellationToken) => EventWait(waitable, Ms(timeout), cancellationToken); public static bool EventWait(global::System.Threading.ManualResetEventSlim waitable, int millisecondsTimeout, global::System.Threading.CancellationToken cancellationToken) => Sliced(millisecondsTimeout, ms => waitable.Wait(ms, cancellationToken)); - public static bool EventWait(global::System.Threading.ManualResetEventSlim waitable, global::System.TimeSpan timeout, global::System.Threading.CancellationToken cancellationToken) => Sliced(Ms(timeout), ms => waitable.Wait(ms, cancellationToken)); - public static void CountdownWait(global::System.Threading.CountdownEvent waitable) => Sliced(-1, ms => waitable.Wait(ms)); - public static bool CountdownWait(global::System.Threading.CountdownEvent waitable, int millisecondsTimeout) => Sliced(millisecondsTimeout, ms => waitable.Wait(ms)); - public static bool CountdownWait(global::System.Threading.CountdownEvent waitable, global::System.TimeSpan timeout) => Sliced(Ms(timeout), ms => waitable.Wait(ms)); - public static void CountdownWait(global::System.Threading.CountdownEvent waitable, global::System.Threading.CancellationToken cancellationToken) => Sliced(-1, ms => waitable.Wait(ms, cancellationToken)); + public static void CountdownWait(global::System.Threading.CountdownEvent waitable) => CountdownWait(waitable, -1, default); + public static bool CountdownWait(global::System.Threading.CountdownEvent waitable, int millisecondsTimeout) => CountdownWait(waitable, millisecondsTimeout, default); + public static bool CountdownWait(global::System.Threading.CountdownEvent waitable, global::System.TimeSpan timeout) => CountdownWait(waitable, Ms(timeout), default); + public static void CountdownWait(global::System.Threading.CountdownEvent waitable, global::System.Threading.CancellationToken cancellationToken) => CountdownWait(waitable, -1, cancellationToken); + public static bool CountdownWait(global::System.Threading.CountdownEvent waitable, global::System.TimeSpan timeout, global::System.Threading.CancellationToken cancellationToken) => CountdownWait(waitable, Ms(timeout), cancellationToken); public static bool CountdownWait(global::System.Threading.CountdownEvent waitable, int millisecondsTimeout, global::System.Threading.CancellationToken cancellationToken) => Sliced(millisecondsTimeout, ms => waitable.Wait(ms, cancellationToken)); - public static bool CountdownWait(global::System.Threading.CountdownEvent waitable, global::System.TimeSpan timeout, global::System.Threading.CancellationToken cancellationToken) => Sliced(Ms(timeout), ms => waitable.Wait(ms, cancellationToken)); // Waiting again after a slice misses no Pulse: the lock is held again between two slices. public static bool MonitorWait(object obj) => Sliced(-1, ms => global::System.Threading.Monitor.Wait(obj, ms)); diff --git a/python/code_exec/runner.py b/python/code_exec/runner.py index 29da349..0a41268 100644 --- a/python/code_exec/runner.py +++ b/python/code_exec/runner.py @@ -217,7 +217,7 @@ def _execute(result, logs, code, include, include_root, imports, include_outside except BaseException as error: # SystemExit, KeyboardInterrupt, ...: a snippet must never take the host down _make_room(saved) watchdog.stop() - if isinstance(error, SnippetThreadStopped) and threads.stop_type is not None and threads.stopping: + if isinstance(error, SnippetThreadStopped) and threads.stopping: # A thread stopped at the deadline handed its stop on (Future.result): the run ran out of time. error = threads.stop_type().with_traceback(error.__traceback__) result["errorKind"] = "Timeout" if isinstance(error, SnippetTimeout) else "Runtime" @@ -312,7 +312,8 @@ def _watch(self, stop): if now < deadline: _sleep(min(_POLL_SECONDS, deadline - now)) elif self._armed: - first, self._fired = not self._fired, True + first = not self._fired + self._fired = True if not self._threads.main_waiting: _async_exc(self._thread_id, stop) if first: @@ -403,14 +404,14 @@ def _long(timeout): def _sliced(timeout, wait): - """A blocking wait of a snippet's thread (``wait(seconds)`` returns whether it is over), in slices of at most - ``_SLICE_SECONDS``: between two slices the thread is stopped when its run is (:meth:`_SnippetThreads.check`), so a - wait that never ends cannot outlast the time budget. Returns what the last slice returned, or ``None`` when the - thread belongs to no run (the caller then waits as Python does).""" + """A blocking wait (``wait(seconds)`` returns whether it is over; ``None`` = no bound). On a snippet's thread, a long + one waits in slices of at most ``_SLICE_SECONDS``: between two the thread is stopped when its run is + (:meth:`_SnippetThreads.check`), so a wait that never ends cannot outlast the time budget. Any other wait is Python's + own. Returns what the last slice returned.""" ident = threading.get_ident() - run = _thread_run.get(ident) - if run is None: - return None + run = _thread_run.get(ident) if _thread_run else None + if run is None or not _long(timeout) or not run.may_stop(ident): + return wait(timeout) end = None if timeout is None else _monotonic() + timeout while True: left = _SLICE_SECONDS if end is None else max(0.0, min(_SLICE_SECONDS, end - _monotonic())) @@ -424,25 +425,21 @@ def _sliced(timeout, wait): def _sliced_sleep(seconds): """``time.sleep`` while snippets run (spec section 4).""" - if not (_thread_run and _long(seconds) and seconds is not None): - return _original("sleep")(seconds) sleep = _original("sleep") - if _sliced(seconds, lambda left: sleep(left) or False) is None: - sleep(seconds) + if seconds is None: # Python's own TypeError, not a wait without bound + return sleep(seconds) + _sliced(seconds, lambda left: sleep(left) or False) def _sliced_join(thread, timeout=None): """``threading.Thread.join`` while snippets run (spec section 4).""" join = _original("join") - if not (_thread_run and _long(timeout)): - return join(thread, timeout) def once(left): join(thread, left) # raises as Python's own does: not started, the current thread return not _is_alive(thread) - if _sliced(timeout, once) is None: - join(thread, timeout) + _sliced(timeout, once) def _sliced_wait(condition, timeout=None): @@ -450,10 +447,7 @@ def _sliced_wait(condition, timeout=None): ``queue.Queue.get``, ``concurrent.futures.Future.result`` and ``Barrier.wait`` wait through it. Waiting again after a slice misses no ``notify``: the condition's lock is held between two slices.""" wait = _original("wait") - if not (_thread_run and _long(timeout)): - return wait(condition, timeout) - done = _sliced(timeout, lambda left: wait(condition, left)) - return wait(condition, timeout) if done is None else done + return _sliced(timeout, lambda left: wait(condition, left)) _PATCHES = { @@ -548,14 +542,18 @@ def check(self, ident): def stops(self, ident): """Whether thread ``ident`` (one the run started) is to stop: the run is stopping and the agent's code is on the calling thread's stack.""" - if not self.stopping or ident == self.main_ident: - return False - frame = sys._getframe(1) - while frame is not None: - if self._is_agent_file(frame.f_code.co_filename): - return True + return self.stopping and ident != self.main_ident and self._agent_frame(sys._getframe(1)) is not None + + def may_stop(self, ident): + """Whether a wait of thread ``ident`` may ever have to stop: a thread the run (or an outer run) started, or the + snippet's thread of a run with a time budget. Others are not worth slicing.""" + return any(run.main_ident != ident or run.stop_type is not None for run in self._chain(ident)) + + def _agent_frame(self, frame): + """The innermost frame of the agent's code from ``frame`` outwards, or None.""" + while frame is not None and not self._is_agent_file(frame.f_code.co_filename): frame = frame.f_back - return False + return frame def start_tracked(self, thread, args, kwargs): start = _original("start") @@ -643,14 +641,12 @@ def _stop_running(self): watching = True continue thread = entry.thread - top = frame = frames.get(thread.ident) - while frame is not None and not self._is_agent_file(frame.f_code.co_filename): - frame = frame.f_back - if frame is None or not _is_alive(thread): - if self.ended: # ended, never started, or none of the agent's code left on its stack - entry.dismissed = True - else: - watching = True # between two tasks, or not running the agent's code yet: look again + top = frames.get(thread.ident) + if self._agent_frame(top) is None or not _is_alive(thread): + # Ended, never started, or none of the agent's code left on its stack: once the run has ended, for good. + # Before (at the deadline) it may be between two tasks or not run the agent's code yet: look again. + entry.dismissed = self.ended + watching = watching or not self.ended continue watching = True if self._is_agent_file(top.f_code.co_filename) and not entry.done: From 57a1cce6e5c015bb87d21e5b06cec0dd43f1886a Mon Sep 17 00:00:00 2001 From: Ivan Murzak Date: Thu, 29 Oct 2026 16:16:20 -0700 Subject: [PATCH 3/4] code review: lost-notify-free sliced Condition.wait, safer C# rewrites Python: - A sliced Condition.wait re-entered Python's own wait per slice, so a notify landing as a slice timed out was lost (Queue.get / Event.wait could hang until the budget). It now keeps ONE waiter for the whole wait and acquires only the waiter lock in slices; a due stop is raised after the lock is held again. - The host's value converter runs unbudgeted (the run is disarmed first). - Finite timeouts of any size are sliced (was capped at 1e9 s). - Regression tests: notify across a slice boundary (Condition, Queue), the converter, Event.wait(1e10); the timing assertion is a range. C#: - Waits inside a conditional access (a?.Task.Result) and property pattern paths (x is { Task.Result: 3 }) are left alone: the rewrite broke them. - .NET 9+ Task.WaitAll(IEnumerable, CancellationToken = default) is matched (the table key lacked the token, so it hung). - A wait that succeeded is never followed by a stop (it may hold a semaphore or mutex); deadlines use Stopwatch (TickCount let sleeps end early). - A wait without bound where nothing could stop it is the framework's own, so it still runs a queued task inline; the remaining difference under a budget is documented. - ConfigureAwait(b).GetAwaiter().GetResult() is sliced too. - Spec: the Python time-budget paragraph is whole again; coverage updated. Co-Authored-By: Claude Opus 5.5 --- dotnet/CodeExec.Roslyn/src/GuardWriter.cs | 57 +++++++++++-- dotnet/CodeExec.Tests/GuardTests.cs | 47 ++++++++++- dotnet/CodeExec/ExecutionGuard.cs | 23 +++--- python/code_exec/runner.py | 97 +++++++++++++++++------ python/tests/test_code_exec.py | 29 ++++++- spec/contract.md | 47 ++++++----- 6 files changed, 235 insertions(+), 65 deletions(-) diff --git a/dotnet/CodeExec.Roslyn/src/GuardWriter.cs b/dotnet/CodeExec.Roslyn/src/GuardWriter.cs index f11a525..20f0d63 100644 --- a/dotnet/CodeExec.Roslyn/src/GuardWriter.cs +++ b/dotnet/CodeExec.Roslyn/src/GuardWriter.cs @@ -258,7 +258,7 @@ void Add(string type, string method, string guard, params string[][] overloads) Add("System.Threading.Tasks.Task", "Wait", "TaskWait", none, new[] { Ms }, new[] { Span }, new[] { Token }, new[] { Ms, Token }); Add("System.Threading.Tasks.Task", "WaitAll", "TaskWaitAll", new[] { TaskArray }, new[] { TaskArray, Ms }, new[] { TaskArray, Span }, new[] { TaskArray, Token }, new[] { TaskArray, Ms, Token }, - new[] { "System.Collections.Generic.IEnumerable" }); + new[] { "System.Collections.Generic.IEnumerable", Token }); // .NET 9+, the token optional targets[WaitAllSpan] = "TaskWaitAll"; Add("System.Threading.Tasks.Task", "WaitAny", "TaskWaitAny", new[] { TaskArray }, new[] { TaskArray, Ms }, new[] { TaskArray, Span }, new[] { TaskArray, Token }, new[] { TaskArray, Ms, Token }); @@ -288,12 +288,16 @@ static void Wait(InvocationExpressionSyntax call, Semantics semantics, List. + // x.GetAwaiter().GetResult() and x.ConfigureAwait(b).GetAwaiter().GetResult() of a Task / Task. if (call.Expression is MemberAccessExpressionSyntax outer && outer.Expression is InvocationExpressionSyntax awaiter && awaiter.ArgumentList.Arguments.Count == 0 && awaiter.Expression is MemberAccessExpressionSyntax getAwaiter - && getAwaiter.Name.Identifier.ValueText == "GetAwaiter" && !(getAwaiter.Expression is BaseExpressionSyntax) - && semantics.Model.GetSymbolInfo(awaiter).Symbol is IMethodSymbol get && !get.IsStatic && IsTask(get.ContainingType)) - Ready(getAwaiter.Expression, edits); + && getAwaiter.Name.Identifier.ValueText == "GetAwaiter" && !(getAwaiter.Expression is BaseExpressionSyntax)) + { + var awaited = getAwaiter.Expression; + var task = TaskCalled(awaited, "ConfigureAwait", semantics) ?? (IsTask(semantics.Model.GetTypeInfo(awaited).Type) ? awaited : null); + if (task != null) + Ready(task, edits); + } return; } var signature = Signature(method); @@ -309,7 +313,7 @@ static void Wait(InvocationExpressionSyntax call, Semantics semantics, List a is InvocationExpressionSyntax nameOf && nameOf.Expression is IdentifierNameSyntax name - && name.Identifier.ValueText == "nameof") + || access.Ancestors().Any(a => a is SubpatternSyntax // `x is { Task.Result: 3 }` (C# 10): a property path + || (a is InvocationExpressionSyntax nameOf && nameOf.Expression is IdentifierNameSyntax name && name.Identifier.ValueText == "nameof")) || InExpressionTree(access, semantics)) return; Ready(access.Expression, edits); } + /// The task x of when it is x.Name(...) on a Task / Task<T>, else null. + static ExpressionSyntax? TaskCalled(ExpressionSyntax call, string name, Semantics semantics) => + call is InvocationExpressionSyntax invocation && invocation.Expression is MemberAccessExpressionSyntax member + && member.Name.Identifier.ValueText == name && !(member.Expression is BaseExpressionSyntax) + && semantics.Model.GetSymbolInfo(invocation).Symbol is IMethodSymbol method && !method.IsStatic && IsTask(method.ContainingType) + ? member.Expression : null; + + /// + /// Whether the expression starts inside a conditional access (a?.Task.Result, a?[0].Wait()): no code + /// can be put in front of it there. + /// + static bool InConditionalAccess(ExpressionSyntax expression) + { + while (true) + { + switch (expression) + { + case MemberAccessExpressionSyntax member: + expression = member.Expression; + break; + case InvocationExpressionSyntax invocation: + expression = invocation.Expression; + break; + case ElementAccessExpressionSyntax element: + expression = element.Expression; + break; + case PostfixUnaryExpressionSyntax postfix: // `a?.b!.Result` + expression = postfix.Operand; + break; + default: + return expression is MemberBindingExpressionSyntax || expression is ElementBindingExpressionSyntax; + } + } + } + static void Ready(ExpressionSyntax task, List edits) { + if (InConditionalAccess(task)) + return; edits.Add(new Edit(task.SpanStart, ExecutionGuard.QualifiedName + ".Ready(", task.SpanStart, edits.Count)); edits.Add(new Edit(task.Span.End, ")", task.Span.End, edits.Count, closer: true)); } diff --git a/dotnet/CodeExec.Tests/GuardTests.cs b/dotnet/CodeExec.Tests/GuardTests.cs index 8a6c038..c2b65af 100644 --- a/dotnet/CodeExec.Tests/GuardTests.cs +++ b/dotnet/CodeExec.Tests/GuardTests.cs @@ -340,6 +340,51 @@ public void BlockingWaitsStopAtTheBudget(string code) Assert.Equal("3", run.After); } + // Waits the rewrite cannot reach into stay as written: they must still compile and mean what they meant. + [Theory] + [InlineData("var tcs = new TaskCompletionSource();\ntcs.SetResult(3);\nTaskCompletionSource n = tcs;\nreturn n?.Task.Result;", "3", 0)] + [InlineData("var tcs = new TaskCompletionSource();\ntcs.SetResult(3);\nTaskCompletionSource n = tcs;\nn?.Task.Wait();\nreturn n?.Task.GetAwaiter().GetResult();", "3", 0)] + [InlineData("var tasks = new[] { Task.FromResult(4) };\nreturn tasks?[0].Result;", "4", 0)] + [InlineData("var tcs = new TaskCompletionSource();\ntcs.SetResult(3);\nreturn tcs is { Task.Result: 3 };", "True", 4)] // C# 10 + public void WaitsTheRewriteSkipsStillCompile(string code, string expected, int minRoslyn) + { + if (typeof(Microsoft.CodeAnalysis.CSharp.CSharpCompilation).Assembly.GetName().Version!.Major < minRoslyn) + return; + var result = Fixtures.Runner().Execute(new ExecuteRequest { Code = code }); + Assert.True(result.Success, result.Error + " " + string.Join(" | ", result.Diagnostics)); + Assert.Equal(expected, result.Value); + } + + [Fact] + public void ConfigureAwaitGetResultIsSliced() + { + var run = ChildProcess.Run("return new TaskCompletionSource().Task.ConfigureAwait(false).GetAwaiter().GetResult();", timeoutMs: 300); + Assert.True(run.ErrorKind == nameof(ErrorKind.Timeout), run.ErrorKind + ": " + run.Error); + } + + // .NET 9+ Task.WaitAll(IEnumerable, CancellationToken = default). In a child process: unguarded, it hangs. + [Fact] + public void WaitAllOfAnEnumerableIsSliced() + { + if (typeof(Task).GetMethod("WaitAll", new[] { typeof(System.Collections.Generic.IEnumerable), typeof(System.Threading.CancellationToken) }) == null) + return; + var run = ChildProcess.Run("Task.WaitAll(new List { new TaskCompletionSource().Task });", timeoutMs: 1000); + Assert.True(run.ErrorKind == nameof(ErrorKind.Timeout), run.ErrorKind + ": " + run.Error); + Assert.Equal("3", run.After); + } + + // Where nothing could stop it (no budget, not work of a run), a wait without bound is the framework's own: it still + // runs a queued task inline, which a scheduler that only runs tasks inline needs. + [Fact] + public void AnUnboundedRunKeepsInlineTaskExecution() + { + const string code = "var t = Task.Factory.StartNew(() => 1, System.Threading.CancellationToken.None, TaskCreationOptions.None, new S());\nreturn t.Result;\n" + + "class S : TaskScheduler { protected override IEnumerable GetScheduledTasks() => null; protected override void QueueTask(Task t) { } protected override bool TryExecuteTaskInline(Task t, bool q) => TryExecuteTask(t); }"; + var result = Fixtures.Runner(configure: o => o.DefaultTimeoutMs = 0).Execute(new ExecuteRequest { Code = code }); + Assert.True(result.Success, result.Error); + Assert.Equal("1", result.Value); + } + [Fact] public void ATimedOutWaitShowsTheAgentsLine() { @@ -420,7 +465,7 @@ public void GuardsCanBeTurnedOff() [InlineData("var t = new System.Threading.Thread(() => System.Threading.Thread.Sleep(150));\nt.Start();\n$\"{t.Join(10)} {t.Join(5000)} {t.Join(TimeSpan.FromSeconds(1))}\"", "False True True")] [InlineData("var s = new System.Threading.SemaphoreSlim(1);\n$\"{s.Wait(10)} {s.Wait(120)} {s.Wait(TimeSpan.FromMilliseconds(80), System.Threading.CancellationToken.None)}\"", "True False False")] [InlineData("var s = new System.Threading.SemaphoreSlim(0);\ns.Wait(millisecondsTimeout: 100)", "False")] - [InlineData("var w = System.Diagnostics.Stopwatch.StartNew();\nSystem.Threading.Thread.Sleep(300);\nSystem.Threading.Thread.Sleep(TimeSpan.FromMilliseconds(100));\nw.ElapsedMilliseconds >= 390", "True")] + [InlineData("var w = System.Diagnostics.Stopwatch.StartNew();\nSystem.Threading.Thread.Sleep(300);\nSystem.Threading.Thread.Sleep(TimeSpan.FromMilliseconds(100));\nw.ElapsedMilliseconds >= 360", "True")] [InlineData("Task.WaitAny(Task.Delay(10_000), Task.FromResult(1))", "1")] [InlineData("Task.WaitAny(new[] { Task.Delay(10_000) }, 120)", "-1")] [InlineData("Task.WaitAll(new[] { Task.Delay(10_000) }, 120)", "False")] diff --git a/dotnet/CodeExec/ExecutionGuard.cs b/dotnet/CodeExec/ExecutionGuard.cs index 94491bc..c4a178a 100644 --- a/dotnet/CodeExec/ExecutionGuard.cs +++ b/dotnet/CodeExec/ExecutionGuard.cs @@ -272,21 +272,23 @@ static TResult Get(Run run, global::System.Func work) static bool Sliced(int timeout, global::System.Func wait) { - if (timeout < -1 || (timeout >= 0 && timeout <= SliceMs)) - return wait(timeout); // short, or invalid: the framework's own wait (and its exception) - var start = global::System.Environment.TickCount; + // Short or invalid: the framework's own wait (and its exception). Without bound where nothing can stop it (no + // budget, not work of a run): the framework's own too, which may run a queued task inline (sliced waits never do). + if (timeout < -1 || (timeout >= 0 && timeout <= SliceMs) || (timeout == -1 && _budget <= 0 && Current() == null)) + return wait(timeout); + var start = global::System.Diagnostics.Stopwatch.GetTimestamp(); // TickCount is too coarse: a sleep would end early for (;;) { if (timeout > 0) { - var left = timeout - unchecked(global::System.Environment.TickCount - start); + var elapsed = (global::System.Diagnostics.Stopwatch.GetTimestamp() - start) * 1000 / global::System.Diagnostics.Stopwatch.Frequency; + var left = timeout - elapsed; if (left <= SliceMs) - return wait(left > 0 ? left : 0); + return wait(left > 0 ? (int)left : 0); } - var done = wait(SliceMs); - Check(); // also when the wait is over: what it waited for may have been stopped by the budget - if (done) - return true; + if (wait(SliceMs)) + return true; // never a stop after the wait succeeded: it may have acquired something (a semaphore, a mutex) + Check(); } } @@ -311,7 +313,8 @@ static int Ms(global::System.TimeSpan timeout) public static bool TaskWait(global::System.Threading.Tasks.Task task, int millisecondsTimeout, global::System.Threading.CancellationToken cancellationToken) => Sliced(millisecondsTimeout, ms => task.Wait(ms, cancellationToken)); public static void TaskWaitAll(params global::System.Threading.Tasks.Task[] tasks) => TaskWaitAll(tasks, -1, default); - public static void TaskWaitAll(global::System.Collections.Generic.IEnumerable tasks) => TaskWaitAll(new global::System.Collections.Generic.List(tasks).ToArray(), -1, default); + public static void TaskWaitAll(global::System.Collections.Generic.IEnumerable tasks, global::System.Threading.CancellationToken cancellationToken = default) => + TaskWaitAll(new global::System.Collections.Generic.List(tasks).ToArray(), -1, cancellationToken); public static bool TaskWaitAll(global::System.Threading.Tasks.Task[] tasks, int millisecondsTimeout) => TaskWaitAll(tasks, millisecondsTimeout, default); public static bool TaskWaitAll(global::System.Threading.Tasks.Task[] tasks, global::System.TimeSpan timeout) => TaskWaitAll(tasks, Ms(timeout), default); public static void TaskWaitAll(global::System.Threading.Tasks.Task[] tasks, global::System.Threading.CancellationToken cancellationToken) => TaskWaitAll(tasks, -1, cancellationToken); diff --git a/python/code_exec/runner.py b/python/code_exec/runner.py index 0a41268..62d9344 100644 --- a/python/code_exec/runner.py +++ b/python/code_exec/runner.py @@ -6,6 +6,7 @@ import collections import io import json +import _thread import linecache import operator import os @@ -212,6 +213,7 @@ def _execute(result, logs, code, include, include_root, imports, include_outside value = eval(tail, scope) if tail is not None else None _make_room(saved) watchdog.stop() + threads.disarm() result["value"] = _convert(value) result["success"] = True except BaseException as error: # SystemExit, KeyboardInterrupt, ...: a snippet must never take the host down @@ -400,27 +402,44 @@ def _long(timeout): negative one, NaN, infinity, a type Python's own rejects) goes to Python's own wait unchanged.""" if timeout is None: return True - return isinstance(timeout, (int, float)) and _SLICE_SECONDS < timeout < 1e9 + return isinstance(timeout, (int, float)) and _SLICE_SECONDS < timeout < float("inf") -def _sliced(timeout, wait): - """A blocking wait (``wait(seconds)`` returns whether it is over; ``None`` = no bound). On a snippet's thread, a long - one waits in slices of at most ``_SLICE_SECONDS``: between two the thread is stopped when its run is - (:meth:`_SnippetThreads.check`), so a wait that never ends cannot outlast the time budget. Any other wait is Python's - own. Returns what the last slice returned.""" +def _run_to_slice(timeout): + """The run whose rules a wait of ``timeout`` seconds on this thread follows when it is sliced, else None: the thread + belongs to a run that may stop it and the wait is long (:func:`_long`).""" ident = threading.get_ident() run = _thread_run.get(ident) if _thread_run else None - if run is None or not _long(timeout) or not run.may_stop(ident): - return wait(timeout) + return run if run is not None and _long(timeout) and run.may_stop(ident) else None + + +def _slices(run, timeout, wait): + """Waits ``wait(seconds)`` (returns whether the wait is over; ``None`` = no bound) for at most ``timeout`` seconds + (``None`` = no bound) in slices of at most ``_SLICE_SECONDS``. Stops at the first slice after which the thread is due + a stop (:meth:`_SnippetThreads.due`). Returns ``(done, stop)``: what the last slice returned, and the stop not raised + yet (the caller raises it once its state is consistent again), or None.""" + ident = threading.get_ident() end = None if timeout is None else _monotonic() + timeout while True: - left = _SLICE_SECONDS if end is None else max(0.0, min(_SLICE_SECONDS, end - _monotonic())) if run.ended and not run.stops(ident): # Its run is over and nothing of the agent's is left on its stack: wait the rest as Python does. - return wait(None if end is None else max(0.0, end - _monotonic())) - done = run.wait_slice(ident, wait, left) - if done or (end is not None and _monotonic() >= end): - return done + return wait(None if end is None else max(0.0, end - _monotonic())), None + left = _SLICE_SECONDS if end is None else max(0.0, min(_SLICE_SECONDS, end - _monotonic())) + done, stop = run.wait_slice(ident, wait, left) + if stop is not None or done or (end is not None and _monotonic() >= end): + return done, stop + + +def _sliced(timeout, wait): + """A blocking wait on a snippet's thread, sliced (:func:`_slices`) when :func:`_run_to_slice` says so, so a wait that + never ends cannot outlast the time budget; any other wait is Python's own. Returns what the last slice returned.""" + run = _run_to_slice(timeout) + if run is None: + return wait(timeout) + done, stop = _slices(run, timeout, wait) + if stop is not None: + raise stop + return done def _sliced_sleep(seconds): @@ -444,10 +463,31 @@ def once(left): def _sliced_wait(condition, timeout=None): """``threading.Condition.wait`` while snippets run (spec section 4): ``Event.wait``, ``Semaphore.acquire``, - ``queue.Queue.get``, ``concurrent.futures.Future.result`` and ``Barrier.wait`` wait through it. Waiting again after - a slice misses no ``notify``: the condition's lock is held between two slices.""" - wait = _original("wait") - return _sliced(timeout, lambda left: wait(condition, left)) + ``queue.Queue.get`` / ``put``, ``concurrent.futures.Future.result`` and ``Barrier.wait`` wait through it. Sliced, it + is Python's own ``Condition.wait`` with the waiter lock acquired in slices: ONE waiter for the whole wait, so a + ``notify`` is never lost between two slices; a stop due between two is raised once the lock is held again.""" + run = _run_to_slice(timeout) + if run is None: + return _original("wait")(condition, timeout) + if not condition._is_owned(): + raise RuntimeError("cannot wait on un-acquired lock") + waiter = _thread.allocate_lock() + waiter.acquire() + condition._waiters.append(waiter) + saved_state = condition._release_save() + gotit = False + try: + gotit, stop = _slices(run, timeout, lambda left: waiter.acquire() if left is None else waiter.acquire(True, left)) + finally: # restore state no matter what, as Python's own does + condition._acquire_restore(saved_state) + if not gotit: + try: + condition._waiters.remove(waiter) + except ValueError: + pass + if stop is not None: + raise stop + return gotit _PATCHES = { @@ -517,7 +557,7 @@ def _chain(self, ident): run = run._previous.get(ident) def wait_slice(self, ident, wait, seconds): - """One slice of a sliced wait of thread ``ident``, then the stop it is due, if any.""" + """One slice of a sliced wait of thread ``ident``: ``(done, the stop it is due or None)``.""" mains = [run for run in self._chain(ident) if run.main_ident == ident] for run in mains: run.main_waiting = True @@ -526,18 +566,23 @@ def wait_slice(self, ident, wait, seconds): finally: for run in mains: run.main_waiting = False - for run in self._chain(ident): # also when the wait is over: what it waited for may have been stopped - run.check(ident) - return done + # Also when the wait is over: what it waited for may have been stopped by the deadline. + stop = next((stop for stop in (run.due(ident) for run in self._chain(ident)) if stop is not None), None) + return done, stop - def check(self, ident): - """Raises the stop thread ``ident`` is due between two slices of a wait: the run's timeout in its snippet's thread - once the deadline has passed, :class:`SnippetThreadStopped` in a thread it started once it is stopping.""" + def due(self, ident): + """The stop thread ``ident`` is due between two slices of a wait, or None: the run's timeout in its snippet's + thread once the deadline has passed, :class:`SnippetThreadStopped` in a thread it started once it is stopping.""" if ident == self.main_ident: if self.stop_type is not None and _monotonic() >= self.deadline: - raise self.stop_type() + return self.stop_type() elif self.stops(ident): - raise SnippetThreadStopped() + return SnippetThreadStopped() + return None + + def disarm(self): + """The snippet has finished in time: what runs after it (the host's value converter) is no longer budgeted.""" + self.stop_type = None def stops(self, ident): """Whether thread ``ident`` (one the run started) is to stop: the run is stopping and the agent's code is on the diff --git a/python/tests/test_code_exec.py b/python/tests/test_code_exec.py index cb05190..1094047 100644 --- a/python/tests/test_code_exec.py +++ b/python/tests/test_code_exec.py @@ -596,6 +596,7 @@ def test_waits_nothing_ever_ends(self): "import threading\nthreading.Semaphore(0).acquire()", "import threading\nc = threading.Condition()\nwith c:\n c.wait()", "import threading\nthreading.Event().wait(60)", + "import threading\nthreading.Event().wait(1e10)", "import time\ntime.sleep(60)"] for code in codes: with self.subTest(code): @@ -618,7 +619,33 @@ def test_waits_that_end_keep_their_results(self): "time.sleep(0.25)\n(got, round(time.monotonic() - started, 1))") result, _ = self.run_child(code, timeout_ms=5000) self.assertTrue(result["success"], result) - self.assertEqual("[[true, 5, false, null, true], 0.8]", result["value"]) + got, elapsed = json.loads(result["value"]) + self.assertEqual([True, 5, False, None, True], got) + self.assertTrue(0.7 <= elapsed < 1.5, elapsed) + + def test_a_notify_while_a_slice_ends_is_not_lost(self): + # The notifier holds the lock across the end of the waiter's slice: the wake-up must still reach the waiter. + code = ("import threading, time, queue\nc = threading.Condition()\nq = queue.Queue()\ngot = []\n" + "def waiter():\n with c:\n got.append(c.wait())\n" + "def consumer():\n got.append(q.get())\n" + "threads = [threading.Thread(target=waiter), threading.Thread(target=consumer)]\n" + "for t in threads:\n t.start()\ntime.sleep(0.1)\n" + "with c:\n time.sleep(0.2)\n c.notify()\n" + "with q.mutex:\n time.sleep(0.2)\n q._put(7)\n q.unfinished_tasks += 1\n q.not_empty.notify()\n" + "for t in threads:\n t.join(2)\n(sorted(got, key=str), [t.is_alive() for t in threads])") + result, _ = self.run_child(code, timeout_ms=10000) + self.assertTrue(result["success"], result) + self.assertEqual("[[7, true], [false, false]]", result["value"]) + + def test_the_host_converter_is_not_budgeted(self): + # The snippet finished in time: a converter that waits past the deadline does not turn it into a Timeout. + code_exec.set_value_converter(lambda value: (threading.Event().wait(0.4), value)[1]) + try: + result = code_exec.run("5", timeout_ms=150) + finally: + code_exec.set_value_converter(None) + self.assertTrue(result["success"], result) + self.assertEqual(5, result["value"]) def test_stand_ins_are_restored_after_the_run(self): originals = (threading.Thread.start, threading.Thread.join, threading.Condition.wait, time.sleep) diff --git a/spec/contract.md b/spec/contract.md index b188c4c..e90a992 100644 --- a/spec/contract.md +++ b/spec/contract.md @@ -154,9 +154,15 @@ the host turns them off (`CodeRunnerOptions.ExecutionGuards`): `Thread.Sleep`, `Task.Wait` / `Task.WaitAll` / `Task.WaitAny`, `Task.Result`, `task.GetAwaiter().GetResult()`, `SemaphoreSlim.Wait`, `ManualResetEventSlim.Wait`, `CountdownEvent.Wait`, `Monitor.Wait`, `WaitHandle.WaitOne` / `WaitAll` / `WaitAny`, in their forms without a timeout and with one (`int`, `TimeSpan`, with or without a - `CancellationToken`). Each becomes the guard's stand-in of the same parameters; values, exceptions, timeouts and - wake-ups stay the same, and a wait of at most 50 ms is the framework's own. Not covered (stopped only when the call - returns): such a wait with named arguments, through `?.`, `base.`, a method group or a delegate; `.NET 9` + `CancellationToken`; .NET 9+ `Task.WaitAll(IEnumerable)`), and `task.ConfigureAwait(b).GetAwaiter().GetResult()`. + Each becomes the guard's stand-in of the same parameters; values, exceptions, timeouts and wake-ups stay the same, a + wait that succeeded is never followed by a stop (it may have acquired a semaphore or a mutex), and a wait of at most + 50 ms is the framework's own, as is a wait without bound where nothing could stop it (a run without budget, not work + of a run). One difference: under a budget, a wait without bound for a task does not run the task inline on the + waiting thread when it is still queued (a task only its scheduler's inline path can run, such as a + synchronization-context scheduler waited on from its own thread, then waits until the budget). Not covered (stopped + only when the call returns): such a wait with named arguments, inside a conditional access (`a?.Task.Result`), as a + property pattern path (`x is { Task.Result: 1 }`), through `base.`, a method group or a delegate; .NET 9+ `Task.WaitAll` given a span; `lock` / `Monitor.Enter` and other waits (`BlockingCollection.Take`, `Barrier`, `SpinWait`, `Process.WaitForExit`, I/O); and every wait inside host, engine or framework code the snippet calls (an engine call that never returns). A native thread cannot be stopped at all. @@ -211,24 +217,27 @@ the host turns them off (`CodeRunnerOptions.ExecutionGuards`): snippet's thread, and again every 100 ms while the snippet catches it: `Timeout`, `SnippetTimeout: The snippet ran longer than its time budget of N ms (timeoutMs) and was stopped.` plus the snippet's frames. Python delivers it between bytecodes: pure Python loops stop; a call into C (an engine call) finishes first, and a snippet that catches - `BaseException` in its loop is not stopped. At the deadline the threads the snippet started are stopped too (see - Threads), so a snippet waiting for one of them (`join`, `Future.result`) gets it back. A `SnippetThreadStopped` that - reaches the snippet that way (`Future.result` of stopped work) is reported as the `Timeout`. + `BaseException` in its loop is not stopped. Python 3.10 never delivers it, nor releases the GIL, in a loop whose body + begins with `try` or `with` (`while True: try: ...`): such a run hangs the interpreter there (3.11+ stop it). At the + deadline the threads the snippet started are stopped too (see Threads), so a snippet waiting for one of them + (`join`, `Future.result`) gets it back; a `SnippetThreadStopped` that reaches the snippet that way (`Future.result` + of stopped work) is reported as the `Timeout`. It needs `ctypes` (`PyThreadState_SetAsyncExc`); without it the budget + is not enforced. The stop is disarmed before the runner returns (the host's value converter runs unbudgeted): none + reaches the host afterwards. - **Blocking waits.** While a snippet runs, `threading.Thread.join`, `threading.Condition.wait` (and so `Event.wait`, `Semaphore.acquire`, `Barrier.wait`, `queue.Queue.get` / `put`, `concurrent.futures.Future.result`, pool shutdown) - and `time.sleep` are replaced: called from the snippet's thread or a thread it started, a wait longer than 50 ms waits - in slices of at most 50 ms. Between two, the snippet's thread raises its `SnippetTimeout` once the deadline has - passed (where the wait's lock is held again: the watchdog sends no asynchronous stop into a sliced wait), and a - thread the snippet started raises `SnippetThreadStopped` once the run is stopping while the agent's code is on its - stack. So `time.sleep(60)` with a 1 s budget stops after about 1 s; values, timeouts and wake-ups stay as Python's - own (`notify` cannot fall between two slices: the lock is held there). Host threads wait as before; the originals - are restored when no snippet runs (a host that replaced one meanwhile keeps the stand-in in its chain, which then - waits as Python does outside runs). Not covered (stopped only when the call returns): `Lock.acquire` / - `RLock.acquire` (C), `queue.SimpleQueue.get`, `select` / sockets / `subprocess` waits, `asyncio`'s event loop waiting - for I/O, waits inside C extensions or the engine, and `from time import sleep` bound before the run. Python 3.10 never delivers it, nor releases the GIL, in a loop - whose body begins with `try` or `with` (`while True: try: ...`): such a run hangs the interpreter there (3.11+ stop - it). It needs `ctypes` (`PyThreadState_SetAsyncExc`); without it - the budget is not enforced. The stop is disarmed before the runner returns: none reaches the host afterwards. + and `time.sleep` are replaced: called from a thread the snippet started, or from the snippet's thread when the run + has a budget, a finite wait longer than 50 ms (or one without bound) waits in slices of at most 50 ms. Between two, + the snippet's thread raises its `SnippetTimeout` once the deadline has passed (the watchdog sends no asynchronous + stop into a sliced wait), and a thread the snippet started raises `SnippetThreadStopped` once the run is stopping + while the agent's code is on its stack; a `Condition.wait` raises it only once it holds its lock again, as after any + wait. So `time.sleep(60)` with a 1 s budget stops after about 1 s. Values, timeouts and wake-ups stay as Python's own: + a sliced `Condition.wait` keeps ONE waiter for the whole wait (only the waiter lock is acquired in slices), so no + `notify` is lost between two slices. Host threads wait as before; the originals are restored when no snippet runs (a + host that replaced one meanwhile keeps the stand-in in its chain, which then waits as Python does outside runs). Not + covered (stopped only when the call returns): `Lock.acquire` / `RLock.acquire` (C), `queue.SimpleQueue.get`, + `select` / sockets / `subprocess` waits, `asyncio`'s event loop waiting for I/O, waits inside C extensions or the + engine, and `from time import sleep` bound before the run. - **Threads.** Threads the snippet starts (`threading.Thread.start`, `threading.Timer` included, called from the snippet's thread or from one of those threads while it runs) may not outlive it. When it ends (finished or timed out), a `threading.Timer` that has not fired is cancelled, and every other such thread still alive gets From 611254fca501e779c6d041520dec2e3774f54d6d Mon Sep 17 00:00:00 2001 From: Ivan Murzak Date: Thu, 29 Oct 2026 16:45:57 -0700 Subject: [PATCH 4/4] code review: a run that outran its budget in an uninterruptible call is a Timeout Godot wave 7: `Thread.Sleep(60000); 1` returned success after 60 s at any budget (now sliced, see the earlier commits). For waits no slicing can reach (a lock, a host or engine call, a C call in Python), a run that returns after its budget is no longer a success: - C#: the synchronous part of a guarded run is timed from Arm to SetResult; past the budget it is `Timeout` ("it finished only after M ms, blocked in a call the guards cannot interrupt"). Pending tasks keep RunAsync's own budget, and a faulted task still reports its own error. - Python: when the watchdog fired while the snippet still ran (it swallowed the stop or returned before it was raised again), the run is `Timeout` with the same kind of message. - Tests: Thread.Sleep(60000)/(2000) and Task.Delay(60000).Wait() with values after them, an unsliced named-argument Sleep past the budget (C#), a raw lock acquire past the budget whose stop is swallowed (Python); sliced cases assert they were stopped in the wait. A log assertion that raced is gone. Reviewed by a report-only reviewer agent; its findings (async false Timeouts, a test that passed without the change, a weak message) are fixed here. Co-Authored-By: Claude Opus 5.5 --- dotnet/CodeExec.Tests/GuardTests.cs | 13 +++++++++++++ dotnet/CodeExec/CodeRunner.cs | 7 +++++++ python/code_exec/runner.py | 11 +++++++++++ python/tests/test_code_exec.py | 16 +++++++++++++--- spec/contract.md | 8 ++++++-- 5 files changed, 50 insertions(+), 5 deletions(-) diff --git a/dotnet/CodeExec.Tests/GuardTests.cs b/dotnet/CodeExec.Tests/GuardTests.cs index c2b65af..f45f81c 100644 --- a/dotnet/CodeExec.Tests/GuardTests.cs +++ b/dotnet/CodeExec.Tests/GuardTests.cs @@ -320,6 +320,9 @@ public void StoppedThreadsNeverEndTheProcess(string code, string kind) [InlineData("var o = new object();\nlock (o) System.Threading.Monitor.Wait(o);")] [InlineData("System.Threading.Thread.Sleep(60_000);")] [InlineData("System.Threading.Thread.Sleep(System.Threading.Timeout.Infinite);")] + [InlineData("System.Threading.Thread.Sleep(60000);\n1")] + [InlineData("System.Threading.Thread.Sleep(2000);\n1")] + [InlineData("Task.Delay(60_000).Wait();\n1")] [InlineData("new System.Threading.ManualResetEvent(false).WaitOne();")] [InlineData("System.Threading.WaitHandle.WaitAny(new System.Threading.WaitHandle[] { new System.Threading.ManualResetEvent(false) });")] [InlineData("new TaskCompletionSource().Task.Wait(60_000);")] @@ -336,6 +339,7 @@ public void BlockingWaitsStopAtTheBudget(string code) var run = ChildProcess.Run(code, timeoutMs: 1000, lingerMs: 300); Assert.True(run.ErrorKind == nameof(ErrorKind.Timeout), run.ErrorKind + ": " + run.Error); Assert.StartsWith("System.TimeoutException: The snippet ran longer than its time budget of 1000 ms (timeoutMs)", run.Error); + Assert.DoesNotContain("finished only after", run.Error); // stopped in the wait, not reported after it returned Assert.True(run.ExecuteMs < 2500, $"stopped after {run.ExecuteMs} ms"); Assert.Equal("3", run.After); } @@ -385,6 +389,15 @@ public void AnUnboundedRunKeepsInlineTaskExecution() Assert.Equal("1", result.Value); } + // A wait the guards cannot slice (here a named argument) that returns past the budget: the run is no success. + [Fact] + public void ARunThatOutranItsBudgetInAnUnslicedCallIsATimeout() + { + var result = Fixtures.Runner().Execute(new ExecuteRequest { Code = "System.Threading.Thread.Sleep(millisecondsTimeout: 600);\n1", TimeoutMs = 200 }); + Assert.Equal(ErrorKind.Timeout, result.ErrorKind); + Assert.StartsWith("System.TimeoutException: The snippet ran longer than its time budget of 200 ms (timeoutMs): it finished only after", result.Error); + } + [Fact] public void ATimedOutWaitShowsTheAgentsLine() { diff --git a/dotnet/CodeExec/CodeRunner.cs b/dotnet/CodeExec/CodeRunner.cs index b7b8dee..a8483fa 100644 --- a/dotnet/CodeExec/CodeRunner.cs +++ b/dotnet/CodeExec/CodeRunner.cs @@ -687,6 +687,7 @@ public Execution(CodeRunner runner, PreparedSnippet prepared) Type? _guard; object? _run; int _budgetMs; + readonly Stopwatch _running = new Stopwatch(); public bool Guarded => _guard != null; /// Starts the guard's time budget for this run (0 = none). @@ -698,6 +699,7 @@ public void Arm(int budgetMs) ExecutionGuard.Arm(_guard, budgetMs, Math.Max(_runner._options.MaxStackBytes, 0)); _run = ExecutionGuard.Begin(_guard); _budgetMs = budgetMs; + _running.Start(); } /// @@ -746,6 +748,7 @@ public void Fail(Exception e) /// Records the invocation result, judged by the DECLARED return type: awaitables become the task to await. public void SetResult(DeclaredResult declared, object? raw) { + _running.Stop(); // the synchronous part ended here: awaiting a pending task is RunAsync's budget if (raw != null && declared.IsValueTask) raw = declared.Declared.GetMethod("AsTask", Type.EmptyTypes)!.Invoke(raw, null); if (raw is Task task && declared.IsTask) @@ -785,6 +788,10 @@ public ExecuteResult Finish() else if (stopped != 0) Fail(ErrorKind.Runtime, "System.InsufficientExecutionStackException: Recursion too deep in a thread the snippet started: " + "it was stopped before a stack overflow could end the host process; bound the recursion depth or use a loop."); + else if (_errorKind == ErrorKind.None && _budgetMs > 0 && _running.ElapsedMilliseconds > _budgetMs + && (_task == null || _task.Status == TaskStatus.RanToCompletion)) // a faulted task reports its own error + Fail(ErrorKind.Timeout, $"System.TimeoutException: The snippet ran longer than its time budget of {_budgetMs} ms (timeoutMs): " + + $"it finished only after {_running.ElapsedMilliseconds} ms, blocked in a call the guards cannot interrupt (a lock, a host or engine call)."); if (_errorKind == ErrorKind.None) value = _runner._options.ValueConverter(Result()); } diff --git a/python/code_exec/runner.py b/python/code_exec/runner.py index 62d9344..d1a8b93 100644 --- a/python/code_exec/runner.py +++ b/python/code_exec/runner.py @@ -213,6 +213,10 @@ def _execute(result, logs, code, include, include_root, imports, include_outside value = eval(tail, scope) if tail is not None else None _make_room(saved) watchdog.stop() + if watchdog.fired: # it outran its budget in a call nothing could interrupt (a C call, a lock): no success + raise threads.stop_type( + "The snippet ran longer than its time budget of %d ms (timeoutMs): it finished only after %d ms, blocked " + "in a call that cannot be interrupted (a C call, a lock, an engine call)." % (timeout_ms, _ms(executing))) threads.disarm() result["value"] = _convert(value) result["success"] = True @@ -274,6 +278,8 @@ class SnippetTimeout(BaseException): budget_ms = 0 def __str__(self): + if self.args: + return str(self.args[0]) return "The snippet ran longer than its time budget of %d ms (timeoutMs) and was stopped." % type(self).budget_ms @@ -324,6 +330,11 @@ def _watch(self, stop): finally: self._exited = True + @property + def fired(self): + """Whether the deadline passed while the snippet still ran.""" + return self._fired + def stop(self): while self._thread_id is not None: try: diff --git a/python/tests/test_code_exec.py b/python/tests/test_code_exec.py index 1094047..c9821a1 100644 --- a/python/tests/test_code_exec.py +++ b/python/tests/test_code_exec.py @@ -603,9 +603,9 @@ def test_waits_nothing_ever_ends(self): self.assert_times_out(code) def test_join_on_a_thread_blocked_in_a_wait(self): - result = self.assert_times_out("import threading\ne = threading.Event()\ndef f():\n e.wait()\n" - "t = threading.Thread(target=f, daemon=True)\nt.start()\nt.join()") - self.assertEqual([], result["logs"]) # stopped in its wait at the deadline, gone by the end + # The thread is stopped in its wait at the deadline (it may still be ending as the run ends: no log assertion). + self.assert_times_out("import threading\ne = threading.Event()\ndef f():\n e.wait()\n" + "t = threading.Thread(target=f, daemon=True)\nt.start()\nt.join()") def test_result_of_work_blocked_in_a_wait(self): self.assert_times_out("import threading\nfrom concurrent.futures import ThreadPoolExecutor\n" @@ -637,6 +637,16 @@ def test_a_notify_while_a_slice_ends_is_not_lost(self): self.assertTrue(result["success"], result) self.assertEqual("[[7, true], [false, false]]", result["value"]) + def test_a_run_that_outran_its_budget_in_a_c_call_is_a_timeout(self): + # A raw lock acquire is C, never sliced: it returns after the deadline; the snippet swallows the stop and returns + # before it is raised again, and the run is still no success. + code = ("import _thread\nlock = _thread.allocate_lock()\nlock.acquire()\ntry:\n lock.acquire(True, 0.6)\n" + "except BaseException:\n pass\n1") + result = code_exec.run(code, timeout_ms=200) + self.assertEqual("Timeout", result["errorKind"], result) + self.assertTrue(result["error"].startswith("SnippetTimeout: The snippet ran longer than its time budget of 200 ms " + "(timeoutMs): it finished only after"), result["error"]) + def test_the_host_converter_is_not_budgeted(self): # The snippet finished in time: a converter that waits past the deadline does not turn it into a Timeout. code_exec.set_value_converter(lambda value: (threading.Event().wait(0.4), value)[1]) diff --git a/spec/contract.md b/spec/contract.md index e90a992..75dd5d7 100644 --- a/spec/contract.md +++ b/spec/contract.md @@ -165,7 +165,10 @@ the host turns them off (`CodeRunnerOptions.ExecutionGuards`): property pattern path (`x is { Task.Result: 1 }`), through `base.`, a method group or a delegate; .NET 9+ `Task.WaitAll` given a span; `lock` / `Monitor.Enter` and other waits (`BlockingCollection.Take`, `Barrier`, `SpinWait`, `Process.WaitForExit`, I/O); and every wait inside host, engine or framework code the snippet calls (an - engine call that never returns). A native thread cannot be stopped at all. + engine call that never returns). A native thread cannot be stopped at all. A synchronous run that returns only after + its budget (blocked in such a call) is no success either: `Timeout`, `System.TimeoutException: The snippet ran longer + than its time budget of N ms (timeoutMs): it finished only after M ms, blocked in a call the guards cannot interrupt + (a lock, a host or engine call).`, and its value is dropped. - **Once stopped**, every further check throws again, and the agent's `catch` clauses that could catch the stop (no type, `Exception`, `SystemException` or the stop's own type) get a filter that lets it pass, so it reaches the runtime; `finally` blocks still run. That filter makes a few invalid programs compile (a later `catch` that an earlier @@ -237,7 +240,8 @@ the host turns them off (`CodeRunnerOptions.ExecutionGuards`): host that replaced one meanwhile keeps the stand-in in its chain, which then waits as Python does outside runs). Not covered (stopped only when the call returns): `Lock.acquire` / `RLock.acquire` (C), `queue.SimpleQueue.get`, `select` / sockets / `subprocess` waits, `asyncio`'s event loop waiting for I/O, waits inside C extensions or the - engine, and `from time import sleep` bound before the run. + engine, and `from time import sleep` bound before the run. A run whose deadline passed while it still ran (blocked + in such a call) is a `Timeout` even when it then returns without the stop reaching it. - **Threads.** Threads the snippet starts (`threading.Thread.start`, `threading.Timer` included, called from the snippet's thread or from one of those threads while it runs) may not outlive it. When it ends (finished or timed out), a `threading.Timer` that has not fired is cancelled, and every other such thread still alive gets