Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Change method signature to not use interface{}
  • Loading branch information
ChristianZaccaria committed Aug 8, 2023
commit c592c590ccc906ffa0816bbd4fec3467290409f4
15 changes: 8 additions & 7 deletions pkg/controller/queuejob/heap.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,23 +19,24 @@ import (
"container/heap"
"fmt"
"k8s.io/client-go/tools/cache"
arbv1 "github.com/project-codeflare/multi-cluster-app-dispatcher/pkg/apis/controller/v1beta1"
)

// Below is the implementation of the a heap. The logic is pretty much the same
// as cache.heap, however, this heap does not perform synchronization. It leaves
// synchronization to the SchedulingQueue.

type LessFunc func(interface{}, interface{}) bool
type LessFunc func(qj1, qj2 *arbv1.AppWrapper) bool
type KeyFunc func(obj interface{}) (string, error)

type heapItem struct {
obj interface{} // The object which is stored in the heap.
obj *arbv1.AppWrapper // The object which is stored in the heap.
index int // The index of the object's key in the Heap.queue.
}

type itemKeyValue struct {
key string
obj interface{}
obj *arbv1.AppWrapper
}

// heapData is an internal struct that implements the standard heap interface
Expand Down Expand Up @@ -121,7 +122,7 @@ type Heap struct {

// Add inserts an item, and puts it in the queue. The item is updated if it
// already exists.
func (h *Heap) Add(obj interface{}) error {
func (h *Heap) Add(obj *arbv1.AppWrapper) error {
key, err := h.data.keyFunc(obj)
if err != nil {
return cache.KeyError{Obj: obj, Err: err}
Expand All @@ -136,7 +137,7 @@ func (h *Heap) Add(obj interface{}) error {
}

// BulkAdd adds all the items in the list to the queue.
func (h *Heap) BulkAdd(list []interface{}) error {
func (h *Heap) BulkAdd(list []*arbv1.AppWrapper) error {
for _, obj := range list {
key, err := h.data.keyFunc(obj)
if err != nil {
Expand All @@ -154,7 +155,7 @@ func (h *Heap) BulkAdd(list []interface{}) error {

// AddIfNotPresent inserts an item, and puts it in the queue. If an item with
// the key is present in the map, no changes is made to the item.
func (h *Heap) AddIfNotPresent(obj interface{}) error {
func (h *Heap) AddIfNotPresent(obj *arbv1.AppWrapper) error {
key, err := h.data.keyFunc(obj)
if err != nil {
return cache.KeyError{Obj: obj, Err: err}
Expand All @@ -167,7 +168,7 @@ func (h *Heap) AddIfNotPresent(obj interface{}) error {

// Update is the same as Add in this implementation. When the item does not
// exist, it is added.
func (h *Heap) Update(obj interface{}) error {
func (h *Heap) Update(obj *arbv1.AppWrapper) error {
return h.Add(obj)
}

Expand Down
2 changes: 1 addition & 1 deletion pkg/controller/queuejob/queuejob_controller_ex.go
Original file line number Diff line number Diff line change
Expand Up @@ -976,7 +976,7 @@ func (qjm *XController) ScheduleNext() {
pq := qjm.qjqueue.(*PriorityQueue)
if qjm.qjqueue.Length() > 0 {
for key, element := range pq.activeQ.data.items {
qjtemp := element.obj.(*arbv1.AppWrapper)
qjtemp := element.obj
klog.V(4).Infof("[ScheduleNext] AfterCalc: qjqLength=%d Key=%s index=%d Priority=%.1f SystemPriority=%.1f QueueJobState=%s",
qjm.qjqueue.Length(), key, element.index, float64(qjtemp.Spec.Priority), qjtemp.Status.SystemPriority, qjtemp.Status.QueueJobState)
}
Expand Down
3 changes: 2 additions & 1 deletion pkg/controller/queuejob/scheduling_queue.go
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ import (
"reflect"
"sync"

arbv1 "github.com/project-codeflare/multi-cluster-app-dispatcher/pkg/apis/controller/v1beta1"
qjobv1 "github.com/project-codeflare/multi-cluster-app-dispatcher/pkg/apis/controller/v1beta1"
"k8s.io/client-go/tools/cache"
"k8s.io/klog/v2"
Expand Down Expand Up @@ -320,7 +321,7 @@ func (p *PriorityQueue) Delete(qj *qjobv1.AppWrapper) error {
func (p *PriorityQueue) MoveAllToActiveQueue() {
p.lock.Lock()
defer p.lock.Unlock()
var unschedulableQJs []interface{}
var unschedulableQJs []*arbv1.AppWrapper
for _, qj := range p.unschedulableQ.pods {
unschedulableQJs = append(unschedulableQJs, qj)
}
Expand Down
4 changes: 2 additions & 2 deletions pkg/controller/queuejob/utils.go
Original file line number Diff line number Diff line change
Expand Up @@ -33,8 +33,8 @@ func GetXQJFullName(qj *arbv1.AppWrapper) string {
return qj.Name + "_" + qj.Namespace
}

func HigherSystemPriorityQJ(qj1, qj2 interface{}) bool {
return qj1.(*arbv1.AppWrapper).Status.SystemPriority > qj2.(*arbv1.AppWrapper).Status.SystemPriority
func HigherSystemPriorityQJ(qj1, qj2 *arbv1.AppWrapper) bool {
return qj1.Status.SystemPriority > qj2.Status.SystemPriority
}

func createAppWrapperKind(config *rest.Config) error {
Expand Down