关于kubernetes:kubernetes-fifo源码解析

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.lock

return 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