diff --git a/go/views/follow.go b/go/views/follow.go index e0a4bf99..9aa25d8d 100644 --- a/go/views/follow.go +++ b/go/views/follow.go @@ -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() diff --git a/go/views/reader.go b/go/views/reader.go index dc5b5f0b..1cdf6065 100644 --- a/go/views/reader.go +++ b/go/views/reader.go @@ -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 } diff --git a/go/views/views.go b/go/views/views.go index d6989047..379f7a6e 100644 --- a/go/views/views.go +++ b/go/views/views.go @@ -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) } diff --git a/go/views/views_test.go b/go/views/views_test.go index a93186a7..9ed21cf1 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -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 `_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 {