diff --git a/node-tools/internal/runtime/events_test.go b/node-tools/internal/runtime/events_test.go index b2b7ebf..3f98d4e 100644 --- a/node-tools/internal/runtime/events_test.go +++ b/node-tools/internal/runtime/events_test.go @@ -160,7 +160,7 @@ func TestALongRunningBundleThatExitsIsStartedAgain(t *testing.T) { func fmtSprintf(format string, args ...any) string { return fmt.Sprintf(format, args...) } func errorf(format string, args ...any) error { return fmt.Errorf(format, args...) } -// A bundle loaded again sets its heartbeat handler while the consumer's loop hands a word on: no data race (-race), +// A bundle loaded again sets its event and heartbeat handlers while the consumer's loop hands a word or an event on: no data race (-race), // and the heartbeat is asked of the bus only while some bundle holds a handler. func TestAHeartbeatHandlerSetAgainWhileWordsAreHandedOnIsNotARace(t *testing.T) { m := &moduleEvents{} @@ -170,16 +170,23 @@ func TestAHeartbeatHandlerSetAgainWhileWordsAreHandedOnIsNotARace(t *testing.T) if m.beatAsked() { t.Fatal("a heartbeat is asked while no bundle gave a handler") } + took := &deliverSlot{} + took.set(func(json.RawMessage) error { return nil }) + m.delivers = []*deliverSlot{took} var got atomic.Int32 done := make(chan struct{}) go func() { defer close(done) for i := 0; i < 200; i++ { slot.set(func(json.RawMessage) { got.Add(1) }) + took.set(func(json.RawMessage) error { got.Add(1); return nil }) } }() for i := 0; i < 200; i++ { all.handBeat(m, bus.Beat{At: time.Now()}) + if err := all.handOut("mod", m, bus.Envelope{Key: "a.b", Body: json.RawMessage(`{}`)}); err != nil { + t.Fatal(err) + } } <-done if !m.beatAsked() { diff --git a/node-tools/internal/runtime/reload_test.go b/node-tools/internal/runtime/reload_test.go index 52599df..7375d77 100644 --- a/node-tools/internal/runtime/reload_test.go +++ b/node-tools/internal/runtime/reload_test.go @@ -343,7 +343,7 @@ func TestAnEventHeardDuringAReloadIsOfferedAgainNotAPanic(t *testing.T) { c := &consumers{logf: func(string, ...any) {}, of: map[string]*moduleEvents{}} var got []string before := func(json.RawMessage) error { got = append(got, "before"); return nil } - old := &moduleEvents{delivers: []*func(json.RawMessage) error{&before}} + old := &moduleEvents{delivers: []*deliverSlot{{f: before}}} c.of["ledger"] = old c.dropModule("ledger") env := bus.Envelope{Key: "alpha.happened", Body: json.RawMessage(`{"n":1}`)} @@ -359,7 +359,7 @@ func TestAnEventHeardDuringAReloadIsOfferedAgainNotAPanic(t *testing.T) { t.Error("an event heard while the module is loaded again was taken, not offered again") } after := func(json.RawMessage) error { got = append(got, "after"); return nil } - fresh := &moduleEvents{delivers: []*func(json.RawMessage) error{&after}} + fresh := &moduleEvents{delivers: []*deliverSlot{{f: after}}} c.of["ledger"] = fresh if err := hand(fresh); err != nil { t.Errorf("the bundle loaded again was not handed the event: %v", err) diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index 597fffb..de633ed 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -774,29 +774,34 @@ type consumers struct { type moduleEvents struct { stop func() - delivers []*func(json.RawMessage) error + delivers []*deliverSlot beats []*beatSlot } -// beatSlot is where a bundle's heartbeat handler is kept; a bundle loaded again sets it while the consumer's loop -// reads it, so both go through its own lock. -type beatSlot struct { +// slot is where a bundle's handler is kept (its event handler, its heartbeat handler); a bundle loaded again sets it +// while the consumer's loop reads it, so both go through its own lock. +type slot[F any] struct { mu sync.Mutex - f func(json.RawMessage) + f F } -func (s *beatSlot) get() func(json.RawMessage) { +func (s *slot[F]) get() F { s.mu.Lock() defer s.mu.Unlock() return s.f } -func (s *beatSlot) set(f func(json.RawMessage)) { +func (s *slot[F]) set(f F) { s.mu.Lock() defer s.mu.Unlock() s.f = f } +type ( + deliverSlot = slot[func(json.RawMessage) error] + beatSlot = slot[func(json.RawMessage)] +) + // beatAsked says whether any bundle of the module asked for the heartbeat; the caller holds c.mu. func (m *moduleEvents) beatAsked() bool { for _, t := range m.beats { @@ -851,7 +856,7 @@ type moduleBus struct { all *consumers module string mu sync.Mutex - deliver *func(json.RawMessage) error + deliver *deliverSlot beat *beatSlot } @@ -893,11 +898,9 @@ func (b *moduleBus) Ask(params json.RawMessage) (json.RawMessage, error) { func (b *moduleBus) Subscribe(deliver func(json.RawMessage) error, beat func(json.RawMessage)) error { b.mu.Lock() if b.deliver == nil { - d := deliver - b.deliver = &d - } else { - *b.deliver = deliver + b.deliver = &deliverSlot{} } + b.deliver.set(deliver) mine := b.deliver if b.beat == nil { b.beat = &beatSlot{} @@ -976,10 +979,10 @@ func (c *consumers) handOut(module string, m *moduleEvents, env bus.Envelope) er c.mu.Unlock() return errModuleReloading } - targets := append([]*func(json.RawMessage) error(nil), m.delivers...) + targets := append([]*deliverSlot(nil), m.delivers...) c.mu.Unlock() for _, d := range targets { - if err := (*d)(raw); err != nil { + if err := d.get()(raw); err != nil { c.logf("[mesh-tools] %s did not take %s%s: %s; offered again in %s", module, env.Key, eventRef(env), whyNotTaken(err), bus.NakDelay) return err