Add a heap to client-go. Heap orders items with heap invariant ordering. #49752
Conversation
k8s-ci-robot
added
sig/scheduling
k8s-github-robot
added
size/XL
| _ = heap.Interface(&HeapData{}) // HeapData is a standard heap | ||
| ) | ||
|
|
||
| // LessFunc compares two objects and returns true if they first one should go |
| return obj | ||
| } | ||
|
|
||
| // -------------------------------------------------------------------- |
|
|
||
| // deleteItem removes the item from Heap.items and its key from Heap.queue. | ||
| func (h *Heap) deleteItem(key string) { | ||
| item := h.data.items[key] |
There was a problem hiding this comment.
Better to change to
item, ok := h.data.items[key]| } | ||
|
|
||
| // Pop waits until an item is ready. If multiple items are | ||
| // ready, they are returned in the order given by Heap.lessFunc. |
There was a problem hiding this comment.
s/Heap.lessFunc/Heap.data.lessFunc
|
Thanks, @dixudx for your review. |
|
|
||
| // addIfNotPresent assumes the lock is already held and adds the the provided | ||
| // item to the queue if it does not already exist. | ||
| func (h *Heap) addIfNotPresent(key string, obj interface{}) { |
|
|
||
| // Indication the queue is closed. | ||
| // Used to indicate a queue is closed so a control loop can exit when a queue is empty. | ||
| // Currently, not used to gate any of CRED operations. |
There was a problem hiding this comment.
I guess! Was copied from fifo.go and was not carrying much info anyways. I deleted the line.
| h.lock.Lock() | ||
| defer h.lock.Unlock() | ||
| if _, exists := h.data.items[key]; exists { | ||
| h.deleteItem(key) |
There was a problem hiding this comment.
Why not change the item and call Fix()? It's less expensive.
bsalamat
force-pushed
the
priority_scheduler
branch
3 times, most recently
from
f3f9d2d
to
5042dfe
Compare
|
@caesarxuchao Any more comments? |
| index int // The index of the object's key in the Heap.queue. | ||
| } | ||
|
|
||
| // HeapData is an internal struct that implements the standard heap interface |
There was a problem hiding this comment.
why make internal struct public? heapData?
bsalamat
force-pushed
the
priority_scheduler
branch
from
5042dfe
to
b9051aa
Compare
| } | ||
|
|
||
| // Push is supposed to be called by heap.Push only. | ||
| func (h *heapData) Push(key interface{}) { |
There was a problem hiding this comment.
the implementation here confuses me at the first look. it does not take care of the item map. not sure why leave that out.
There was a problem hiding this comment.
Heap is the type that manipulates heapData and is in charge of adding/removing elements to all the relevant arrays and maps. heapData implements the standard golang heap interface and the queue is the array that stores elements with the heap ordering. As a result, the heapData.Push and heapData.Pop only add and remove items from the queue. Removal of items from the items map is done by Heap.
In theory, Heap could implement standard golang heap, but it would require Heap to implement Push and Pop of the standard heap interface. In that case, those functions would be public. In order to hide those functions, I have added heapData.
There was a problem hiding this comment.
and is in charge of adding/removing elements to all the relevant arrays and maps.
isnt the backing array is managed inside heapdata, and manipulated by the go's heap pkg? it is confusing to manage the "pointer" in heapdata, and store the actual item in heapdata, but manage the actual item outside heapdata.
There was a problem hiding this comment.
Your point is valid. I made changes to Push and Pop of heapData to perform heapData.items manipulations.
| closedLock sync.Mutex | ||
| } | ||
|
|
||
| // Close the Heap. |
There was a problem hiding this comment.
what does close a heap mean?
There was a problem hiding this comment.
Pop blocks until there is an element in the queue or the heap is closed. Closing the heap signals the condition variable.
There was a problem hiding this comment.
probably add that to the comment. it was not clear why a heap needs to be closed.
| // When Close() is called, the h.closed is set and the condition is broadcast. | ||
| // Which causes this loop to continue and return from the Pop(). | ||
| if h.IsClosed() { | ||
| return nil, fmt.Errorf("Heap is closed") |
| @@ -0,0 +1,325 @@ | |||
| /* | |||
| Copyright 2014 The Kubernetes Authors. | |||
bsalamat
force-pushed
the
priority_scheduler
branch
4 times, most recently
from
e75545b
to
2d22d91
Compare
|
LGTM. Thanks! You need to ping the owners to review it again and get approved though :P |
|
Thanks, @xiang90! I have already spoken with Chao. He is very busy, but he will take a look soon. |
| // 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. | ||
| // | ||
| // This is useful in a single producer/consumer scenario so that the consumer can |
There was a problem hiding this comment.
Do you mean the consumer's requeue function should call AddIfNotPresent? Why it only useful in a single producer/consumer scenario? Isn't it useful in general?
There was a problem hiding this comment.
This is a comment copied from fifo.go. The comment doesn't say it is only useful in a single producer/consumer scenario, but if there are multiple consumers, the main benefit of it, which is not contending with the producer, goes away. Consider this scenario:
- Producer adds item: key: test, value: 1
- Consumer 1 takes the item with value 1 and processes it.
- Producer adds item: key: test, value: 2
- Consumer 2 takes the item with value 2 and processes it.
- Consumer 1 faces an error, requeues the stale item with value 1.
There was a problem hiding this comment.
Thanks for the explanation.
There was a problem hiding this comment.
Thanks for the explanation.
| func (h *Heap) deleteItem(key string) { | ||
| item, ok := h.data.items[key] | ||
| if ok { | ||
| heap.Remove(h.data, item.index) |
There was a problem hiding this comment.
Why not inline deleteItem in Delete and save one existence check?
| if h.IsClosed() { | ||
| return nil, fmt.Errorf("heap is closed") | ||
| } | ||
| h.cond.Wait() |
There was a problem hiding this comment.
Close() needs to be protected by h.lock. Otherwise if Close() is called between line 248 and 249, Wait() will sleep forever.
There was a problem hiding this comment.
And if Close() has to be proteced by h.lock anyway, then we can remove the h.closeLock.
There was a problem hiding this comment.
That's a very good point. Looks like fifo.go has the same problem. I used fifo.go as a template to implement heap.go.
I think Close() should hold both closedLock and lock. IsClosed() holds closedLock() only. It cannot hold the same lock as it is called inside Pop and after locking lock.
| ) | ||
|
|
||
| type LessFunc func(interface{}, interface{}) bool | ||
| type HeapItem struct { |
| old := h.queue | ||
| n := len(old) | ||
| key := old[n-1] | ||
| h.queue = old[0 : n-1] |
There was a problem hiding this comment.
Why not
key := h.queue(len(h.queue)-1)
h.queue = h.queue[:len(h.queue)-1]
| // This is an error | ||
| return nil | ||
| } | ||
| obj := item.obj |
There was a problem hiding this comment.
Why is this extra copy needed? Can we just return item.obj?
| func (h *Heap) Pop() (interface{}, error) { | ||
| h.lock.Lock() | ||
| defer h.lock.Unlock() | ||
| for { |
| h.lock.RLock() | ||
| defer h.lock.RUnlock() | ||
| item, exists := h.data.items[key] | ||
| var obj interface{} |
There was a problem hiding this comment.
Why not
if !exists {
return nil, false, nil
}
return item.obj, true, nil
| } | ||
|
|
||
| // TestHeap_Delete tests Heap.Delete and ensures that heap invariant is | ||
| // preserved after adding items. |
|
|
||
| // Indication the queue is closed. | ||
| // Used to indicate a queue is closed so a control loop can exit when a queue is empty. | ||
| closed bool |
There was a problem hiding this comment.
Shall we prevent adding more elements to the heap after it's closed? If you agree that's the expected behavior, then consider adding some tests.
There was a problem hiding this comment.
Makes sense. This is another example of following whatever fifo.go did. FIFO does not prevent addition and I didn't either, but I just added the checks and a test.
There was a problem hiding this comment.
The only downside of this check is that IsClosed holds a lock to check whether the heap is closed. Doing so before every addition to the heap hurts performance. That may be the reason fifo.go doesn't check it.
There was a problem hiding this comment.
I changed the code to directly check closed instead of calling IsClosed in order to avoid the extra lock/unlock.
| t.Errorf("unexpected item in the list: %v", key) | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
Could you add a test Pop() of an empty heap properly returns after Close() is called?
bsalamat
force-pushed
the
priority_scheduler
branch
from
2d22d91
to
bed1e5f
Compare
| ) | ||
|
|
||
| type LessFunc func(interface{}, interface{}) bool | ||
| type HeapItem struct { |
| old := h.queue | ||
| n := len(old) | ||
| key := old[n-1] | ||
| h.queue = old[0 : n-1] |
| // This is an error | ||
| return nil | ||
| } | ||
| obj := item.obj |
|
|
||
| // Indication the queue is closed. | ||
| // Used to indicate a queue is closed so a control loop can exit when a queue is empty. | ||
| closed bool |
There was a problem hiding this comment.
Makes sense. This is another example of following whatever fifo.go did. FIFO does not prevent addition and I didn't either, but I just added the checks and a test.
| // 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. | ||
| // | ||
| // This is useful in a single producer/consumer scenario so that the consumer can |
There was a problem hiding this comment.
This is a comment copied from fifo.go. The comment doesn't say it is only useful in a single producer/consumer scenario, but if there are multiple consumers, the main benefit of it, which is not contending with the producer, goes away. Consider this scenario:
- Producer adds item: key: test, value: 1
- Consumer 1 takes the item with value 1 and processes it.
- Producer adds item: key: test, value: 2
- Consumer 2 takes the item with value 2 and processes it.
- Consumer 1 faces an error, requeues the stale item with value 1.
| func (h *Heap) deleteItem(key string) { | ||
| item, ok := h.data.items[key] | ||
| if ok { | ||
| heap.Remove(h.data, item.index) |
| func (h *Heap) Pop() (interface{}, error) { | ||
| h.lock.Lock() | ||
| defer h.lock.Unlock() | ||
| for { |
| h.lock.RLock() | ||
| defer h.lock.RUnlock() | ||
| item, exists := h.data.items[key] | ||
| var obj interface{} |
| } | ||
|
|
||
| // TestHeap_Delete tests Heap.Delete and ensures that heap invariant is | ||
| // preserved after adding items. |
| t.Errorf("unexpected item in the list: %v", key) | ||
| } | ||
| } | ||
| } |
bsalamat
force-pushed
the
priority_scheduler
branch
2 times, most recently
from
a21b514
to
2c3b39e
Compare
|
ping @caesarxuchao |
| // 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. | ||
| // | ||
| // This is useful in a single producer/consumer scenario so that the consumer can |
There was a problem hiding this comment.
Thanks for the explanation.
| // When the queue is empty, invocation of Pop() is blocked until new item is enqueued. | ||
| // When Close() is called, the h.closed is set and the condition is broadcast. | ||
| // Which causes this loop to continue and return from the Pop(). | ||
| if h.IsClosed() { |
There was a problem hiding this comment.
Can you directly check h.closed here? Then IsClosed() can hold the lock, and closedLock can be removed.
There was a problem hiding this comment.
Otherwise lgtm. Thanks @bsalamat.
bsalamat
force-pushed
the
priority_scheduler
branch
from
2c3b39e
to
897f5b4
Compare
|
/lgtm |
k8s-ci-robot
added
the
lgtm
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: bsalamat, caesarxuchao Associated issue: 47604 The full list of commands accepted by this bot can be found here.
Needs approval from an approver in each of these OWNERS Files:
You can indicate your approval by writing |
k8s-github-robot
added
the
approved
|
@bsalamat please squash as you see proper. |
bsalamat
force-pushed
the
priority_scheduler
branch
from
897f5b4
to
68926a2
Compare
k8s-github-robot
removed
the
lgtm
|
Thanks, @caesarxuchao! Done. |
caesarxuchao
added
the
lgtm
|
/retest |
|
/test all [submit-queue is verifying that this PR is safe to merge] |
|
Automatic merge from submit-queue |
What this PR does / why we need it:
Heap is useful in implementing priority queues. Some components may need such ordering to process their highest priority objects first. Scheduler is going to be the first user of the heap. It will store pending pods ordered by their priority, so that the highest priority pods are popped first to be scheduled.
Which issue this PR fixes (optional, in
fixes #<issue number>(, fixes #<issue_number>, ...)format, will close that issue when PR gets merged): fixes #Special notes for your reviewer:
Release note:
ref/ #47604
ref/ #48646
@kubernetes/api-reviewers @kubernetes/sig-scheduling-pr-reviews @davidopp
/assign @caesarxuchao