This is an automated email from the ASF dual-hosted git repository.
manirajv06 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/yunikorn-core.git
The following commit(s) were added to refs/heads/master by this push:
new aa59599e [YUNIKORN-3409] Fix concurrent map read and write on
rejectedApplications (#1144)
aa59599e is described below
commit aa59599e9be439778f21b1b437bce42ecc1def17
Author: hedger9487 <[email protected]>
AuthorDate: Fri Sep 4 12:04:29 2026 +0530
[YUNIKORN-3409] Fix concurrent map read and write on rejectedApplications
(#1144)
- Add partition write lock to AddRejectedApplication around
rejectedApplications mutation
- Protect rejectedApplications map pointer read with RLock in
getRejectedAppsByState
- Add comprehensive concurrent unit tests using a start gate barrier
Closes: #1144
Signed-off-by: mani <[email protected]>
---
pkg/scheduler/partition.go | 18 ++++--
pkg/scheduler/partition_test.go | 118 ++++++++++++++++++++++++++++++++++++++++
2 files changed, 130 insertions(+), 6 deletions(-)
diff --git a/pkg/scheduler/partition.go b/pkg/scheduler/partition.go
index dd900fad..3c3c9f20 100644
--- a/pkg/scheduler/partition.go
+++ b/pkg/scheduler/partition.go
@@ -1111,9 +1111,7 @@ func (pc *PartitionContext) GetRejectedApplications()
[]*objects.Application {
return appList
}
-func (pc *PartitionContext) getAppsState(appMap
map[string]*objects.Application, state string) []string {
- pc.RLock()
- defer pc.RUnlock()
+func (pc *PartitionContext) getAppsStateInternal(appMap
map[string]*objects.Application, state string) []string {
var apps []string
for appID, app := range appMap {
if app.CurrentState() == state {
@@ -1126,17 +1124,23 @@ func (pc *PartitionContext) getAppsState(appMap
map[string]*objects.Application,
// getAppsByState returns a slice of applicationIDs for the current
applications filtered by state
// Completed and Rejected applications are tracked in a separate map and will
never be included.
func (pc *PartitionContext) getAppsByState(state string) []string {
- return pc.getAppsState(pc.applications, state)
+ pc.RLock()
+ defer pc.RUnlock()
+ return pc.getAppsStateInternal(pc.applications, state)
}
// getRejectedAppsByState returns a slice of applicationIDs for the rejected
applications filtered by state.
func (pc *PartitionContext) getRejectedAppsByState(state string) []string {
- return pc.getAppsState(pc.rejectedApplications, state)
+ pc.RLock()
+ defer pc.RUnlock()
+ return pc.getAppsStateInternal(pc.rejectedApplications, state)
}
// getCompletedAppsByState returns a slice of applicationIDs for the completed
applicationIDs filtered by state.
func (pc *PartitionContext) getCompletedAppsByState(state string) []string {
- return pc.getAppsState(pc.completedApplications, state)
+ pc.RLock()
+ defer pc.RUnlock()
+ return pc.getAppsStateInternal(pc.completedApplications, state)
}
// cleanupExpiredApps cleans up applications in the Expired state from the
three tracking maps
@@ -1763,6 +1767,8 @@ func (pc *PartitionContext)
AddRejectedApplication(rejectedApplication *objects.
zap.String("currentState",
rejectedApplication.CurrentState()),
zap.Error(err))
}
+ pc.Lock()
+ defer pc.Unlock()
if pc.rejectedApplications == nil {
pc.rejectedApplications = make(map[string]*objects.Application)
}
diff --git a/pkg/scheduler/partition_test.go b/pkg/scheduler/partition_test.go
index 03d0495a..99db05a6 100644
--- a/pkg/scheduler/partition_test.go
+++ b/pkg/scheduler/partition_test.go
@@ -22,6 +22,7 @@ import (
"fmt"
"strconv"
"strings"
+ "sync"
"testing"
"time"
@@ -2924,6 +2925,123 @@ func TestCleanupRejectedApps(t *testing.T) {
assert.Equal(t, 0,
len(partition.getRejectedAppsByState(objects.Expired.String())), "the partition
should have 0 expired app")
}
+func TestAddRejectedApplicationConcurrent(t *testing.T) {
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
+ var wg sync.WaitGroup
+ numGoroutines := 20
+ appsPerGoroutine := 10
+ start := make(chan struct{})
+
+ // Concurrent Writers
+ for i := 0; i < numGoroutines; i++ {
+ wg.Add(1)
+ go func(workerID int) {
+ defer wg.Done()
+ <-start
+ for j := 0; j < appsPerGoroutine; j++ {
+ appID := fmt.Sprintf("app-%d-%d", workerID, j)
+ app := newApplication(appID, "default",
defQueue)
+ partition.AddRejectedApplication(app, "rejected
reason")
+ }
+ }(i)
+ }
+
+ // Concurrent Readers
+ for i := 0; i < numGoroutines; i++ {
+ wg.Add(1)
+ go func(workerID int) {
+ defer wg.Done()
+ <-start
+ for j := 0; j < appsPerGoroutine; j++ {
+ _ = partition.GetRejectedApplications()
+ appID := fmt.Sprintf("app-%d-%d", workerID, j)
+ _ = partition.getRejectedApplication(appID)
+ _ =
partition.getRejectedAppsByState(objects.Rejected.String())
+ }
+ }(i)
+ }
+
+ close(start)
+ wg.Wait()
+ assert.Equal(t, numGoroutines*appsPerGoroutine,
len(partition.GetRejectedApplications()), "all rejected applications should be
tracked")
+}
+
+func TestAddRejectedApplicationConcurrentWithCleanup(t *testing.T) {
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
+ var wg sync.WaitGroup
+ numGoroutines := 20
+ appsPerGoroutine := 10
+ start := make(chan struct{})
+
+ // Concurrent Writers
+ for i := 0; i < numGoroutines; i++ {
+ wg.Add(1)
+ go func(workerID int) {
+ defer wg.Done()
+ <-start
+ for j := 0; j < appsPerGoroutine; j++ {
+ appID := fmt.Sprintf("cleanup-app-%d-%d",
workerID, j)
+ app := newApplication(appID, "default",
defQueue)
+ partition.AddRejectedApplication(app, "rejected
reason")
+ if j%2 == 0 {
+ app.SetState(objects.Expired.String())
+ }
+ }
+ }(i)
+ }
+
+ // Concurrent Cleaners and Readers
+ for i := 0; i < numGoroutines; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ <-start
+ for j := 0; j < appsPerGoroutine; j++ {
+ partition.cleanupExpiredApps()
+ _ = partition.GetRejectedApplications()
+ _ =
partition.getRejectedAppsByState(objects.Expired.String())
+ _ =
partition.getRejectedAppsByState(objects.Rejected.String())
+ }
+ }()
+ }
+
+ close(start)
+ wg.Wait()
+ partition.cleanupExpiredApps()
+ assert.Equal(t, 0,
len(partition.getRejectedAppsByState(objects.Expired.String())), "no expired
apps should remain after cleanup")
+}
+
+func TestAddRejectedApplicationDuplicate(t *testing.T) {
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
+ var wg sync.WaitGroup
+ numGoroutines := 20
+ appID := "duplicate-app"
+ start := make(chan struct{})
+
+ for i := 0; i < numGoroutines; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ <-start
+ app := newApplication(appID, "default", defQueue)
+ partition.AddRejectedApplication(app, "duplicate
rejected reason")
+ }()
+ }
+
+ close(start)
+ wg.Wait()
+ assert.Equal(t, 1, len(partition.GetRejectedApplications()), "duplicate
rejected app should only have 1 entry")
+}
+
func TestUpdateNode(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err, "test partition create failed with error")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
