Skip to content

Commit bb75230

Browse files
Hold a rewrite recording per flow instead of per thread (#863)
Closes #859. RewriteRecording kept 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 now is and as the cancellation token in MathS.Multithreading already was. The trap #857 flagged applies here and is why this is more than a field swap. An AsyncLocal flows the reference, so once the pointer reaches child tasks two of them can report to one recording at once, and the steps were accumulating into a List. That is a torn write, not a merged list. The store is a ConcurrentQueue, `closed` is volatile since it is now read from flows other than the one that set it, and Steps copies out rather than handing back a live view. One existing test encoded the old semantics and is rewritten rather than deleted: ARecordingOnOneThreadDoesNotSeeAnother started a thread inside an open recording and asserted its work was not collected. ExecutionContext flows to a manually started thread, so that work is now collected -- which is the point of the change, not a regression of it. It becomes WorkStartedUnderARecordingIsCollectedWhereverItRuns, and the isolation it was really reaching for is covered by SiblingRecordingsDoNotSeeEachOther. RewriteAllocationTest still passes untouched, so being off is still free: one ambient read per rule set, nothing allocated, which is what #746 asks of every layer above the tree. Verified: 6064 C# tests and 130 F# tests pass. Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
1 parent c6b03b1 commit bb75230

3 files changed

Lines changed: 155 additions & 35 deletions

File tree

‎BREAKING-CHANGES.md‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,7 @@ read first.
6060
| loud | implicit `List<Entity>` to `Entity` | made a `FiniteSet`, and made three `params` overloads uncallable | removed |
6161
| loud | implicit `Entity[]` to `Entity` | made a `FiniteSet`, discarding order and repeats | removed |
6262
| **silent** | a `MathS.Settings` scope across an `await`, or inside a task | lost, or somebody else's | follows the call |
63+
| **silent** | a `RewriteRecording` across an `await`, or work started under it | lost, or somebody else's | follows the call |
6364

6465
---
6566

@@ -218,6 +219,48 @@ null on every access, and that is dearer than the async-local lookup that replac
218219
Opening a scope got much cheaper because it no longer mints a `Guid` to identify itself,
219220
though it allocates more, since assigning an `AsyncLocal` copies the flow's value map. Reads
220221
outnumber scope openings by orders of magnitude in any real workload.
222+
### A rewrite recording belongs to the call, not to the thread
223+
224+
`RewriteRecording` held its ambient scope in a `[ThreadStatic]` field, and documented the
225+
consequence rather than fixing it:
226+
227+
> **A synchronous scope, and it has to be.** Do not `await` inside one.
228+
229+
It no longer has to be. The scope is an `AsyncLocal`, as `MathS.Settings` and the
230+
cancellation token in `MathS.Multithreading` are, so it survives an `await` and work
231+
started under it reports to it wherever it runs.
232+
233+
| | was | is |
234+
|---|---|---|
235+
| a recording across an `await` | lost, and the thread could collect a stranger's rewrites | kept |
236+
| work started under a recording, on another thread | **not** collected | collected |
237+
| two flows each with their own recording | separate | separate |
238+
| a recording opened inside a task, seen after it ends | no | no |
239+
240+
**What breaks.** The second row. A recording no longer stops at the thread boundary:
241+
242+
```csharp
243+
using var recording = RewriteRecording.Start();
244+
var t = new Thread(() => expr.Simplify());
245+
t.Start(); t.Join();
246+
recording.Steps; // used to be empty of that work; now contains it
247+
```
248+
249+
That is what the code says and what the scope is for, but a caller who fanned out inside a
250+
recording and expected only their own thread's rewrites will now see everything. Open the
251+
recording inside the callback to keep the old behaviour.
252+
253+
**`Steps` is now a snapshot.** It was a live view of the underlying list; because the store
254+
has to tolerate concurrent writers it is a `ConcurrentQueue`, and `Steps` copies out of it.
255+
Reading it after disposing — the documented use — is unchanged. Holding the returned list
256+
across further recording and expecting it to grow no longer works.
257+
258+
**Order across parallel work is not defined.** Steps from one flow keep the order they fired
259+
in; two flows recording into the same recording interleave however they happen to run. The
260+
single-threaded case, which is what `Simplify` is, is unaffected.
261+
262+
Being off is still free: no recording open still costs one ambient read per rule set and
263+
allocates nothing, which `RewriteAllocationTest` continues to hold to.
221264

222265
### `Minusf`'s two operands exchanged names
223266

‎Sources/AngouriMath/Core/Transformations/RewriteRecording.cs‎

Lines changed: 49 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,8 @@
66
//
77

88
using System;
9+
using System.Collections.Concurrent;
10+
using System.Threading;
911

1012
namespace AngouriMath.Core.Transformations
1113
{
@@ -16,26 +18,23 @@ namespace AngouriMath.Core.Transformations
1618
/// <remarks>
1719
/// <para>
1820
/// Off unless asked for, and off is free: with no recording open, applying a rule set
19-
/// costs one thread-static read more than it did before — per rule set, not per node —
20-
/// and allocates nothing. That is the condition
21+
/// costs one ambient read more than it did before — per rule set, not per node — and
22+
/// allocates nothing. That is the condition
2123
/// <a href="https://github.com/asc-community/AngouriMath/issues/746">#746</a> puts on
2224
/// every layer above the tree, and it is why this is a scope rather than a setting that
2325
/// something might leave on.
2426
/// </para>
2527
/// <para>
26-
/// Per thread, like <see cref="MathS.Settings"/>: a recording opened on one thread does
27-
/// not see rewrites on another, so a parallel caller records its own work and nobody
28-
/// else's.
28+
/// Per flow, like <see cref="MathS.Settings"/>: the recording belongs to the call rather
29+
/// than to the thread running it. It survives an <see langword="await"/>, and work
30+
/// started under it — including on another thread — reports to it. A recording opened
31+
/// inside a task is invisible to that task's siblings and to whatever started it.
2932
/// </para>
3033
/// <para>
31-
/// <b>A synchronous scope, and it has to be.</b> Do not <see langword="await"/> inside
32-
/// one. The recording follows the thread rather than the call, so yielding lets whatever
33-
/// else that thread picks up be recorded as if it were yours, and the continuation may
34-
/// come back on a different thread than the one holding it. Closing is written to
35-
/// survive both — a recording closed elsewhere leaves the opening thread pointing at
36-
/// something that ignores what it is handed rather than at a list that keeps growing —
37-
/// but what gets collected in between is not something this can make meaningful. Record
38-
/// around synchronous work, and await outside the scope.
34+
/// <b>Order is not guaranteed once work is parallel.</b> Steps from one flow keep the
35+
/// order they fired in, but two flows recording into the same open recording interleave
36+
/// however they happen to run. The single-threaded case — which is what
37+
/// <see cref="Entity.Simplify(int)"/> is — is unaffected.
3938
/// </para>
4039
/// <para>
4140
/// <b>What this is not.</b> It records rewrites — which is what
@@ -61,57 +60,78 @@ namespace AngouriMath.Core.Transformations
6160
/// </example>
6261
public sealed class RewriteRecording : IDisposable
6362
{
64-
[ThreadStatic]
65-
private static RewriteRecording? current;
63+
/// <summary>
64+
/// Which recording the current call reports to. An <see cref="AsyncLocal{T}"/> rather
65+
/// than <c>[ThreadStatic]</c>, so the scope follows the call: held per thread, a
66+
/// recording was lost at the first <see langword="await"/>, and a pool thread carried
67+
/// a stale one into whoever borrowed it next.
68+
/// </summary>
69+
[ConcurrentField]
70+
private static readonly AsyncLocal<RewriteRecording?> current = new();
6671

6772
private readonly RewriteRecording? enclosing;
68-
private readonly List<RewriteStep> steps = new();
69-
private bool closed;
73+
74+
/// <summary>
75+
/// Concurrent because the pointer above flows into child tasks, so two of them can
76+
/// report to one recording at once. A <see cref="List{T}"/> here would be a torn
77+
/// write rather than a merged list.
78+
/// </summary>
79+
private readonly ConcurrentQueue<RewriteStep> steps = new();
80+
81+
private volatile bool closed;
7082

7183
private RewriteRecording(RewriteRecording? enclosing) => this.enclosing = enclosing;
7284

7385
/// <summary>
74-
/// Opens a recording on this thread. Dispose it to close it — the value is meant to
86+
/// Opens a recording for this call. Dispose it to close it — the value is meant to
7587
/// be held in a <see langword="using"/>, as <see cref="MathS.Settings"/> values are.
7688
/// </summary>
7789
/// <remarks>
7890
/// Recordings nest: opening one inside another hides the outer one until the inner
7991
/// is disposed, so a caller who records a subcomputation does not silently add its
8092
/// steps to somebody else's list.
8193
/// </remarks>
82-
public static RewriteRecording Start() => current = new RewriteRecording(current);
94+
public static RewriteRecording Start()
95+
{
96+
var recording = new RewriteRecording(current.Value);
97+
current.Value = recording;
98+
return recording;
99+
}
83100

84101
/// <summary>
85102
/// The rewrites that fired while this recording was open, in the order they fired.
86103
/// </summary>
87-
public IReadOnlyList<RewriteStep> Steps => steps;
104+
/// <remarks>
105+
/// A snapshot taken when you ask, not a live view, since the underlying store has to
106+
/// tolerate concurrent writers. Reading it after disposing — which is the usual way —
107+
/// gives the complete list either way.
108+
/// </remarks>
109+
public IReadOnlyList<RewriteStep> Steps => steps.ToArray();
88110

89111
/// <summary>Closes the recording. <see cref="Steps"/> stays readable afterwards.</summary>
90112
public void Dispose()
91113
{
92114
if (closed)
93115
return;
94116
closed = true;
95-
// Only this thread's chain is ours to put back. Disposing on a thread other than
96-
// the one that opened it -- which is what awaiting inside a recording leads to --
97-
// would otherwise clear whatever that thread was recording into, and leave the
98-
// opening thread pointing at a closed recording. Add ignores that case, so the
99-
// worst a stray reference can do is nothing.
100-
if (ReferenceEquals(current, this))
101-
current = enclosing;
117+
// Only this flow's chain is ours to put back. Disposing from a flow that did not
118+
// open it would otherwise clear whatever that flow was recording into. Add
119+
// ignores a closed recording, so the worst a stray reference can do is nothing.
120+
if (ReferenceEquals(current.Value, this))
121+
current.Value = enclosing;
102122
}
103123

104124
/// <summary>
105125
/// The recording to report to, or <see langword="null"/> where nobody is listening —
106126
/// which is the case this has to stay free for.
107127
/// </summary>
108-
internal static RewriteRecording? Current => current;
128+
internal static RewriteRecording? Current => current.Value;
109129

110130
internal void Add(RewriteRuleSet ruleSet, Entity before, Entity after)
111131
{
112132
if (closed)
113133
return;
114-
steps.Add(new RewriteStep(ruleSet, before, after));
134+
steps.Enqueue(new RewriteStep(ruleSet, before, after));
115135
}
116136
}
117137
}

‎Sources/Tests/UnitTests/Core/Transformations/RewriteRecordingTest.cs‎

Lines changed: 63 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
using System.Collections.Generic;
99
using System.Linq;
1010
using System.Threading;
11+
using System.Threading.Tasks;
1112
using AngouriMath;
1213
using AngouriMath.Core.Transformations;
1314
using Xunit;
@@ -139,21 +140,77 @@ public void AnInnerRecordingDoesNotFeedTheOuterOne()
139140
}
140141

