pbacsko commented on code in PR #1062:
URL: https://github.com/apache/yunikorn-k8shim/pull/1062#discussion_r3766760128
##########
pkg/cache/task.go:
##########
@@ -343,45 +353,70 @@ func (task *Task) postTaskPending() {
// This routine binds the pod to the allocated node.
// It calls K8s api to bind a pod to the assigned node, this may need some
time,
// so we do a delay binding, background process, to avoid blocking main
process.
-// The result of the binding is tracked and failures are properly handled.
-// If successful, we move task to next state BOUND, otherwise we fail the task
+// Volume binding and pod binding are retried with a backoff; if they
ultimately fail
+// the allocation is rolled back to a pending ask so the core can re-schedule
the task
+// on a different node. On success we move the task to the next state BOUND.
func (task *Task) postTaskAllocated() {
go func() {
- // we need to obtain task's lock first,
- // this ensures no other threads modifying task state at the
time being
- task.lock.Lock()
- defer task.lock.Unlock()
+ // Snapshot the fields needed for binding without holding the
lock during the
+ // potentially slow bind calls. This avoids blocking other
goroutines and lets the
+ // rollback/reschedule path (which acquires the lock) run when
binding fails.
+ task.lock.RLock()
+ pod := task.pod
+ alias := task.alias
+ nodeName := task.nodeName
+ allocationKey := task.allocationKey
+ task.lock.RUnlock()
Review Comment:
Copies can be obtained before `go func() { ... }` because lock is hold
inside `postTaskAllocated()`. Not a big deal, but we can save a cycle of
locking:
```
func (task *Task) postTaskAllocated() {
// safe to copy, we have the mutex here
pod := task.pod (OR task.pod.DeepCopy() which might be even better)
alias := task.alias
nodeName := task.nodeName
go func(_, _, _) { // new signature
...
}(pod, alias, nodeName)
```
##########
pkg/cache/context.go:
##########
@@ -798,14 +797,6 @@ func (ctx *Context) bindPodVolumes(pod *v1.Pod) error {
zap.Error(err))
return err
}
- if volumes.StaticBindings == nil {
- // convert nil to empty array
- volumes.StaticBindings =
make([]*volumebinding.BindingInfo, 0)
- }
- if volumes.DynamicProvisions == nil {
- // convert nil to empty array
- volumes.DynamicProvisions =
make([]*volumebinding.DynamicProvision, 0)
- }
Review Comment:
Why was this removed? I think it's a regression and it causes an e2e test to
fail.
##########
pkg/cache/task.go:
##########
@@ -669,6 +712,22 @@ func (task *Task)
rollbackOnAssumePodFailure(allocationKey, nodeID string) {
zap.String("allocationKey", allocationKey))
}
+// rescheduleOnBindFailure is called when volume or pod binding fails after
all retries.
+// It moves the task from Allocated back to Scheduling and rolls the
allocation back to a
+// pending ask so the core can re-schedule it on a different node.
+// Must be called without holding the task lock.
+func (task *Task) rescheduleOnBindFailure(allocationKey, nodeID, eventReason,
eventMsg string) {
+ // Move the task back to Scheduling before releasing to the core, so
the re-delivered
+ // allocation (valid only from the Scheduling state) is accepted by the
state machine.
Review Comment:
Misleading comment, a better one:
```
// Move the task back to Scheduling before rolling back the allocation, so a
// subsequent TaskAllocated event from the core (which requires Scheduling)
is accepted.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
