One guarded handler slot for a bundle's event and heartbeat handlers, so a reload never races the consumer's loop
Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -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() {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user