1
0
Fork 0

Avoid possible race condition with waitgroups

Looks like calling wg.Add(1) from a goroutine can cause a race condition
with wg.Wait() - can be easily avoided by calling Add() before the
subroutine is created.
This commit is contained in:
Michał Rudowicz 2025-01-02 13:40:25 +01:00
parent cf790e88ff
commit 64f4930b4e
4 changed files with 44 additions and 32 deletions

View File

@ -21,9 +21,10 @@ func dumpMemoryProfile(log *log.Logger) {
} }
func WriteMemoryProfilePeriodically(wg *sync.WaitGroup, log *log.Logger, close <-chan interface{}) { func WriteMemoryProfilePeriodically(wg *sync.WaitGroup, log *log.Logger, close <-chan interface{}) {
wg.Add(1)
go func() { go func() {
wg.Add(1)
defer wg.Done() defer wg.Done()
memoryProfileTicker := time.NewTicker(24 * time.Hour) memoryProfileTicker := time.NewTicker(24 * time.Hour)
defer memoryProfileTicker.Stop() defer memoryProfileTicker.Stop()
select { select {

View File

@ -26,19 +26,20 @@ func FilterByTypeOrIndex(ev <-chan satel.Event, wg *sync.WaitGroup, allowedTypes
if (len(allowedTypes) == 0) && (len(allowedIndexes) == 0) { if (len(allowedTypes) == 0) && (len(allowedIndexes) == 0) {
// no allowed types == all types are allowed // no allowed types == all types are allowed
wg.Add(1)
go func() { go func() {
wg.Add(1)
defer wg.Done() defer wg.Done()
defer close(returnChan)
for e := range ev { for e := range ev {
returnChan <- e returnChan <- e
} }
close(returnChan)
}() }()
} else { } else {
wg.Add(1)
go func() { go func() {
wg.Add(1)
defer wg.Done() defer wg.Done()
defer close(returnChan)
for e := range ev { for e := range ev {
retEv := satel.Event{BasicEvents: make([]satel.BasicEventElement, 0)} retEv := satel.Event{BasicEvents: make([]satel.BasicEventElement, 0)}
@ -51,7 +52,6 @@ func FilterByTypeOrIndex(ev <-chan satel.Event, wg *sync.WaitGroup, allowedTypes
returnChan <- retEv returnChan <- retEv
} }
} }
close(returnChan)
}() }()
} }
@ -61,9 +61,10 @@ func FilterByTypeOrIndex(ev <-chan satel.Event, wg *sync.WaitGroup, allowedTypes
func FilterByLastSeen(ev <-chan satel.Event, wg *sync.WaitGroup, dataStore *DataStore, logger *log.Logger) <-chan satel.Event { func FilterByLastSeen(ev <-chan satel.Event, wg *sync.WaitGroup, dataStore *DataStore, logger *log.Logger) <-chan satel.Event {
returnChan := make(chan satel.Event) returnChan := make(chan satel.Event)
wg.Add(1)
go func() { go func() {
wg.Add(1)
defer wg.Done() defer wg.Done()
defer close(returnChan)
for e := range ev { for e := range ev {
retEv := satel.Event{BasicEvents: make([]satel.BasicEventElement, 0)} retEv := satel.Event{BasicEvents: make([]satel.BasicEventElement, 0)}
@ -81,7 +82,6 @@ func FilterByLastSeen(ev <-chan satel.Event, wg *sync.WaitGroup, dataStore *Data
} }
} }
logger.Print("Satel disconnected.") logger.Print("Satel disconnected.")
close(returnChan)
}() }()
return returnChan return returnChan
@ -111,8 +111,11 @@ func Throttle(inputEvents <-chan GenericMessage, wg *sync.WaitGroup, sleeper Sle
returnChan := make(chan GenericMessage) returnChan := make(chan GenericMessage)
timeoutEvents := make(chan interface{}) timeoutEvents := make(chan interface{})
wg.Add(1)
go func() { go func() {
wg.Add(1) defer wg.Done()
defer close(returnChan)
var currentEvent *GenericMessage = nil var currentEvent *GenericMessage = nil
loop: loop:
for { for {
@ -137,8 +140,6 @@ func Throttle(inputEvents <-chan GenericMessage, wg *sync.WaitGroup, sleeper Sle
if currentEvent != nil { if currentEvent != nil {
returnChan <- *currentEvent returnChan <- *currentEvent
} }
close(returnChan)
wg.Done()
}() }()
return returnChan return returnChan

View File

@ -16,12 +16,13 @@ func TestSatelEventTypeFiltering(t *testing.T) {
receivedEvents := make([]satel.Event, 0) receivedEvents := make([]satel.Event, 0)
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
wg.Add(1)
go func() { go func() {
wg.Add(1) defer wg.Done()
for e := range FilterByTypeOrIndex(testEvents, &wg, []SatelChangeType{{satel.ArmedPartition}, {satel.PartitionFireAlarm}}, []int{}) { for e := range FilterByTypeOrIndex(testEvents, &wg, []SatelChangeType{{satel.ArmedPartition}, {satel.PartitionFireAlarm}}, []int{}) {
receivedEvents = append(receivedEvents, e) receivedEvents = append(receivedEvents, e)
} }
wg.Done()
}() }()
testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true) testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
@ -44,12 +45,13 @@ func TestSatelEventTypeFiltering_NoAllowedEventTypesMeansAllAreAllowed(t *testin
receivedEvents := make([]satel.Event, 0) receivedEvents := make([]satel.Event, 0)
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
wg.Add(1)
go func() { go func() {
wg.Add(1) defer wg.Done()
for e := range FilterByTypeOrIndex(testEvents, &wg, []SatelChangeType{}, []int{}) { for e := range FilterByTypeOrIndex(testEvents, &wg, []SatelChangeType{}, []int{}) {
receivedEvents = append(receivedEvents, e) receivedEvents = append(receivedEvents, e)
} }
wg.Done()
}() }()
for index, ct := range SUPPORTED_CHANGE_TYPES { for index, ct := range SUPPORTED_CHANGE_TYPES {
@ -70,12 +72,13 @@ func TestSatelIndexFiltering(t *testing.T) {
receivedEvents := make([]satel.Event, 0) receivedEvents := make([]satel.Event, 0)
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
wg.Add(1)
go func() { go func() {
wg.Add(1) defer wg.Done()
for e := range FilterByTypeOrIndex(testEvents, &wg, []SatelChangeType{}, []int{1, 3}) { for e := range FilterByTypeOrIndex(testEvents, &wg, []SatelChangeType{}, []int{1, 3}) {
receivedEvents = append(receivedEvents, e) receivedEvents = append(receivedEvents, e)
} }
wg.Done()
}() }()
testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true) testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
@ -99,12 +102,13 @@ func TestSatelIndexFiltering_NoAllowedEventTypesMeansAllAreAllowed(t *testing.T)
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
myReasonableMaxIndex := 100 // I wanted to use math.MaxInt at first, but it's kind of a waste of time here myReasonableMaxIndex := 100 // I wanted to use math.MaxInt at first, but it's kind of a waste of time here
wg.Add(1)
go func() { go func() {
wg.Add(1) defer wg.Done()
for e := range FilterByTypeOrIndex(testEvents, &wg, []SatelChangeType{}, []int{}) { for e := range FilterByTypeOrIndex(testEvents, &wg, []SatelChangeType{}, []int{}) {
receivedEvents = append(receivedEvents, e) receivedEvents = append(receivedEvents, e)
} }
wg.Done()
}() }()
for i := 0; i < myReasonableMaxIndex; i++ { for i := 0; i < myReasonableMaxIndex; i++ {
@ -132,12 +136,13 @@ func TestSatelLastSeenFiltering(t *testing.T) {
fakeLog := log.New(io.Discard, "", log.Ltime) fakeLog := log.New(io.Discard, "", log.Ltime)
ds := MakeDataStore(fakeLog, tempFileName) ds := MakeDataStore(fakeLog, tempFileName)
wg.Add(1)
go func() { go func() {
wg.Add(1) defer wg.Done()
for e := range FilterByLastSeen(testEvents, &wg, &ds, fakeLog) { for e := range FilterByLastSeen(testEvents, &wg, &ds, fakeLog) {
receivedEvents = append(receivedEvents, e) receivedEvents = append(receivedEvents, e)
} }
wg.Done()
}() }()
testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true) testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
@ -168,12 +173,13 @@ func TestSatelLastSeenFilteringWithPersistence(t *testing.T) {
fakeLog := log.New(io.Discard, "", log.Ltime) fakeLog := log.New(io.Discard, "", log.Ltime)
ds := MakeDataStore(fakeLog, tempFileName) ds := MakeDataStore(fakeLog, tempFileName)
wg.Add(1)
go func() { go func() {
wg.Add(1) defer wg.Done()
for e := range FilterByLastSeen(testEvents, &wg, &ds, fakeLog) { for e := range FilterByLastSeen(testEvents, &wg, &ds, fakeLog) {
receivedEvents = append(receivedEvents, e) receivedEvents = append(receivedEvents, e)
} }
wg.Done()
}() }()
testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true) testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
@ -194,12 +200,13 @@ func TestSatelLastSeenFilteringWithPersistence(t *testing.T) {
testEvents = make(chan satel.Event) testEvents = make(chan satel.Event)
receivedEvents = make([]satel.Event, 0) receivedEvents = make([]satel.Event, 0)
ds = MakeDataStore(fakeLog, tempFileName) ds = MakeDataStore(fakeLog, tempFileName)
wg.Add(1)
go func() { go func() {
wg.Add(1) defer wg.Done()
for e := range FilterByLastSeen(testEvents, &wg, &ds, fakeLog) { for e := range FilterByLastSeen(testEvents, &wg, &ds, fakeLog) {
receivedEvents = append(receivedEvents, e) receivedEvents = append(receivedEvents, e)
} }
wg.Done()
}() }()
receivedEvents = make([]satel.Event, 0) receivedEvents = make([]satel.Event, 0)
@ -243,12 +250,13 @@ func TestThrottle(t *testing.T) {
tplMessageTest4 = satel.BasicEventElement{Type: satel.ZoneViolation, Index: 2, Value: false} tplMessageTest4 = satel.BasicEventElement{Type: satel.ZoneViolation, Index: 2, Value: false}
) )
wg.Add(1)
go func() { go func() {
wg.Add(1) defer wg.Done()
for e := range Throttle(testEvents, &wg, &mockSleeper, fakeLog) { for e := range Throttle(testEvents, &wg, &mockSleeper, fakeLog) {
receivedEvents = append(receivedEvents, e) receivedEvents = append(receivedEvents, e)
} }
wg.Done()
}() }()
testEvents <- GenericMessage{[]satel.BasicEventElement{tplMessageTest1}} testEvents <- GenericMessage{[]satel.BasicEventElement{tplMessageTest1}}
@ -295,12 +303,12 @@ func TestThrottle_ManyMessagesInOneEvent(t *testing.T) {
tplMessageTest4 = satel.BasicEventElement{Type: satel.ZoneViolation, Index: 2, Value: false} tplMessageTest4 = satel.BasicEventElement{Type: satel.ZoneViolation, Index: 2, Value: false}
) )
wg.Add(1)
go func() { go func() {
wg.Add(1) defer wg.Done()
for e := range Throttle(testEvents, &wg, &mockSleeper, fakeLog) { for e := range Throttle(testEvents, &wg, &mockSleeper, fakeLog) {
receivedEvents = append(receivedEvents, e) receivedEvents = append(receivedEvents, e)
} }
wg.Done()
}() }()
testEvents <- makeMassiveEvent(tplMessageTest1, 100) testEvents <- makeMassiveEvent(tplMessageTest1, 100)

View File

@ -31,9 +31,11 @@ func Consume(events <-chan GenericMessage) {
func SendToTg(events <-chan GenericMessage, s Sender, wg *sync.WaitGroup, logger *log.Logger, tpl *template.Template) <-chan GenericMessage { func SendToTg(events <-chan GenericMessage, s Sender, wg *sync.WaitGroup, logger *log.Logger, tpl *template.Template) <-chan GenericMessage {
returnEvents := make(chan GenericMessage) returnEvents := make(chan GenericMessage)
wg.Add(1)
go func() { go func() {
wg.Add(1)
defer wg.Done() defer wg.Done()
defer close(returnEvents)
for e := range events { for e := range events {
returnEvents <- e returnEvents <- e
err := s.Send(e, tpl) err := s.Send(e, tpl)
@ -42,7 +44,6 @@ func SendToTg(events <-chan GenericMessage, s Sender, wg *sync.WaitGroup, logger
panic(err) panic(err)
} }
} }
close(returnEvents)
}() }()
return returnEvents return returnEvents
@ -74,9 +75,11 @@ func notifyAllHttp(urls []string, logger *log.Logger, wg *sync.WaitGroup) {
func NotifyViaHTTP(events <-chan GenericMessage, config AppConfig, wg *sync.WaitGroup, logger *log.Logger) <-chan GenericMessage { func NotifyViaHTTP(events <-chan GenericMessage, config AppConfig, wg *sync.WaitGroup, logger *log.Logger) <-chan GenericMessage {
returnEvents := make(chan GenericMessage) returnEvents := make(chan GenericMessage)
wg.Add(1)
go func() { go func() {
wg.Add(1)
defer wg.Done() defer wg.Done()
defer close(returnEvents)
for e := range events { for e := range events {
returnEvents <- e returnEvents <- e
inner_arm: inner_arm:
@ -101,7 +104,6 @@ func NotifyViaHTTP(events <-chan GenericMessage, config AppConfig, wg *sync.Wait
} }
} }
close(returnEvents)
}() }()
return returnEvents return returnEvents