diff --git a/changes/unreleased/quiescent-behavior-scan.fixed.md b/changes/unreleased/quiescent-behavior-scan.fixed.md new file mode 100644 index 000000000..53297177d --- /dev/null +++ b/changes/unreleased/quiescent-behavior-scan.fixed.md @@ -0,0 +1 @@ +- Checking a constraint or writing a feature no longer rescans every object's behaviors when nothing has changed since the last scan found them all idle, so a batch of checks over many instantiated objects is linear again. diff --git a/internal/exec/runtime/advance.go b/internal/exec/runtime/advance.go index 5caf38d09..602dc3e97 100644 --- a/internal/exec/runtime/advance.go +++ b/internal/exec/runtime/advance.go @@ -123,10 +123,10 @@ func (ctx *Context) AdvanceUntil(duration float64, halted func() bool) (AdvanceR if !ok || next > deadline { break } - ctx.clock.now = next + ctx.setClock(next) progress.unsettle() } - ctx.clock.now = deadline + ctx.setClock(deadline) report.To = deadline return report.counting(progress, ctx.run.notes[noted:]), ctx.advanceEnded() } @@ -155,7 +155,7 @@ func (ctx *Context) advanceToNextDue(progress *dueProgress) bool { if !ok { return false } - ctx.clock.now = next + ctx.setClock(next) progress.unsettle() return true } diff --git a/internal/exec/runtime/check_run.go b/internal/exec/runtime/check_run.go index 77df2ed80..6577502b6 100644 --- a/internal/exec/runtime/check_run.go +++ b/internal/exec/runtime/check_run.go @@ -162,7 +162,7 @@ func (r *invocationRun) advanceClock() bool { if !ok || next <= r.ctx.clock.now { return false } - r.ctx.clock.now = next + r.ctx.setClock(next) r.turn = nil return true } diff --git a/internal/exec/runtime/classifier_behavior.go b/internal/exec/runtime/classifier_behavior.go index 00c0b28ce..b17df48a9 100644 --- a/internal/exec/runtime/classifier_behavior.go +++ b/internal/exec/runtime/classifier_behavior.go @@ -662,30 +662,77 @@ func (ctx *Context) startBehaviorsOf(inst *Instance) error { inst.behaviors = append(inst.behaviors, behavior) ctx.pendingBehaviors = append(ctx.pendingBehaviors, behavior) ctx.objectBehaviors = append(ctx.objectBehaviors, behavior) + ctx.workChanged() } } return ctx.runAttachedBehaviors() } +// workChanged counts a change that can leave an attached behavior holding work: +// a message posted, the clock moved, an event queued, an executor run or left. +func (ctx *Context) workChanged() { ctx.work++ } + +// quiescence is the memo a full behavior scan leaves when it finds every +// attached behavior idle: the work and write marks it holds under, and whether +// the scan read the objects' data, so writes since then invalidate it. +type quiescence struct { + at uint64 + writes uint64 + readsData bool +} + +// holds reports whether the memo still answers: taken, and nothing it depends +// on moved since, the objects' data counting only where the scan read it. +func (q quiescence) holds(ctx *Context) bool { + return q.at != 0 && q.at == ctx.work && (!q.readsData || q.writes == ctx.writes) +} + +// setClock moves the shared clock, the work due on it moving with it. +func (ctx *Context) setClock(now float64) { + ctx.clock.now = now + ctx.workChanged() +} + // holdDrivenWork marks, at an outermost start, the behaviors already holding // work: a driver put it in flight, so the start leaves it to that driver. Once // the start returns, the behaviors it attached are as their start left them. func (ctx *Context) holdDrivenWork() func() { - if ctx.behaviorRunDepth > 0 || ctx.heldBehaviors != nil { + if ctx.behaviorRunDepth > 0 || ctx.holdingDriven { return func() { /* an outer start already holds them */ } } - held := make(map[*ObjectBehavior]bool) - ctx.behaviorRunDepth++ - for _, behavior := range ctx.objectBehaviors { - if behavior.hasPendingWork() { - held[behavior] = true + var held map[*ObjectBehavior]bool + if ctx.quiescent.holds(ctx) { + // Nothing woke a behavior since a full scan found them all idle. + } else if len(ctx.objectBehaviors) == 0 { + ctx.quiescent = quiescence{at: ctx.work, writes: ctx.writes} + } else { + memo := &pendingMemo{} + saved := ctx.polling + ctx.polling = memo + ctx.behaviorRunDepth++ + for _, behavior := range ctx.objectBehaviors { + if behavior.hasPendingWork() { + if held == nil { + held = map[*ObjectBehavior]bool{} + } + held[behavior] = true + } + } + ctx.behaviorRunDepth-- + ctx.polling = saved + if saved != nil && memo.readsData { + saved.readsData = true + } + if len(held) == 0 { + ctx.quiescent = quiescence{at: ctx.work, writes: ctx.writes, readsData: memo.readsData} } } - ctx.behaviorRunDepth-- ctx.heldBehaviors = held + ctx.holdingDriven = true attached := len(ctx.objectBehaviors) return func() { + ctx.holdingDriven = false ctx.heldBehaviors = nil for _, behavior := range ctx.objectBehaviors[min(attached, len(ctx.objectBehaviors)):] { behavior.settle() @@ -771,6 +818,7 @@ func (ctx *Context) forgetBehaviors(behaviors []*ObjectBehavior) { } ctx.objectBehaviors = behaviorsExcept(ctx.objectBehaviors, dropped) ctx.pendingBehaviors = behaviorsExcept(ctx.pendingBehaviors, dropped) + ctx.workChanged() } // leaveClock releases the behavior's execution, ending the work it left paused @@ -838,11 +886,36 @@ func (ctx *Context) nextRunnableBehavior() (*ObjectBehavior, bool) { return behavior, true } } + // A context a full scan found idle, unchanged since, holds no runnable behavior. + if ctx.quiescent.holds(ctx) { + return nil, false + } + if attached >= len(ctx.objectBehaviors) { + // Every behavior pending is already held by a driver, or there are none. + if attached == 0 && len(ctx.heldBehaviors) == 0 { + ctx.quiescent = quiescence{at: ctx.work, writes: ctx.writes} + } + return nil, false + } + memo := &pendingMemo{} + saved := ctx.polling + ctx.polling = memo for _, behavior := range ctx.objectBehaviors[attached:] { if !ctx.heldBehaviors[behavior] && behavior.hasPendingWork() { + ctx.polling = saved + if saved != nil && memo.readsData { + saved.readsData = true + } return behavior, true } } + ctx.polling = saved + if saved != nil && memo.readsData { + saved.readsData = true + } + if attached == 0 && len(ctx.heldBehaviors) == 0 { + ctx.quiescent = quiescence{at: ctx.work, writes: ctx.writes, readsData: memo.readsData} + } return nil, false } diff --git a/internal/exec/runtime/context.go b/internal/exec/runtime/context.go index e1835cdb1..88690bf51 100644 --- a/internal/exec/runtime/context.go +++ b/internal/exec/runtime/context.go @@ -124,6 +124,9 @@ type Context struct { // heldBehaviors are the behaviors already holding work when the outermost // start under way began: a driver put it in flight, and dispatches it. heldBehaviors map[*ObjectBehavior]bool + // holdingDriven marks that hold however it came out, nil map included; a + // nested start leaves the driving to the outermost one. + holdingDriven bool // objectBehaviors are every behavior an object of this context runs, so a // drain to quiescence can re-run one a sibling's send woke. @@ -241,6 +244,10 @@ type Context struct { // clockRun the run an advance of it draws its due-order choices from. clock Clock clockRun executorRun + // work counts the changes that can leave an attached behavior holding work; + // quiescent is the memo a full scan leaves when it finds them all idle. + work uint64 + quiescent quiescence // onStack lists the runs of the executors whose calls are under way, outermost first. onStack []*executorRun @@ -302,8 +309,7 @@ func NewContext(model *Model, maxSteps int64) *Context { compileCalcs: CalcCompileFromEnv(), run: &runState{ - calcUsageRuns: make(map[int64]map[calcUsageKey]*calcRun), - extentCandidates: make(map[*symbols.Symbol]*extentCandidates), + calcUsageRuns: make(map[int64]map[calcUsageKey]*calcRun), }, calcUsageRunning: make(map[calcUsageKey]*calcShape), @@ -658,9 +664,8 @@ type runState struct { // newRunState is the state a run starts with, under the schedule policy set now. func (ctx *Context) newRunState() *runState { return &runState{ - scheduler: ctx.newScheduler(), - calcUsageRuns: make(map[int64]map[calcUsageKey]*calcRun), - extentCandidates: make(map[*symbols.Symbol]*extentCandidates), + scheduler: ctx.newScheduler(), + calcUsageRuns: make(map[int64]map[calcUsageKey]*calcRun), } } @@ -746,6 +751,7 @@ func (ctx *Context) beginExecutorRun(run *executorRun) func() { run.stir(1) ctx.onStack = append(ctx.onStack, run) leave := ctx.enterRun(run.state) + ctx.workChanged() // A call into an executor whose performer ended in between finds its performance over. if run.exec != nil && run.exec.performerEnded() { run.exec.endTerminated() @@ -754,6 +760,7 @@ func (ctx *Context) beginExecutorRun(run *executorRun) func() { leave() ctx.onStack = ctx.onStack[:len(ctx.onStack)-1] run.stir(-1) + ctx.workChanged() } } @@ -806,7 +813,8 @@ func (ctx *Context) previewExecutorRun(run *executorRun) func() { } else { ctx.run = ctx.newRunState() } - return func() { ctx.run = saved } + ctx.workChanged() + return func() { ctx.run = saved; ctx.workChanged() } } // endExecutorRun brackets the release of a call-by-call driven run: its leftovers diff --git a/internal/exec/runtime/extent.go b/internal/exec/runtime/extent.go index c6ba02360..c355fe7a9 100644 --- a/internal/exec/runtime/extent.go +++ b/internal/exec/runtime/extent.go @@ -407,6 +407,9 @@ func (ctx *Context) extentCandidates(target *symbols.Symbol) []*symbols.Symbol { } } found.judged = append(found.judged, target) + if ctx.run.extentCandidates == nil { + ctx.run.extentCandidates = map[*symbols.Symbol]*extentCandidates{} + } ctx.run.extentCandidates[target] = found } for _, sym := range found.usages { diff --git a/internal/exec/runtime/held_image.go b/internal/exec/runtime/held_image.go index 55e030ab0..86b617ba0 100644 --- a/internal/exec/runtime/held_image.go +++ b/internal/exec/runtime/held_image.go @@ -649,7 +649,7 @@ func (mark materializeMark) rollBack(ctx *Context) { ctx.ids.release(ctx, mark.nextID) } ctx.activations, ctx.runs = mark.activations, mark.runs - ctx.clock.now = mark.clock + ctx.setClock(mark.clock) ctx.clockRun.state = mark.clockRun ctx.lifetimes.dependents = mark.readLives } @@ -721,7 +721,7 @@ func (m *materializing) run() error { dst.livesChanged() dst.activations = max(dst.activations, img.activations) dst.runs = max(dst.runs, img.runs) - dst.clock.now = img.clock + dst.setClock(img.clock) for _, run := range img.runStates { m.runs = append(m.runs, m.runState(run)) } @@ -744,6 +744,7 @@ func (m *materializing) run() error { // Nothing below fails: what names the objects made is installed once they all stand. dst.messages = append(dst.messages, messages...) dst.bus.posts += uint64(len(messages)) + dst.workChanged() for sym, ids := range img.occurrences { dst.occurrences[sym] = slices.Clone(ids) } diff --git a/internal/exec/runtime/held_image_behavior.go b/internal/exec/runtime/held_image_behavior.go index 446de82d5..b67ff035e 100644 --- a/internal/exec/runtime/held_image_behavior.go +++ b/internal/exec/runtime/held_image_behavior.go @@ -484,6 +484,7 @@ func (m *materializing) behavior(b imagedBehavior) error { } inst.behaviors = append(inst.behaviors, behavior) dst.objectBehaviors = append(dst.objectBehaviors, behavior) + dst.workChanged() return nil } diff --git a/internal/exec/runtime/quiescence_test.go b/internal/exec/runtime/quiescence_test.go new file mode 100644 index 000000000..9883685f6 --- /dev/null +++ b/internal/exec/runtime/quiescence_test.go @@ -0,0 +1,228 @@ +package runtime + +import ( + "testing" + + "github.com/Open-MBEE/OpenSysML/internal/semantic/symbols" +) + +const quiescentLampSource = ` + private import ScalarValues::*; + private import SI::*; + attribute def go; + state def Lamp { + entry; then off; + state off; + transition first off accept go then on; + state on; + } + part def Bulb { + attribute level : Integer = 0; + exhibit state lamp : Lamp; +} +` + +const quiescentTimerSource = ` + private import ScalarValues::*; + private import SI::*; + state def Watch { + entry; then waiting; + state waiting; + transition first waiting accept after 1 [s] then done; + state done; + } + part def Winder { exhibit state watch : Watch; } +` + +func quiescentBulb(t *testing.T, src, doc, part string) (*Context, *Instance, *symbols.Scope) { + t.Helper() + idx, _, ctx := buildRuntimeWithLibraries(t, doc, parseAndBuild(t, src)) + root := idx.DocumentRoot(doc) + inst, err := ctx.Instantiate(resolveSymbol(t, root, part)) + if err != nil { + t.Fatalf("Instantiate: %v", err) + } + return ctx, inst, root +} + +// A signal posted into a context a scan found idle is seen as pending work +// again: the next run advances the machine it wakes. +func TestQuiescenceBrokenByPostedSignal(t *testing.T) { + ctx, bulb, root := quiescentBulb(t, quiescentLampSource, "quiescent_lamp.sysml", "Bulb") + if !ctx.quiescent.holds(ctx) { + t.Fatalf("context not quiescent after materialization: work=%d memo=%+v", ctx.work, ctx.quiescent) + } + behavior, ok := bulb.ExhibitedState() + if !ok { + t.Fatal("the bulb exhibits no machine") + } + exec := behavior.State + + msg, err := ctx.SignalMessage(resolveSymbol(t, root, "go"), nil, bulb) + if err != nil { + t.Fatalf("SignalMessage(go): %v", err) + } + ctx.PostMessage(msg) + if ctx.quiescent.holds(ctx) { + t.Fatal("the posted signal left the context quiescent") + } + if err := exec.ProcessNextEvent(); err != nil { + t.Fatalf("ProcessNextEvent: %v", err) + } + if got := activeLeaf(exec); got != "on" { + t.Fatalf("state after go = %s, want on", got) + } +} + +// A clock moved past a wait's due instant is seen as pending work again: the +// advance fires the transition the wait armed. +func TestQuiescenceBrokenByClockAdvance(t *testing.T) { + ctx, winder, _ := quiescentBulb(t, quiescentTimerSource, "quiescent_watch.sysml", "Winder") + if !ctx.quiescent.holds(ctx) { + t.Fatalf("context not quiescent after materialization: work=%d memo=%+v", ctx.work, ctx.quiescent) + } + behavior, ok := winder.ExhibitedState() + if !ok { + t.Fatal("the winder exhibits no machine") + } + exec := behavior.State + if _, err := ctx.Advance(2); err != nil { + t.Fatalf("Advance(2): %v", err) + } + if got := activeLeaf(exec); got != "done" { + t.Fatalf("state after advancing past the timer = %s, want done", got) + } +} + +// A run begun while the context is quiescent counts as work itself: the run +// can leave a machine's new state accepting what the bus already holds. +func TestQuiescenceBrokenByExecutorRun(t *testing.T) { + ctx, bulb, _ := quiescentBulb(t, quiescentLampSource, "quiescent_lamp.sysml", "Bulb") + if !ctx.quiescent.holds(ctx) { + t.Fatalf("context not quiescent after materialization: work=%d memo=%+v", ctx.work, ctx.quiescent) + } + behavior, ok := bulb.ExhibitedState() + if !ok { + t.Fatal("the bulb exhibits no machine") + } + if err := behavior.State.RunToQuiescence(); err != nil { + t.Fatalf("RunToQuiescence: %v", err) + } + if ctx.quiescent.holds(ctx) { + t.Fatal("an executor run left the context quiescent") + } +} + +// A restore puts back the bus, clock and behaviors as they stood: work the +// snapshot saw is work again, so a context found idle since scans once more. +func TestQuiescenceBrokenBySnapshotRestore(t *testing.T) { + ctx, bulb, root := quiescentBulb(t, quiescentLampSource, "quiescent_lamp.sysml", "Bulb") + behavior, ok := bulb.ExhibitedState() + if !ok { + t.Fatal("the bulb exhibits no machine") + } + exec := behavior.State + + msg, err := ctx.SignalMessage(resolveSymbol(t, root, "go"), nil, bulb) + if err != nil { + t.Fatalf("SignalMessage(go): %v", err) + } + ctx.PostMessage(msg) + snap, err := ctx.Snapshot() + if err != nil { + t.Fatalf("Snapshot: %v", err) + } + if err := exec.ProcessNextEvent(); err != nil { + t.Fatalf("ProcessNextEvent: %v", err) + } + if got := activeLeaf(exec); got != "on" { + t.Fatalf("state after go = %s, want on", got) + } + if err := ctx.drainObjectBehaviors(); err != nil { + t.Fatalf("drain: %v", err) + } + if !ctx.quiescent.holds(ctx) { + t.Fatalf("context not quiescent after drain: work=%d memo=%+v", ctx.work, ctx.quiescent) + } + snap.Restore() + if ctx.quiescent.holds(ctx) { + t.Fatal("the restore left the context quiescent") + } + if err := ctx.drainObjectBehaviors(); err != nil { + t.Fatalf("drain after restore: %v", err) + } + if got := activeLeaf(exec); got != "on" { + t.Fatalf("state after restored go = %s, want on", got) + } +} + +// A scan whose poll read the objects' data holds only until the next write; +// one that read nothing holds across writes, as quiescence is meant to. +func TestQuiescenceInvalidatedByWritesOnlyWhenTheScanReadData(t *testing.T) { + ctx, bulb, root := quiescentBulb(t, quiescentLampSource, "quiescent_lamp.sysml", "Bulb") + behavior, ok := bulb.ExhibitedState() + if !ok { + t.Fatal("the bulb exhibits no machine") + } + exec := behavior.State + if !ctx.quiescent.holds(ctx) { + t.Fatalf("context not quiescent after materialization: memo=%+v", ctx.quiescent) + } + if err := bulb.SetFeatureValue(ctx, "level", intArgument(3)); err != nil { + t.Fatalf("SetFeatureValue: %v", err) + } + if !ctx.quiescent.holds(ctx) { + t.Fatal("a write invalidated a scan that read no data") + } + + // A cached poll that read data reports the read to a poll enclosing it, so + // a scan memoizing over it marks its quiescence data-dependent. + exec.pendingSignal() + exec.pending.readsData = true + exec.pending.writes = ctx.writes + probe := &pendingMemo{} + saved := ctx.polling + ctx.polling = probe + exec.pendingSignal() + ctx.polling = saved + if !probe.readsData { + t.Fatal("a cached data-reading poll did not report its read to the enclosing poll") + } + + // Marked as data-reading, the memo no longer covers the write: the work the + // write drives rescans and records a fresh quiescence instead of standing on it. + ctx.quiescent.readsData = true + stale := ctx.quiescent + if err := bulb.SetFeatureValue(ctx, "level", intArgument(4)); err != nil { + t.Fatalf("SetFeatureValue: %v", err) + } + if ctx.quiescent == stale { + t.Fatal("a write left a data-reading scan's quiescence standing") + } + _ = root +} + +// A store on an object with nothing to run allocates no run bookkeeping: the +// behavior scan it brackets finds no behaviors to scan. +func TestIdleStoreAllocatesNoRunBookkeeping(t *testing.T) { + const doc = "idle_store.sysml" + const src = ` + private import ScalarValues::*; + part def Box { attribute level : Integer = 0; } + ` + idx, _, ctx := buildRuntimeWithLibraries(t, doc, parseAndBuild(t, src)) + inst, err := ctx.Instantiate(resolveSymbol(t, idx.DocumentRoot(doc), "Box")) + if err != nil { + t.Fatalf("Instantiate: %v", err) + } + vals := [2]Value{constInt(1), constInt(2)} + i := 0 + if got := testing.AllocsPerRun(1000, func() { + i++ + if err := inst.SetFeatureValue(ctx, "level", vals[i&1]); err != nil { + t.Fatal(err) + } + }); got > 9 { + t.Fatalf("%.0f allocations per idle store, want <= 9", got) + } +} diff --git a/internal/exec/runtime/signal.go b/internal/exec/runtime/signal.go index ff5cb62e6..e67c59744 100644 --- a/internal/exec/runtime/signal.go +++ b/internal/exec/runtime/signal.go @@ -108,6 +108,7 @@ func (ctx *Context) postFrom(msg Message, from *Instance, behavior *symbols.Symb } ctx.messages = append(ctx.messages, msg) ctx.bus.posts++ + ctx.workChanged() if ctx.trace != nil { target, _ := ctx.Instance(msg.Object) ctx.trace.RecordSend(TraceOrigin{At: ctx.clock.now, Object: from, Behavior: behavior}, msg, target) diff --git a/internal/exec/runtime/snapshot.go b/internal/exec/runtime/snapshot.go index 725ab58f6..65222c276 100644 --- a/internal/exec/runtime/snapshot.go +++ b/internal/exec/runtime/snapshot.go @@ -67,6 +67,7 @@ type runCapture struct { evaluations *evaluationLog pendingBehaviors []*ObjectBehavior heldBehaviors mapState[*ObjectBehavior, bool] + holdingDriven bool clockRun *runState } @@ -332,6 +333,7 @@ func (ctx *Context) rollbackJournal(mark journalMark) { ctx.messages = slices.Clone(mark.messages) ctx.bus.cuts++ ctx.writes++ + ctx.workChanged() ctx.abandonCreationSince(mark.created, mark.attached) ctx.clock.now, ctx.clock.waiters = mark.clockNow, slices.Clone(mark.clockWaiters) mark.traced.restore(mark.trace) @@ -349,6 +351,7 @@ func (ctx *Context) captureRun() runCapture { evaluations: ctx.evaluations, pendingBehaviors: slices.Clone(ctx.pendingBehaviors), heldBehaviors: captureMap(ctx.heldBehaviors), + holdingDriven: ctx.holdingDriven, clockRun: ctx.clockRun.state, } return c @@ -369,7 +372,9 @@ func (c runCapture) restore(ctx *Context) { ctx.evaluations = c.evaluations ctx.pendingBehaviors = slices.Clone(c.pendingBehaviors) ctx.heldBehaviors = c.heldBehaviors.restore() + ctx.holdingDriven = c.holdingDriven ctx.clockRun.state = c.clockRun + ctx.workChanged() } // captureRunState captures a run's state once, however many executors share it. diff --git a/internal/exec/runtime/start_behavior.go b/internal/exec/runtime/start_behavior.go index 581814ef2..d6db7a6c4 100644 --- a/internal/exec/runtime/start_behavior.go +++ b/internal/exec/runtime/start_behavior.go @@ -80,6 +80,7 @@ func (ctx *Context) startBehaviorOn(inst *Instance, member *symbols.Symbol) erro inst.behaviors = append(inst.behaviors, behavior) ctx.pendingBehaviors = append(ctx.pendingBehaviors, behavior) ctx.objectBehaviors = append(ctx.objectBehaviors, behavior) + ctx.workChanged() err = ctx.runAttachedBehaviors() endBoundary() if err != nil { diff --git a/internal/exec/runtime/state_executor.go b/internal/exec/runtime/state_executor.go index 878f16cb8..9c4d10f21 100644 --- a/internal/exec/runtime/state_executor.go +++ b/internal/exec/runtime/state_executor.go @@ -556,7 +556,7 @@ func (e *StateExecutor) scheduleCompletionTransitions(state *ast.StateNode) erro if trans.Trigger != nil { continue } - e.eventQueue.Push(Event{ + e.enqueue(Event{ ID: e.nextEventID, Type: EventTime, // Use EventTime with nil trigger Timestamp: e.ctx.clock.now, @@ -623,7 +623,7 @@ func (e *StateExecutor) scheduleTimeTransitions(state *ast.StateNode) error { return err } - e.eventQueue.Push(Event{ + e.enqueue(Event{ ID: e.nextEventID, Type: EventTime, Timestamp: due, @@ -664,7 +664,7 @@ func (e *StateExecutor) processNextEvent() error { } e.moved = true // The clock never lags a dispatched event: a timer popped ahead of it moves it. - e.ctx.clock.now = math.Max(e.ctx.clock.now, event.Timestamp) + e.ctx.setClock(math.Max(e.ctx.clock.now, event.Timestamp)) e.lastEventAt = e.ctx.clock.now e.markDispatch() @@ -1068,7 +1068,7 @@ func (e *StateExecutor) recallDeferredEvents() { continue } event.Timestamp = e.ctx.clock.now - e.eventQueue.Push(event) + e.enqueue(event) } e.deferred = retained } @@ -4155,7 +4155,7 @@ func (e *StateExecutor) InvokeOperation(operation string, args map[string]Value) // queueCall queues a call event carrying the payload. func (e *StateExecutor) queueCall(payload Call) { e.moved = true - e.eventQueue.Push(Event{ + e.enqueue(Event{ ID: e.nextEventID, Type: EventCall, Timestamp: e.ctx.clock.now, @@ -4167,10 +4167,16 @@ func (e *StateExecutor) queueCall(payload Call) { // enqueueSignal queues a message as an accept event, to fire immediately. func (e *StateExecutor) enqueueSignal(msg Message) { e.moved = true - e.eventQueue.Push(e.signalEvent(msg)) + e.enqueue(e.signalEvent(msg)) e.nextEventID++ } +// enqueue queues ev as work the machine's next run takes. +func (e *StateExecutor) enqueue(ev Event) { + e.eventQueue.Push(ev) + e.ctx.workChanged() +} + // signalEvent is the event enqueueSignal queues for a message in flight. func (e *StateExecutor) signalEvent(msg Message) Event { return Event{ @@ -4431,6 +4437,9 @@ func (e *StateExecutor) pendingSignal() (Message, bool) { memo := &e.pending if memo.holds(e) { if memo.ok || memo.bus.posts == e.ctx.bus.posts { + if memo.readsData { + e.ctx.notePollReadsData() + } return memo.msg, memo.ok } // The bus only grew since a negative answer: the messages added are examined.