This is an automated email from the ASF dual-hosted git repository.
wilfred-s pushed a commit to branch branch-1.9
in repository https://gitbox.apache.org/repos/asf/yunikorn-core.git
The following commit(s) were added to refs/heads/branch-1.9 by this push:
new 4886d4e0 [YUNIKORN-3296] Fix race between AddEvent and Stop (#1093)
4886d4e0 is described below
commit 4886d4e0dac6c622681bd87b63d01c7b9e206c0d
Author: Venkateshwaran Shanmugham <[email protected]>
AuthorDate: Mon Jun 22 15:22:22 2026 +1000
[YUNIKORN-3296] Fix race between AddEvent and Stop (#1093)
Stop() closes and clears ec.channel while holding the event system mutex.
AddEvent() previously accessed ec.channel without synchronization. During
TestSchedulerRecoveryQuotaPreemption, scheduler go-routines could still emit
events while StopAll() shut down the event system, resulting in a race
detected by go test -race.
The reported race was between:
EventSystemImpl.AddEvent()
EventSystemImpl.Stop() (close(ec.channel) / ec.channel = nil)
This race could also lead to a send on closed channel panic.
Closes: #1093
Signed-off-by: Venkateshwaran Shanmugham <[email protected]>
Signed-off-by: Wilfred Spiegelenburg <[email protected]>
(cherry picked from commit 20c6c151b54916a54e73af1cc2475a1b895a994b)
---
pkg/events/event_system.go | 13 +++++++++++++
pkg/events/event_system_test.go | 29 +++++++++++++++++++++++++++++
2 files changed, 42 insertions(+)
diff --git a/pkg/events/event_system.go b/pkg/events/event_system.go
index 3306beaf..117669e1 100644
--- a/pkg/events/event_system.go
+++ b/pkg/events/event_system.go
@@ -220,6 +220,10 @@ func (ec *EventSystemImpl) Stop() {
ec.stopped = true
}
+func (ec *EventSystemImpl) isStopped() bool {
+ return ec.stopped || ec.channel == nil
+}
+
// AddEvent adds an event record to the event system. See the interface for
details.
func (ec *EventSystemImpl) AddEvent(event *si.EventRecord) {
if event != nil {
@@ -227,6 +231,15 @@ func (ec *EventSystemImpl) AddEvent(event *si.EventRecord)
{
}
metrics.GetEventMetrics().IncEventsCreated()
+
+ ec.RLock()
+ defer ec.RUnlock()
+
+ if ec.isStopped() {
+ metrics.GetEventMetrics().IncEventsNotChanneled()
+ return
+ }
+
select {
case ec.channel <- event:
metrics.GetEventMetrics().IncEventsChanneled()
diff --git a/pkg/events/event_system_test.go b/pkg/events/event_system_test.go
index 7bce1f1f..18b24ae4 100644
--- a/pkg/events/event_system_test.go
+++ b/pkg/events/event_system_test.go
@@ -21,6 +21,7 @@ package events
import (
"strconv"
"strings"
+ "sync"
"testing"
"time"
@@ -241,3 +242,31 @@ func TestTruncateEventMessage(t *testing.T) {
func getTestString(stringLength int) string {
return strings.Repeat("x", stringLength)
}
+
+// TestAddEventConcurrentStop verifies AddEvent and Stop can run concurrently
without data races.
+func TestAddEventConcurrentStop(t *testing.T) {
+ Init()
+ eventSystem := GetEventSystem().(*EventSystemImpl) //nolint:errcheck
+ eventSystem.StartServiceWithPublisher(false)
+
+ var wg sync.WaitGroup
+ wg.Add(2)
+
+ go func() {
+ defer wg.Done()
+ for i := 0; i < 10000; i++ {
+ eventSystem.AddEvent(&si.EventRecord{
+ Type: si.EventRecord_REQUEST,
+ Message: strconv.Itoa(i),
+ })
+ }
+ }()
+
+ go func() {
+ defer wg.Done()
+ time.Sleep(time.Millisecond)
+ eventSystem.Stop()
+ }()
+
+ wg.Wait()
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
