Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

refactor updatequeuejob thread logic in informer #624

Merged
merged 11 commits into from
Sep 9, 2023
Merged

refactor updatequeuejob thread logic in informer #624

Changes from 1 commit
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
eda97ac
remove preempt queue job thread
asm582 Sep 7, 2023
213e6bd
fix typo in comment
asm582 Sep 7, 2023
5947878
fix unit tests
asm582 Sep 7, 2023
1b2604d
add break condition
asm582 Sep 7, 2023
5e2dd28
do not requeue if AW does not exists
asm582 Sep 7, 2023
12109c0
merge
asm582 Sep 7, 2023
9135383
fix break condition
asm582 Sep 7, 2023
f20eda1
fix duplicate requeues
asm582 Sep 8, 2023
f821d18
address review
asm582 Sep 8, 2023
333b699
handle nil AW
asm582 Sep 8, 2023
43e2f6d
fix tests
asm582 Sep 9, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 22 additions & 9 deletions pkg/controller/queuejob/queuejob_controller_ex.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1613,7 +1613,7 @@ func (cc *XController) addQueueJob(obj interface{}) {
qj.Name, time.Now().Sub(qj.Status.ControllerFirstTimestamp.Time).Seconds(), qj.CreationTimestamp, qj.Status.ControllerFirstTimestamp)

klog.V(6).Infof("[Informer-addQJ] enqueue %s &qj=%p Version=%s Status=%+v", qj.Name, qj, qj.ResourceVersion, qj.Status)
cc.enqueue(qj)

// Requeue the item to be processed again in 30 seconds.
//TODO: tune the frequency of reprocessing an AW
hasCompletionStatus := false
Expand All @@ -1635,17 +1635,23 @@ func (cc *XController) addQueueJob(obj interface{}) {
go func() {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

time.AfterFunc seems a better construct to use here rather than go func() + time.Sleep.

On a side node, it seems it'll be valuable to move to used a delaying queue.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We need to wake the thread up every so often, from what I understand time.AfterFunc would wake up only once.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Right, maybe requeuing within the function passed to time.AfterFunc (possibly calling time.AfterFunc)?

for {
time.Sleep(requeueInterval)
latestAw, exists, err := cc.appwrapperInformer.Informer().GetStore().GetByKey(key)
latestObj, exists, err := cc.appwrapperInformer.Informer().GetStore().GetByKey(key)
if err != nil && !exists {
klog.Warningf("[Informer-addQJ] Recent copy of AW %s not found in cache", qj.Name)
} else {
if latestAw.(*arbv1.AppWrapper).Status.State != arbv1.AppWrapperStateActive && latestAw.(*arbv1.AppWrapper).Status.State != arbv1.AppWrapperStateEnqueued && latestAw.(*arbv1.AppWrapper).Status.State != arbv1.AppWrapperStateRunningHoldCompletion {
klog.V(2).Infof("[Informer-addQJ] Stopping requeue for AW %s with status %s", latestAw.(*arbv1.AppWrapper).Name, latestAw.(*arbv1.AppWrapper).Status.State)
var latestAw *arbv1.AppWrapper
if latestObj != nil {
latestAw = latestObj.(*arbv1.AppWrapper)
} else {
latestAw = qj
}
if latestAw.Status.State != arbv1.AppWrapperStateActive && latestAw.Status.State != arbv1.AppWrapperStateEnqueued && latestAw.Status.State != arbv1.AppWrapperStateRunningHoldCompletion {
klog.V(2).Infof("[Informer-addQJ] Stopping requeue for AW %s with status %s", latestAw.Name, latestAw.Status.State)
break //Exit the loop
}
// Enqueue the latest copy of the AW.
if (qj.Status.State != arbv1.AppWrapperStateCompleted && qj.Status.State != arbv1.AppWrapperStateFailed) && hasCompletionStatus {
cc.UpdateQueueJobs(latestAw.(*arbv1.AppWrapper))
cc.UpdateQueueJobs(latestAw)
klog.V(2).Infof("[Informer-addQJ] requeing AW to determine completion status for AW", qj.Name)
}

Expand All @@ -1665,23 +1671,30 @@ func (cc *XController) addQueueJob(obj interface{}) {
go func() {
for {
time.Sleep(requeueInterval)
latestAw, exists, err := cc.appwrapperInformer.Informer().GetStore().GetByKey(key)
latestObj, exists, err := cc.appwrapperInformer.Informer().GetStore().GetByKey(key)
if err != nil && !exists {
klog.Warningf("[Informer-addQJ] Recent copy of AW %s not found in cache", qj.Name)
} else {
if latestAw.(*arbv1.AppWrapper).Status.State != arbv1.AppWrapperStateActive && latestAw.(*arbv1.AppWrapper).Status.State != arbv1.AppWrapperStateEnqueued && latestAw.(*arbv1.AppWrapper).Status.State != arbv1.AppWrapperStateRunningHoldCompletion {
klog.V(2).Infof("[Informer-addQJ] Stopping requeue for AW %s with status %s", latestAw.(*arbv1.AppWrapper).Name, latestAw.(*arbv1.AppWrapper).Status.State)
var latestAw *arbv1.AppWrapper
if latestObj != nil {
latestAw = latestObj.(*arbv1.AppWrapper)
} else {
latestAw = qj
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I wonder what happens here to qj when we leave the scope of this function. This seems really dangerous. This for loop is going to keep retrying qj outside of addQueueJob since it is a thread. Does it retain a copy of qj? I'm not saying we should change it, I'm just concerned.

}
if latestAw.Status.State != arbv1.AppWrapperStateActive && latestAw.Status.State != arbv1.AppWrapperStateEnqueued && latestAw.Status.State != arbv1.AppWrapperStateRunningHoldCompletion {
klog.V(2).Infof("[Informer-addQJ] Stopping requeue for AW %s with status %s", latestAw.Name, latestAw.Status.State)
break //Exit the loop
}
// Enqueue the latest copy of the AW.
if (qj.Status.State != arbv1.AppWrapperStateCompleted && qj.Status.State != arbv1.AppWrapperStateFailed) && (qj.Spec.SchedSpec.MinAvailable > 0) {
cc.PreemptQueueJobs(latestAw.(*arbv1.AppWrapper))
cc.PreemptQueueJobs(latestAw)
klog.V(2).Infof("[Informer-addQJ] requeing AW to check minScheduling spec for AW", qj.Name)
}
}
}
}()
}
cc.enqueue(qj)
}

func (cc *XController) updateQueueJob(oldObj, newObj interface{}) {
Expand Down