Fix the kind: the package's test-replaced functions (clock, waitBound, subscribe) are held atomically
A test replaced a package function while a goroutine of an earlier subtest still called it (a data race under -race, a timing flake otherwise). Each is a swappable value now, so no replacing races a call; -race -count=5 of the views tests is green. Builds: design 47 G §419 (the tests of this package run in one test function)
This commit is contained in:
+3
-3
@@ -106,12 +106,12 @@ type Status struct {
|
||||
const WaitsBound = 5 * time.Minute
|
||||
|
||||
// waitBound waits d within ctx: true when d passed, false when ctx ended first. A variable so a test ends it at once.
|
||||
var waitBound = func(ctx context.Context, d time.Duration) bool {
|
||||
var waitBound = swappable(func(ctx context.Context, d time.Duration) bool {
|
||||
b, cancel := context.WithTimeout(ctx, d)
|
||||
defer cancel()
|
||||
<-b.Done()
|
||||
return ctx.Err() == nil
|
||||
}
|
||||
})
|
||||
|
||||
// NoneYet is what a seat's revision verb answers for a branch with no revision yet (design 47 G §479, §495): that view
|
||||
// is ready and empty, and nothing waits for it.
|
||||
@@ -134,7 +134,7 @@ func (f *Follower) WaitFor(ctx context.Context, view string) {
|
||||
f.logf("views: %s waits for %s's view %s, from %s", f.Tool, f.Owner, view, s.WaitingSince.Format(time.RFC3339))
|
||||
f.say(s)
|
||||
go func() {
|
||||
if !waitBound(ctx, WaitsBound) {
|
||||
if !waitBound.get()(ctx, WaitsBound) {
|
||||
return
|
||||
}
|
||||
f.mu.Lock()
|
||||
|
||||
+2
-2
@@ -59,7 +59,7 @@ type Reader struct {
|
||||
}
|
||||
|
||||
// subscribe is how the reader asks the runtime for an event pattern; a variable so a test hears without a runtime.
|
||||
var subscribe = stdio.Subscribe
|
||||
var subscribe = swappable(stdio.Subscribe)
|
||||
|
||||
// adoption is how an adopted follower's events are heard: the seat's event, by its local name, and the client's handler.
|
||||
type adoption struct {
|
||||
@@ -315,7 +315,7 @@ func (r *Reader) begin(name string) {
|
||||
handle := r.handler(ctx, pattern)
|
||||
go func() {
|
||||
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
||||
err := subscribe(pattern, handle)
|
||||
err := subscribe.get()(pattern, handle)
|
||||
if err == nil {
|
||||
break
|
||||
}
|
||||
|
||||
+17
-3
@@ -11,6 +11,7 @@ import (
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -46,7 +47,7 @@ const PageBytes = 768 * 1024
|
||||
var PageBudget = 10 * time.Second
|
||||
|
||||
// clock is the page budget's time, replaced by a test.
|
||||
var clock = time.Now
|
||||
var clock = swappable(time.Now)
|
||||
|
||||
// Bound is how long a reader waits for its first view before it raises the condition naming the owner and the verb
|
||||
// (ADR 0615 §6): longer than an owner's start, so a normal start, readers and owners coming up together, tells nobody.
|
||||
@@ -265,12 +266,12 @@ func FirstView(owner, state string, values, marks Source, after string, now time
|
||||
}
|
||||
slices.Sort(keys)
|
||||
size := 0
|
||||
started := clock()
|
||||
started := clock.get()()
|
||||
for _, k := range keys {
|
||||
if after != "" && k <= after {
|
||||
continue
|
||||
}
|
||||
if size > 0 && clock().Sub(started) >= PageBudget {
|
||||
if size > 0 && clock.get()().Sub(started) >= PageBudget {
|
||||
a.Next = lastKey(a.Values)
|
||||
break
|
||||
}
|
||||
@@ -326,3 +327,16 @@ func decode(data []byte, into any) error {
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// swapped is a function a test replaces while a goroutine of an earlier test may still call it: held atomically, so the
|
||||
// replacing and the calling never race.
|
||||
type swapped[T any] struct{ p atomic.Pointer[T] }
|
||||
|
||||
func swappable[T any](v T) *swapped[T] {
|
||||
s := &swapped[T]{}
|
||||
s.p.Store(&v)
|
||||
return s
|
||||
}
|
||||
|
||||
func (s *swapped[T]) get() T { return *s.p.Load() }
|
||||
func (s *swapped[T]) set(v T) { s.p.Store(&v) }
|
||||
|
||||
+13
-13
@@ -101,8 +101,8 @@ func subViews(t *testing.T) {
|
||||
})
|
||||
t.Run("a first view from a slow source stops at the page budget, with one value at least, and pages on to every key", func(t *testing.T) {
|
||||
at := time.Unix(0, 0)
|
||||
clock = func() time.Time { return at }
|
||||
defer func() { clock = time.Now }()
|
||||
clock.set(func() time.Time { return at })
|
||||
defer clock.set(time.Now)
|
||||
slow := mapSource{}
|
||||
for i := range 5 {
|
||||
slow[fmt.Sprintf("k%d", i)] = Entry{Value: json.RawMessage(`{"n":1}`), Revision: uint64(i + 1)}
|
||||
@@ -306,9 +306,9 @@ func subViews(t *testing.T) {
|
||||
// when; past the bound the part's view-waits is raised once, and holding the view clears it.
|
||||
t.Run("a part that waits for a view says so, raises view-waits past its bound, and holding it clears that", func(t *testing.T) {
|
||||
ended := make(chan bool)
|
||||
kept := waitBound
|
||||
waitBound = func(context.Context, time.Duration) bool { return <-ended }
|
||||
defer func() { waitBound = kept }()
|
||||
kept := waitBound.get()
|
||||
waitBound.set(func(context.Context, time.Duration) bool { return <-ended })
|
||||
defer waitBound.set(kept)
|
||||
var raised, cleared, said []Status
|
||||
var mu sync.Mutex
|
||||
raisedCh := make(chan struct{}, 4) // told after the part's raise, so the test reads it only once it is made
|
||||
@@ -686,9 +686,9 @@ func subViews(t *testing.T) {
|
||||
|
||||
t.Run("a follower a seat's client made is adopted: the reader's findings, reads and tools hold it, and its seat event goes to the client", func(t *testing.T) {
|
||||
ended := make(chan bool, 1)
|
||||
kept := waitBound
|
||||
waitBound = func(context.Context, time.Duration) bool { return <-ended }
|
||||
defer func() { waitBound = kept }()
|
||||
kept := waitBound.get()
|
||||
waitBound.set(func(context.Context, time.Duration) bool { return <-ended })
|
||||
defer waitBound.set(kept)
|
||||
fs := &signalFindings{put: make(chan seenFinding, 4)}
|
||||
r := &Reader{Module: "reader", Findings: fs, Now: now}
|
||||
var own []string
|
||||
@@ -792,17 +792,17 @@ func subViews(t *testing.T) {
|
||||
// dropped one hears nothing more, leaves `<module>_first_views`, and the findings it raised are cleared.
|
||||
t.Run("a view adopted after the start takes its first view and follows, and a dropped one leaves with its findings cleared", func(t *testing.T) {
|
||||
ended := make(chan bool, 2)
|
||||
keptBound, keptSubscribe := waitBound, subscribe
|
||||
waitBound = func(context.Context, time.Duration) bool { return <-ended }
|
||||
keptBound, keptSubscribe := waitBound.get(), subscribe.get()
|
||||
waitBound.set(func(context.Context, time.Duration) bool { return <-ended })
|
||||
var hmu sync.Mutex
|
||||
handlers := map[string]func(stdio.Envelope) error{}
|
||||
subscribe = func(p string, h func(stdio.Envelope) error) error {
|
||||
subscribe.set(func(p string, h func(stdio.Envelope) error) error {
|
||||
hmu.Lock()
|
||||
defer hmu.Unlock()
|
||||
handlers[p] = h
|
||||
return nil
|
||||
}
|
||||
defer func() { waitBound, subscribe = keptBound, keptSubscribe }()
|
||||
})
|
||||
defer func() { waitBound.set(keptBound); subscribe.set(keptSubscribe) }()
|
||||
ready := func(f *Follower) {
|
||||
t.Helper()
|
||||
select {
|
||||
|
||||
Reference in New Issue
Block a user