This is an automated email from the ASF dual-hosted git repository.
HuangTing-Yao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/yunikorn-k8shim.git
The following commit(s) were added to refs/heads/master by this push:
new ba0fe9a4 [YUNIKORN-3427] Fix unsynchronised context and task access on
deferred release (#1097)
ba0fe9a4 is described below
commit ba0fe9a45345aa4686021de638ab596a8d5624e6
Author: rjgoyln <[email protected]>
AuthorDate: Mon Sep 21 21:20:26 2026 +0800
[YUNIKORN-3427] Fix unsynchronised context and task access on deferred
release (#1097)
flushReleaseableTasks runs as the enter-Accepted callback under the
application
lock and nothing else. On the all-terminated path it deleted the application
from the context map directly, while the scheduling loop lists that map on
every
tick: a concurrent map read and write, which the runtime treats as fatal.
On the
other path it read task fields that the core callback writes when an
allocation
lands.
The context lock is taken before the application lock everywhere else, so
the
removal cannot run from inside the transition and moves to handle(), which
has
released the application lock by then. The task lock is the one that comes
after
the application lock, so the release replay can take it directly.
[YUNIKORN-3427] Name the locked half of handle after what it does
The two halves were called handle and handleEvent, which are interchangeable
names that leave the doc comment to carry the whole lock invariant. The test
helper the second test grew is useful to the first one as well.
[YUNIKORN-3427] Say which lock the callback rule is about
The rule above handle() predates a callback that takes a lock of its own,
and
reads as if any lock were forbidden. Only the application lock is.
Closes: #1097
Signed-off-by: HuangTing-Yao <[email protected]>
---
pkg/cache/application.go | 31 +++++++++++----
pkg/cache/application_test.go | 90 +++++++++++++++++++++++++++++++++++++++++++
pkg/cache/context.go | 4 --
pkg/cache/task.go | 8 ++++
4 files changed, 122 insertions(+), 11 deletions(-)
diff --git a/pkg/cache/application.go b/pkg/cache/application.go
index 3295d888..2faa0f4c 100644
--- a/pkg/cache/application.go
+++ b/pkg/cache/application.go
@@ -64,6 +64,7 @@ type Application struct {
originatingTask *Task // Original Pod which creates the
requests
releaseableTasks []*Task
context *Context
+ removeFromContext bool // handle() does the removal: it needs
the context lock
}
const transitionErr = "no transition"
@@ -98,6 +99,18 @@ func NewApplication(appID, queueName, user string, groups
[]string, tags map[str
}
func (app *Application) handle(ev events.ApplicationEvent) error {
+ removeFrom, err := app.runTransition(ev)
+ // Context.RemoveApplication takes the context lock, which every other
path takes before
+ // the application lock, so the removal can only run once runTransition
has released it.
+ if removeFrom != nil {
+ removeFrom.RemoveApplication(app.applicationID)
+ }
+ return err
+}
+
+// runTransition runs the event through the state machine while holding the
application lock and
+// reports the context the transition asked to remove the application from, if
any.
+func (app *Application) runTransition(ev events.ApplicationEvent) (*Context,
error) {
// Locking mechanism:
// 1) when handle event transitions, we first obtain the object's lock,
// this helps us to place a pre-check before entering here, in case
@@ -106,15 +119,21 @@ func (app *Application) handle(ev
events.ApplicationEvent) error {
// to protect the transition phase.
// 2) Note, state machine calls those callbacks here, we must ensure
// they are lock-free calls. Otherwise the callback will be blocked
- // because the lock is already held here.
+ // because the lock is already held here. A lock that is ordered
after
+ // this one, the task lock, is safe for a callback to take.
app.lock.Lock()
defer app.lock.Unlock()
err := app.sm.Event(context.Background(), ev.GetEvent(), app,
ev.GetArgs())
+ var removeFrom *Context
+ if app.removeFromContext {
+ app.removeFromContext = false
+ removeFrom = app.context
+ }
// handle the same state transition not nil error (limit of fsm).
if err != nil && err.Error() != transitionErr {
- return err
+ return removeFrom, err
}
- return nil
+ return removeFrom, nil
}
func (app *Application) canHandle(ev events.ApplicationEvent) bool {
@@ -781,14 +800,12 @@ func (app *Application) flushReleaseableTasks() {
if app.areAllTasksTerminated() {
app.removeFromSchedulerCore()
- if app.context != nil {
- app.context.removeApplication(app.applicationID)
- }
+ app.removeFromContext = true
return
}
for _, task := range tasks {
- task.releaseAllocation(true)
+ task.forceReleaseAllocation()
}
}
diff --git a/pkg/cache/application_test.go b/pkg/cache/application_test.go
index 4f7fe911..afb5619d 100644
--- a/pkg/cache/application_test.go
+++ b/pkg/cache/application_test.go
@@ -1672,6 +1672,96 @@ func TestTaskMapReadersAreRaceFree(t *testing.T) {
}
}
+// TestAcceptRemovesApplicationUnderContextLock covers the removal of an
application that has no
+// live task left by the time the core accepts it. The delete used to run
under the application
+// lock only, while the scheduling loop lists the applications on every tick.
+func TestAcceptRemovesApplicationUnderContextLock(t *testing.T) {
+ context := initContextForTest()
+
+ const appCount = 50
+ apps := make([]*Application, 0, appCount)
+ for i := 0; i < appCount; i++ {
+ app := NewApplication(fmt.Sprintf("app-%d", i), "root.a",
"testuser", testGroups, map[string]string{}, newMockSchedulerAPI())
+ context.addApplicationToContext(app)
+ task := addTaskHelper(context, app, fmt.Sprintf("task-%d", i))
+ assert.NilError(t,
app.handle(NewSubmitApplicationEvent(app.applicationID)))
+ assert.NilError(t,
task.handle(NewSimpleTaskEvent(app.applicationID, task.taskID, CompleteTask)))
+ apps = append(apps, app)
+ }
+
+ start := make(chan struct{})
+ acceptErrs := make([]error, appCount)
+ var wg sync.WaitGroup
+ wg.Add(2)
+ go func() {
+ defer wg.Done()
+ <-start
+ for i, app := range apps {
+ acceptErrs[i] =
app.handle(NewSimpleApplicationEvent(app.applicationID, AcceptApplication))
+ }
+ }()
+ go func() {
+ defer wg.Done()
+ <-start
+ for i := 0; i < appCount; i++ {
+ context.GetAllApplications()
+ }
+ }()
+
+ close(start)
+ wg.Wait()
+
+ for _, err := range acceptErrs {
+ assert.NilError(t, err)
+ }
+ assert.Equal(t, len(context.GetAllApplications()), 0)
+}
+
+// TestDeferredReleaseReadsTaskFieldsUnderTaskLock covers the release replay
that runs when the
+// application still has live tasks. It read the task fields that the core
callback writes when
+// an allocation lands, without holding the task lock.
+func TestDeferredReleaseReadsTaskFieldsUnderTaskLock(t *testing.T) {
+ context := initContextForTest()
+ app := NewApplication(appID, "root.a", "testuser", testGroups,
map[string]string{}, newMockSchedulerAPI())
+ context.addApplicationToContext(app)
+
+ deferred := addTaskHelper(context, app, "task-deferred")
+ addTaskHelper(context, app, "task-running")
+
+ assert.NilError(t,
app.handle(NewSubmitApplicationEvent(app.applicationID)))
+ assert.NilError(t,
deferred.handle(NewSimpleTaskEvent(app.applicationID, deferred.taskID,
CompleteTask)))
+ assert.Equal(t, len(app.releaseableTasks), 1)
+
+ start := make(chan struct{})
+ var acceptErr error
+ var wg sync.WaitGroup
+ wg.Add(2)
+ go func() {
+ defer wg.Done()
+ <-start
+ acceptErr =
app.handle(NewSimpleApplicationEvent(app.applicationID, AcceptApplication))
+ }()
+ go func() {
+ defer wg.Done()
+ <-start
+ deferred.setAllocationKey("alloc-01")
+ }()
+
+ close(start)
+ wg.Wait()
+
+ assert.NilError(t, acceptErr)
+ assert.Equal(t, app.GetApplicationState(), ApplicationStates().Accepted)
+ assert.Assert(t, context.GetApplication(app.applicationID) != nil, "app
with a live task must stay in the cache")
+}
+
+func addTaskHelper(context *Context, app *Application, taskID string) *Task {
+ pod := &v1.Pod{ObjectMeta: apis.ObjectMeta{Name: taskID, UID:
types.UID(taskID)}}
+ task := NewTask(taskID, app, context, pod)
+ app.addTask(task)
+ return task
+}
+
func (ctx *Context) addApplicationToContext(app *Application) {
app.setContext(ctx)
ctx.lock.Lock()
diff --git a/pkg/cache/context.go b/pkg/cache/context.go
index 6e7e2fec..ee0cf9c0 100644
--- a/pkg/cache/context.go
+++ b/pkg/cache/context.go
@@ -1137,10 +1137,6 @@ func (ctx *Context) getApplication(appID string)
*Application {
func (ctx *Context) RemoveApplication(appID string) {
ctx.lock.Lock()
defer ctx.lock.Unlock()
- ctx.removeApplication(appID)
-}
-
-func (ctx *Context) removeApplication(appID string) {
if _, exist := ctx.applications[appID]; !exist {
log.Log(log.ShimContext).Debug("Attempted to remove
non-existent application", zap.String("appID", appID))
return
diff --git a/pkg/cache/task.go b/pkg/cache/task.go
index 20da78b6..61fb7afb 100644
--- a/pkg/cache/task.go
+++ b/pkg/cache/task.go
@@ -555,6 +555,14 @@ func (task *Task) releaseAllocation(force bool) {
}
}
+// forceReleaseAllocation releases the allocation for callers that do not
already hold the task
+// lock. The application lock is taken before the task lock on every path that
holds both.
+func (task *Task) forceReleaseAllocation() {
+ task.lock.Lock()
+ defer task.lock.Unlock()
+ task.releaseAllocation(true)
+}
+
func (task *Task) shouldAppRelease() bool {
task.lock.Unlock()
defer task.lock.Lock()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