141142
[Fact]
142-
public void ARecordingOnOneThreadDoesNotSeeAnother()
143+
public void WorkStartedUnderARecordingIsCollectedWhereverItRuns()
143144
{
145+
// The recording belongs to the call, so work begun under it reports to it even
146+
// on another thread. Held per thread this collected nothing, because the new
147+
// thread had never seen the recording.
144148
using var recording = RewriteRecording.Start();
145149
Parse("a / (b / c)").Simplify();
146150
var mine = recording.Steps.Count;
151+
Assert.NotEqual(0, mine);
147152

148-
// A thread joined synchronously rather than an awaited task. Awaiting would
149-
// yield this thread while the recording is still open on it, and this thread is
150-
// exactly where the test runner is free to start something else -- which the
151-
// recording would then collect, because it follows the thread and not the call.
152153
var other = new Thread(() => Parse("sin(x) / tan(x) + u / (v / w)").Simplify());
153154
other.Start();
154155
other.Join();
155156

156-
Assert.Equal(mine, recording.Steps.Count);
157+
Assert.True(recording.Steps.Count > mine);
158+
}
159+
160+
[Fact]
161+
public async Task ARecordingSurvivesAnAwait()
162+
{
163+
using var recording = RewriteRecording.Start();
164+
await Task.Delay(20).ConfigureAwait(false);
165+
Parse("a / (b / c)").Simplify();
166+
Assert.NotEmpty(recording.Steps);
167+
}
168+
169+
[Fact]
170+
public async Task SiblingRecordingsDoNotSeeEachOther()
171+
{
172+
// The isolation that the per-thread field did give and that this must keep. The
173+
// barrier makes both recordings open before either does any work.
174+
using var barrier = new Barrier(2);
175+
async Task<(int mine, IReadOnlyList<RewriteStep> steps)> Branch(string expr)
176+
{
177+
await Task.Yield();
178+
using var recording = RewriteRecording.Start();
179+
barrier.SignalAndWait();
180+
Parse(expr).Simplify();
181+
await Task.Delay(20).ConfigureAwait(false);
182+
return (recording.Steps.Count, recording.Steps);
183+
}
184+
185+
var left = Branch("a / (b / c)");
186+
var right = Branch("sin(x) / tan(x)");
187+
var results = await Task.WhenAll(left, right);
188+
189+
Assert.All(results, r => Assert.NotEqual(0, r.mine));
190+
// Neither collected the other's work: together they would be the union, and each
191+
// is strictly smaller than that.
192+
var combined = results.Sum(r => r.mine);
193+
Assert.All(results, r => Assert.True(r.mine < combined));
194+
}
195+
196+
[Fact]
197+
public async Task ARecordingOpenedInsideATaskDoesNotEscapeIt()
198+
{
199+
RewriteRecording? inner = null;
200+
await Task.Run(() =>
201+
{
202+
// Left open on purpose: what is under test is that the pointer does not leak
203+
// out of the task, not that Dispose puts it back.
204+
inner = RewriteRecording.Start();
205+
Parse("a / (b / c)").Simplify();
206+
});
207+
208+
var recording = Assert.IsType<RewriteRecording>(inner);
209+
var whenTheTaskEnded = recording.Steps.Count;
210+
Assert.NotEqual(0, whenTheTaskEnded);
211+
212+
Parse("sin(x) / tan(x) + u / (v / w)").Simplify();
213+
Assert.Equal(whenTheTaskEnded, recording.Steps.Count);
157214
}
158215

159216
[Fact]

0 commit comments

Comments
 (0)