libs: Fix event concurrency flaw (#2519)

* fix event concurrency flaw
* modify changelog
* fix a mistake
* fix a lint issue
* modify changelog
* modify for review issue
* modify for review issue
* modify for review issue

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)

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 {

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)