kubernetes fifo源码解析
1.介绍
kubernetes fifo是一个先入先出队列，实现了Add、Update、Delete、Get、Pop等根本API，以及Replace、HasSynced等API，具体如下：
type FIFO struct {
lock sync.RWMutex
cond sync.Cond
// key和obj的映射
items map[string]interface{}
// key的队列，去重
queue []string
// 当Delete/Add/Update被首先调用，或Replace()的items全副被pop时populated为true
populated bool
// Replace()首先被调用时的objs的数量
initialPopulationCount int
// keyFunc是用来将obj生成key的
keyFunc KeyFunc
// 队列是否敞开，用在Pop办法内的循环管制中
closed bool}
func NewFIFO(keyFunc KeyFunc) *FIFO
创立一个先入先出队列
func (f *FIFO) Add(obj interface{}) error
增加一个obj，当f.queue中已存在对应的key时，f.queue不再增加
func (f *FIFO) AddIfNotPresent(obj interface{}) error
当f.items不存在obj对应的key时才增加，这在繁多生产者/消费者有用，消费者能够平安的重试，防止与生产者竞争以及重入队已生产的item
func (f *FIFO) Close()
敞开队列
func (f *FIFO) Delete(obj interface{}) error
删除不存在f.queue中的item，因为这个实现假如使用者只关怀对象，而不关怀创立/增加对象的程序
func (f *FIFO) Get(obj interface{}) (item interface{}, exists bool, err error)
返回申请的item，不存在时exists为false
func (f *FIFO) GetByKey(key string) (item interface{}, exists bool, err error)
返回申请的item，不存在时exists为false
func (f *FIFO) HasSynced() bool
当Add/Update/Delete/AddIfNotPresent先被调用，或者先被Replace()插入的items都被Pop时，HasSynced返回true
func (f *FIFO) IsClosed() bool
检车队列是否敞开
func (f *FIFO) List() []interface{}
返回所有items.
func (f *FIFO) ListKeys() []string
返回以后FIFO中所有的key
func (f *FIFO) Pop(process PopProcessFunc) (interface{}, error)
Pop会等到f.queue中有对象，并且会调用PopProcessFunc解决item。如果f.queue中有多个待处理的对象，则将依照Add/Update的程序返回。在调用PopProcessFunc之前，会从队列(和存储)中删除item。如果PopProcessFunc返回ErrRequeue，会应用AddIfNotPresent()将其增加回来，因而保障可反复生产。PopProcessFunc是在锁定状态下调用的，因而在PopProcessFunc中操作FIFO的数据结构是平安的。
func (f *FIFO) Replace(list []interface{}, resourceVersion string) error
会依据list从新生成一个map，并将f.items指向新的map，根据该map从新入队f.queue，所以f.queue是无序的
func (f *FIFO) Resync() error
Resync会保障f.items中的key全副存在f.queue中，个别不应该调用该办法，因为其余api该当维持关联关系
func (f *FIFO) Update(obj interface{}) error
与Add实现统一
2.应用
参考TestFIFO_requeueOnPop[1]
// 取testFifoObject中name作为key
func testFifoObjectKeyFunc(obj interface{}) (string, error) {
return obj.(testFifoObject).name, nil}
type testFifoObject struct {
name string
val interface{}}
func mkFifoObj(name string, val interface{}) testFifoObject {
return testFifoObject{name: name, val: val}}
func TestFIFO_requeueOnPop(t *testing.T) {
// 创立FIFO实例
f := NewFIFO(testFifoObjectKeyFunc)
// 增加obj
f.Add(mkFifoObj("foo", 10))
// Pop操作，但返回ErrRequeue，这时会重入队
_, err := f.Pop(func(obj interface{}) error {
if obj.(testFifoObject).name != "foo" {
t.Fatalf("unexpected object: %#v", obj)
}
return ErrRequeue{Err: nil}
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
// GetByKey，还在队列中
if _, ok, err := f.GetByKey("foo"); !ok || err != nil {
t.Fatalf("object should have been requeued: %t %v", ok, err)
}
_, err = f.Pop(func(obj interface{}) error {
if obj.(testFifoObject).name != "foo" {
t.Fatalf("unexpected object: %#v", obj)
}
return ErrRequeue{Err: fmt.Errorf("test error")}
})
if err == nil || err.Error() != "test error" {
t.Fatalf("unexpected error: %v", err)
}
if _, ok, err := f.GetByKey("foo"); !ok || err != nil {
t.Fatalf("object should have been requeued: %t %v", ok, err)
}
// Pop操作，返回nil，不在队列中了
_, err = f.Pop(func(obj interface{}) error {
if obj.(testFifoObject).name != "foo" {
t.Fatalf("unexpected object: %#v", obj)
}
return nil
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
// GetByKey，不在队列中
if _, ok, err := f.GetByKey("foo"); ok || err != nil {
t.Fatalf("object should have been removed: %t %v", ok, err)
}}
3.源码解析
func NewFIFO(keyFunc KeyFunc) *FIFO {
f := &FIFO{
// key和obj的映射
items: map[string]interface{}{},
// key的队列，先入先出
queue: []string{},
// obj和key的映射函数
keyFunc: keyFunc,
}
// f.cond.L持有f.lock
f.cond.L = &f.lockreturn f
}
func (f *FIFO) Add(obj interface{}) error {
id, err := f.keyFunc(obj)
if err != nil {
return KeyError{obj, err}
}
f.lock.Lock()
defer f.lock.Unlock()
f.populated = true
// items中不存在时，才入队
if _, exists := f.items[id]; !exists {
f.queue = append(f.queue, id)
}
f.items[id] = obj
// 唤醒所有期待在f.cond的协程，其实就是Pop在期待f.cond
f.cond.Broadcast()
return nil}
func (f *FIFO) Pop(process PopProcessFunc) (interface{}, error) {
f.lock.Lock()
defer f.lock.Unlock()
for {
for len(f.queue) == 0 {
// 当队列为空时, 防止只有item入队时Pop才能够退出；当f.Close()调用时，Pop也能够退出
if f.closed {
return nil, ErrFIFOClosed
}
// 期待条件变量唤醒
f.cond.Wait()
}
// 从对头取，先入先出
id := f.queue[0]
f.queue = f.queue[1:]
// 当Replace先被调用时，initialPopulationCount才可能大于0
if f.initialPopulationCount > 0 {
f.initialPopulationCount--
}
item, ok := f.items[id]
if !ok {
// item有可能随后被删除，当被删除时不进行后续操作
continue
}
// 删除item
delete(f.items, id)
// 调用item处理函数，如果返回ErrRequeue时，重入队，以便反复生产
err := process(item)
if e, ok := err.(ErrRequeue); ok {
f.addIfNotPresent(id, item)
err = e.Err
}
return item, err
}}
func (f *FIFO) addIfNotPresent(id string, obj interface{}) {
f.populated = true
if _, exists := f.items[id]; exists {
return
}
f.queue = append(f.queue, id)
f.items[id] = obj
f.cond.Broadcast()}
4.总结
kubernetes fifo在实现先入先出队列上，值得咱们学习借鉴
援用链接
[1] TestFIFO_requeueOnPop: https://github.com/kubernetes/kubernetes/blob/v1.26.3/staging/src/k8s.io/client-go/tools/cache/fifo_test.go#L75
发表回复