diff --git a/CHANGELOG_PENDING.md b/CHANGELOG_PENDING.md index 1a927a329..0cd8bd2fe 100644 --- a/CHANGELOG_PENDING.md +++ b/CHANGELOG_PENDING.md @@ -50,3 +50,4 @@ timeoutPrecommit before starting next round - [common/bit_array] Fixed a bug in the `Sub` function (@bradyjoestar) - [common] \#2534 make bit array's PickRandom choose uniformly from true bits - [p2p] \#2555 fix p2p switch FlushThrottle value (@goolAdapter) +- [libs/event] \#2518 fix event concurrency flaw (@goolAdapter) diff --git a/libs/events/events.go b/libs/events/events.go index 9c7f0fd05..864365563 100644 --- a/libs/events/events.go +++ b/libs/events/events.go @@ -4,15 +4,23 @@ Pub-Sub in go with event caching package events import ( + "fmt" "sync" cmn "github.com/tendermint/tendermint/libs/common" ) +type ErrListenerWasRemoved struct { + listener string +} + +func (e ErrListenerWasRemoved) Error() string { + return fmt.Sprintf("listener %s was removed", e.listener) +} + // Generic event data can be typed and registered with tendermint/go-amino // via concrete implementation of this interface type EventData interface { - //AssertIsEventData() } // reactors and other modules should export @@ -30,7 +38,7 @@ type EventSwitch interface { cmn.Service Fireable - AddListenerForEvent(listenerID, event string, cb EventCallback) + AddListenerForEvent(listenerID, event string, cb EventCallback) error RemoveListenerForEvent(event string, listenerID string) RemoveListener(listenerID string) } @@ -58,7 +66,7 @@ func (evsw *eventSwitch) OnStart() error { func (evsw *eventSwitch) OnStop() {} -func (evsw *eventSwitch) AddListenerForEvent(listenerID, event string, cb EventCallback) { +func (evsw *eventSwitch) AddListenerForEvent(listenerID, event string, cb EventCallback) error { // Get/Create eventCell and listener evsw.mtx.Lock() eventCell := evsw.eventCells[event] @@ -74,8 +82,12 @@ func (evsw *eventSwitch) AddListenerForEvent(listenerID, event string, cb EventC evsw.mtx.Unlock() // Add event and listener - eventCell.AddListener(listenerID, cb) - listener.AddEvent(event) + err := listener.AddEvent(event) + if err == nil { + eventCell.AddListener(listenerID, cb) + } + + return err } func (evsw *eventSwitch) RemoveListener(listenerID string) { @@ -168,10 +180,15 @@ func (cell *eventCell) RemoveListener(listenerID string) int { func (cell *eventCell) FireEvent(data EventData) { cell.mtx.RLock() + var listenerCopy []EventCallback for _, listener := range cell.listeners { - listener(data) + listenerCopy = append(listenerCopy, listener) } cell.mtx.RUnlock() + + for _, listener := range listenerCopy { + listener(data) + } } //----------------------------------------------------------------------------- @@ -194,14 +211,16 @@ func newEventListener(id string) *eventListener { } } -func (evl *eventListener) AddEvent(event string) { +func (evl *eventListener) AddEvent(event string) error { evl.mtx.Lock() defer evl.mtx.Unlock() if evl.removed { - return + return ErrListenerWasRemoved{listener: evl.id} } + evl.events = append(evl.events, event) + return nil } func (evl *eventListener) GetEvents() []string { diff --git a/libs/events/events_test.go b/libs/events/events_test.go index a01fbbb77..02ec44c4f 100644 --- a/libs/events/events_test.go +++ b/libs/events/events_test.go @@ -17,6 +17,7 @@ func TestAddListenerForEventFireOnce(t *testing.T) { if err != nil { t.Errorf("Failed to start EventSwitch, error: %v", err) } + defer evsw.Stop() messages := make(chan EventData) evsw.AddListenerForEvent("listener", "event", func(data EventData) { @@ -37,6 +38,7 @@ func TestAddListenerForEventFireMany(t *testing.T) { if err != nil { t.Errorf("Failed to start EventSwitch, error: %v", err) } + defer evsw.Stop() doneSum := make(chan uint64) doneSending := make(chan uint64) numbers := make(chan uint64, 4) @@ -66,6 +68,7 @@ func TestAddListenerForDifferentEvents(t *testing.T) { if err != nil { t.Errorf("Failed to start EventSwitch, error: %v", err) } + defer evsw.Stop() doneSum := make(chan uint64) doneSending1 := make(chan uint64) doneSending2 := make(chan uint64) @@ -111,6 +114,7 @@ func TestAddDifferentListenerForDifferentEvents(t *testing.T) { if err != nil { t.Errorf("Failed to start EventSwitch, error: %v", err) } + defer evsw.Stop() doneSum1 := make(chan uint64) doneSum2 := make(chan uint64) doneSending1 := make(chan uint64) @@ -162,6 +166,54 @@ func TestAddDifferentListenerForDifferentEvents(t *testing.T) { } } +func TestAddAndRemoveListenerConcurrency(t *testing.T) { + var ( + stopInputEvent = false + roundCount = 2000 + ) + + evsw := NewEventSwitch() + err := evsw.Start() + if err != nil { + t.Errorf("Failed to start EventSwitch, error: %v", err) + } + defer evsw.Stop() + + done1 := make(chan struct{}) + done2 := make(chan struct{}) + + go func() { + for i := 0; i < roundCount; i++ { + evsw.RemoveListener("listener") + } + done1 <- struct{}{} + }() + + go func() { + for i := 0; i < roundCount; i++ { + index := i //it necessary for closure + evsw.AddListenerForEvent("listener", fmt.Sprintf("event%d", index), + func(data EventData) { + t.Errorf("should not run callback for %d.\n", index) + stopInputEvent = true + }) + } + done2 <- struct{}{} + }() + + <-done1 + <-done2 + + close(done1) + close(done2) + + evsw.RemoveListener("listener") // make sure remove last + + for i := 0; i < roundCount && !stopInputEvent; i++ { + evsw.FireEvent(fmt.Sprintf("event%d", i), uint64(1001)) + } +} + // TestAddAndRemoveListener sets up an EventSwitch, subscribes a listener to // two events, fires a thousand integers for the first event, then unsubscribes // the listener and fires a thousand integers for the second event. @@ -171,6 +223,7 @@ func TestAddAndRemoveListener(t *testing.T) { if err != nil { t.Errorf("Failed to start EventSwitch, error: %v", err) } + defer evsw.Stop() doneSum1 := make(chan uint64) doneSum2 := make(chan uint64) doneSending1 := make(chan uint64) @@ -216,6 +269,7 @@ func TestRemoveListener(t *testing.T) { if err != nil { t.Errorf("Failed to start EventSwitch, error: %v", err) } + defer evsw.Stop() count := 10 sum1, sum2 := 0, 0 // add some listeners and make sure they work @@ -269,6 +323,7 @@ func TestRemoveListenersAsync(t *testing.T) { if err != nil { t.Errorf("Failed to start EventSwitch, error: %v", err) } + defer evsw.Stop() doneSum1 := make(chan uint64) doneSum2 := make(chan uint64) doneSending1 := make(chan uint64)