diff --git a/BREAKING-CHANGES.md b/BREAKING-CHANGES.md index b6be1bfaa..3f7909bce 100644 --- a/BREAKING-CHANGES.md +++ b/BREAKING-CHANGES.md @@ -59,6 +59,7 @@ read first. | **silent** | `abs(x) = c` for a negative `c` | a set of non-solutions | the empty set | | loud | implicit `List` to `Entity` | made a `FiniteSet`, and made three `params` overloads uncallable | removed | | **silent** | a `MathS.Settings` scope across an `await`, or inside a task | lost, or somebody else's | follows the call | +| **silent** | a `RewriteRecording` across an `await`, or work started under it | lost, or somebody else's | follows the call | --- @@ -205,6 +206,48 @@ null on every access, and that is dearer than the async-local lookup that replac Opening a scope got much cheaper because it no longer mints a `Guid` to identify itself, though it allocates more, since assigning an `AsyncLocal` copies the flow's value map. Reads outnumber scope openings by orders of magnitude in any real workload. +### A rewrite recording belongs to the call, not to the thread + +`RewriteRecording` held its ambient scope in a `[ThreadStatic]` field, and documented the +consequence rather than fixing it: + +> **A synchronous scope, and it has to be.** Do not `await` inside one. + +It no longer has to be. The scope is an `AsyncLocal`, as `MathS.Settings` and the +cancellation token in `MathS.Multithreading` are, so it survives an `await` and work +started under it reports to it wherever it runs. + +| | was | is | +|---|---|---| +| a recording across an `await` | lost, and the thread could collect a stranger's rewrites | kept | +| work started under a recording, on another thread | **not** collected | collected | +| two flows each with their own recording | separate | separate | +| a recording opened inside a task, seen after it ends | no | no | + +**What breaks.** The second row. A recording no longer stops at the thread boundary: + +```csharp +using var recording = RewriteRecording.Start(); +var t = new Thread(() => expr.Simplify()); +t.Start(); t.Join(); +recording.Steps; // used to be empty of that work; now contains it +``` + +That is what the code says and what the scope is for, but a caller who fanned out inside a +recording and expected only their own thread's rewrites will now see everything. Open the +recording inside the callback to keep the old behaviour. + +**`Steps` is now a snapshot.** It was a live view of the underlying list; because the store +has to tolerate concurrent writers it is a `ConcurrentQueue`, and `Steps` copies out of it. +Reading it after disposing — the documented use — is unchanged. Holding the returned list +across further recording and expecting it to grow no longer works. + +**Order across parallel work is not defined.** Steps from one flow keep the order they fired +in; two flows recording into the same recording interleave however they happen to run. The +single-threaded case, which is what `Simplify` is, is unaffected. + +Being off is still free: no recording open still costs one ambient read per rule set and +allocates nothing, which `RewriteAllocationTest` continues to hold to. ### `Minusf`'s two operands exchanged names diff --git a/Sources/AngouriMath/Core/Transformations/RewriteRecording.cs b/Sources/AngouriMath/Core/Transformations/RewriteRecording.cs index 60c7e34d3..58bd9a528 100644 --- a/Sources/AngouriMath/Core/Transformations/RewriteRecording.cs +++ b/Sources/AngouriMath/Core/Transformations/RewriteRecording.cs @@ -6,6 +6,8 @@ // using System; +using System.Collections.Concurrent; +using System.Threading; namespace AngouriMath.Core.Transformations { @@ -16,26 +18,23 @@ namespace AngouriMath.Core.Transformations /// /// /// Off unless asked for, and off is free: with no recording open, applying a rule set - /// costs one thread-static read more than it did before — per rule set, not per node — - /// and allocates nothing. That is the condition + /// costs one ambient read more than it did before — per rule set, not per node — and + /// allocates nothing. That is the condition /// #746 puts on /// every layer above the tree, and it is why this is a scope rather than a setting that /// something might leave on. /// /// - /// Per thread, like : a recording opened on one thread does - /// not see rewrites on another, so a parallel caller records its own work and nobody - /// else's. + /// Per flow, like : the recording belongs to the call rather + /// than to the thread running it. It survives an , and work + /// started under it — including on another thread — reports to it. A recording opened + /// inside a task is invisible to that task's siblings and to whatever started it. /// /// - /// A synchronous scope, and it has to be. Do not inside - /// one. The recording follows the thread rather than the call, so yielding lets whatever - /// else that thread picks up be recorded as if it were yours, and the continuation may - /// come back on a different thread than the one holding it. Closing is written to - /// survive both — a recording closed elsewhere leaves the opening thread pointing at - /// something that ignores what it is handed rather than at a list that keeps growing — - /// but what gets collected in between is not something this can make meaningful. Record - /// around synchronous work, and await outside the scope. + /// Order is not guaranteed once work is parallel. Steps from one flow keep the + /// order they fired in, but two flows recording into the same open recording interleave + /// however they happen to run. The single-threaded case — which is what + /// is — is unaffected. /// /// /// What this is not. It records rewrites — which is what @@ -61,17 +60,30 @@ namespace AngouriMath.Core.Transformations /// public sealed class RewriteRecording : IDisposable { - [ThreadStatic] - private static RewriteRecording? current; + /// + /// Which recording the current call reports to. An rather + /// than [ThreadStatic], so the scope follows the call: held per thread, a + /// recording was lost at the first , and a pool thread carried + /// a stale one into whoever borrowed it next. + /// + [ConcurrentField] + private static readonly AsyncLocal current = new(); private readonly RewriteRecording? enclosing; - private readonly List steps = new(); - private bool closed; + + /// + /// Concurrent because the pointer above flows into child tasks, so two of them can + /// report to one recording at once. A here would be a torn + /// write rather than a merged list. + /// + private readonly ConcurrentQueue steps = new(); + + private volatile bool closed; private RewriteRecording(RewriteRecording? enclosing) => this.enclosing = enclosing; /// - /// Opens a recording on this thread. Dispose it to close it — the value is meant to + /// Opens a recording for this call. Dispose it to close it — the value is meant to /// be held in a , as values are. /// /// @@ -79,12 +91,22 @@ public sealed class RewriteRecording : IDisposable /// is disposed, so a caller who records a subcomputation does not silently add its /// steps to somebody else's list. /// - public static RewriteRecording Start() => current = new RewriteRecording(current); + public static RewriteRecording Start() + { + var recording = new RewriteRecording(current.Value); + current.Value = recording; + return recording; + } /// /// The rewrites that fired while this recording was open, in the order they fired. /// - public IReadOnlyList Steps => steps; + /// + /// A snapshot taken when you ask, not a live view, since the underlying store has to + /// tolerate concurrent writers. Reading it after disposing — which is the usual way — + /// gives the complete list either way. + /// + public IReadOnlyList Steps => steps.ToArray(); /// Closes the recording. stays readable afterwards. public void Dispose() @@ -92,26 +114,24 @@ public void Dispose() if (closed) return; closed = true; - // Only this thread's chain is ours to put back. Disposing on a thread other than - // the one that opened it -- which is what awaiting inside a recording leads to -- - // would otherwise clear whatever that thread was recording into, and leave the - // opening thread pointing at a closed recording. Add ignores that case, so the - // worst a stray reference can do is nothing. - if (ReferenceEquals(current, this)) - current = enclosing; + // Only this flow's chain is ours to put back. Disposing from a flow that did not + // open it would otherwise clear whatever that flow was recording into. Add + // ignores a closed recording, so the worst a stray reference can do is nothing. + if (ReferenceEquals(current.Value, this)) + current.Value = enclosing; } /// /// The recording to report to, or where nobody is listening — /// which is the case this has to stay free for. /// - internal static RewriteRecording? Current => current; + internal static RewriteRecording? Current => current.Value; internal void Add(RewriteRuleSet ruleSet, Entity before, Entity after) { if (closed) return; - steps.Add(new RewriteStep(ruleSet, before, after)); + steps.Enqueue(new RewriteStep(ruleSet, before, after)); } } } diff --git a/Sources/Tests/UnitTests/Core/Transformations/RewriteRecordingTest.cs b/Sources/Tests/UnitTests/Core/Transformations/RewriteRecordingTest.cs index ae63015e6..395c77125 100644 --- a/Sources/Tests/UnitTests/Core/Transformations/RewriteRecordingTest.cs +++ b/Sources/Tests/UnitTests/Core/Transformations/RewriteRecordingTest.cs @@ -8,6 +8,7 @@ using System.Collections.Generic; using System.Linq; using System.Threading; +using System.Threading.Tasks; using AngouriMath; using AngouriMath.Core.Transformations; using Xunit; @@ -139,21 +140,77 @@ public void AnInnerRecordingDoesNotFeedTheOuterOne() } [Fact] - public void ARecordingOnOneThreadDoesNotSeeAnother() + public void WorkStartedUnderARecordingIsCollectedWhereverItRuns() { + // The recording belongs to the call, so work begun under it reports to it even + // on another thread. Held per thread this collected nothing, because the new + // thread had never seen the recording. using var recording = RewriteRecording.Start(); Parse("a / (b / c)").Simplify(); var mine = recording.Steps.Count; + Assert.NotEqual(0, mine); - // A thread joined synchronously rather than an awaited task. Awaiting would - // yield this thread while the recording is still open on it, and this thread is - // exactly where the test runner is free to start something else -- which the - // recording would then collect, because it follows the thread and not the call. var other = new Thread(() => Parse("sin(x) / tan(x) + u / (v / w)").Simplify()); other.Start(); other.Join(); - Assert.Equal(mine, recording.Steps.Count); + Assert.True(recording.Steps.Count > mine); + } + + [Fact] + public async Task ARecordingSurvivesAnAwait() + { + using var recording = RewriteRecording.Start(); + await Task.Delay(20).ConfigureAwait(false); + Parse("a / (b / c)").Simplify(); + Assert.NotEmpty(recording.Steps); + } + + [Fact] + public async Task SiblingRecordingsDoNotSeeEachOther() + { + // The isolation that the per-thread field did give and that this must keep. The + // barrier makes both recordings open before either does any work. + using var barrier = new Barrier(2); + async Task<(int mine, IReadOnlyList steps)> Branch(string expr) + { + await Task.Yield(); + using var recording = RewriteRecording.Start(); + barrier.SignalAndWait(); + Parse(expr).Simplify(); + await Task.Delay(20).ConfigureAwait(false); + return (recording.Steps.Count, recording.Steps); + } + + var left = Branch("a / (b / c)"); + var right = Branch("sin(x) / tan(x)"); + var results = await Task.WhenAll(left, right); + + Assert.All(results, r => Assert.NotEqual(0, r.mine)); + // Neither collected the other's work: together they would be the union, and each + // is strictly smaller than that. + var combined = results.Sum(r => r.mine); + Assert.All(results, r => Assert.True(r.mine < combined)); + } + + [Fact] + public async Task ARecordingOpenedInsideATaskDoesNotEscapeIt() + { + RewriteRecording? inner = null; + await Task.Run(() => + { + // Left open on purpose: what is under test is that the pointer does not leak + // out of the task, not that Dispose puts it back. + inner = RewriteRecording.Start(); + Parse("a / (b / c)").Simplify(); + }); + + var recording = Assert.IsType(inner); + var whenTheTaskEnded = recording.Steps.Count; + Assert.NotEqual(0, whenTheTaskEnded); + + Parse("sin(x) / tan(x) + u / (v / w)").Simplify(); + Assert.Equal(whenTheTaskEnded, recording.Steps.Count); } [Fact]