1. 前言

转载请说明原文出处, 尊重他人劳动成果!

源码位置: https://github.com/nicktming/client-go/tree/tming-v13.0/tools/cache
分支: tming-v13.0 (基于v13.0版本)

本文将分析tools/cache包中的fifo. 主要会涉及到fifo.go, 该类在kube-scheduler中的scheduling_queue在没有开启pod优先级的时候会使用FIFIO.

2. 整体接口与实现类

architecture.png

可以看到Queue继承Store接口, 由于Queue是一个队列, 所以增加了Pop方法, 另外FIFO结构体是Queue接口的一个实现.

type FIFO struct {
    // 用于并发控制
    lock sync.RWMutex
    cond sync.Cond
    // We depend on the property that items in the set are in the queue and vice versa.

    // queue里面存的是key 并且有出队列的顺序
    // items里面存的是key与obj之间的对应关系 根据key可以找到obj key->obj
    items map[string]interface{}
    queue []string

    // populated is true if the first batch of items inserted by Replace() has been populated
    // or Delete/Add/Update was called first.
    populated bool
    // initialPopulationCount is the number of items inserted by the first call of Replace()
    // 第一次调用replace时候 加入到queue中的items的个数
    initialPopulationCount int

    // 生成key
    keyFunc KeyFunc

    // 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.
    closed     bool
    closedLock sync.Mutex
}
func NewFIFO(keyFunc KeyFunc) *FIFO {
    f := &FIFO{
        items:   map[string]interface{}{},
        queue:   []string{},
        keyFunc: keyFunc,
    }
    f.cond.L = &f.lock
    return f
}

这里需要特别注意一下populated和initialPopulationCount, 这两个参数在实现HasSynced()方法中会用到, 具体意义在那块进行说明.

3. 方法

Add 和 Update 和 AddIfNotPresent 和 Delete

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
    if _, exists := f.items[id]; !exists {
        f.queue = append(f.queue, id)
    }
    f.items[id] = obj
    f.cond.Broadcast()
    return nil
}
func (f *FIFO) Update(obj interface{}) error {
    return f.Add(obj)
}

func (f *FIFO) AddIfNotPresent(obj interface{}) error {
    id, err := f.keyFunc(obj)
    if err != nil {
        return KeyError{obj, err}
    }
    f.lock.Lock()
    defer f.lock.Unlock()
    f.addIfNotPresent(id, obj)
    return nil
}
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()
}
func (f *FIFO) Delete(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
    delete(f.items, id)
    return err
}

1. 可以看到add和update是一样的操作, 但是需要注意的是如果是更新操作, 也就是说该元素已经存在了, 此时只会更新item里面的obj, 而不会动该obj在queue中的位置.
2. AddIfNotPresent如果已经存在了, 就直接返回.
3. Delete方法可以看到只是从items中删除, 并没有从queue中删除该obj的key., 不过这不会有影响, 在pop方法的时候, 如果从queue里面出来的key在items中找不到, 就认为该obj已经删除了, 就不做处理了. 所以items里面的数据是安全的, queue里面的数据有可能是已经被删除了的.
4. 另外需要注意的是这些方法里面全部都直接设置了populated为true.

Replace

Replace的功能是删除所有的元素, 然后把list的元素全部加入到该FIFO的对象f中.

func (f *FIFO) Replace(list []interface{}, resourceVersion string) error {
    items := make(map[string]interface{}, len(list))
    for _, item := range list {
        key, err := f.keyFunc(item)
        if err != nil {
            return KeyError{item, err}
        }
        items[key] = item
    }

    f.lock.Lock()
    defer f.lock.Unlock()

    // 主要需要注意这里
    // f.populated为false的时候才会设置populated和initialPopulationCount
    // 1. 如果Add/Update/AddIfNotPresent/Delete比Replace先调用 不会进入到这里
    // 2. 如果Replace比Add/Update/AddIfNotPresent/Delete比Replace先调用 并且是第一次调用 会进入此代码块 
    //    后续再次Replace不会进入该代码块
    if !f.populated {
        f.populated = true
        f.initialPopulationCount = len(items)
    }

    // 更新f.items和queue
    f.items = items
    f.queue = f.queue[:0]
    for id := range items {
        f.queue = append(f.queue, id)
    }
    if len(f.queue) > 0 {
        f.cond.Broadcast()
    }
    return nil
}

这里主要需要注意关于populated的操作.
f.populated为false的时候才会设置populated和initialPopulationCount.
1. 如果Add/Update/AddIfNotPresent/Delete比Replace先调用 不会进入到代码块, 因为这种情况下populated已经被设置为true了.
2. 如果Replace比Add/Update/AddIfNotPresent/Delete先调用, 并且是第一次调用, 会进入此代码块,那么initialPopulationCount为第一次调用replace时加入的元素的个数. 如果后续再次Replace不会进入该代码块, 因为populated已经被设置为true, 没有别的地方会把populated设置为false.

pop

type PopProcessFunc func(interface{}) error
type ErrRequeue struct {
    // Err is returned by the Pop function
    Err error
}
var ErrFIFOClosed = errors.New("DeltaFIFO: manipulating with closed queue")
func (e ErrRequeue) Error() string {
    if e.Err == nil {
        return "the popped item should be requeued without returning an error"
    }
    return e.Err.Error()
}
func (f *FIFO) Pop(process PopProcessFunc) (interface{}, error) {
    f.lock.Lock()
    defer f.lock.Unlock()
    for {
        for len(f.queue) == 0 {
            // When the queue is empty, invocation of Pop() is blocked until new item is enqueued.
            // When Close() is called, the f.closed is set and the condition is broadcasted.
            // Which causes this loop to continue and return from the Pop().
            // 如果队列已经关闭 则直接返回错误
            if f.IsClosed() {
                return nil, ErrFIFOClosed
            }
            // 等待 有元素了之后会通知
            f.cond.Wait()
        }
        id := f.queue[0]
        f.queue = f.queue[1:]
        // 如果initialPopulationCount > 0 表明Replace是比Add/Update/AddIfNotPresent/Delete先调用 然后设置了initialPopulationCount
        if f.initialPopulationCount > 0 {
            f.initialPopulationCount--
        }
        item, ok := f.items[id]
        if !ok {
            // 如果已经删除了 不做处理
            // Item may have been deleted subsequently.
            continue
        }
        // 从items中删除id
        delete(f.items, id)
        // 调用用户自己的处理逻辑
        err := process(item)
        if e, ok := err.(ErrRequeue); ok {
            // 如果用户处理逻辑返回错误是ErrRequeue
            // 那么表明需要重新加回到queue里面去
            f.addIfNotPresent(id, item)
            err = e.Err
        }
        return item, err
    }
}

这里需要注意:
1. 如果initialPopulationCount > 0, 表明Replace是比Add/Update/AddIfNotPresent/Delete先调用 然后设置了initialPopulationCount就是第一次调用Replace中加入的元素个数, 那在pop中对于initialPopulationCount--做的操作就是每出来一个元素就减少一个, 等到initialPopulationCount=0的时候, 也就表明第一次调用replace加入的元素已经全部出队列了.
2. 从队列出来的元素有可能已经被删除了(也就是在items中无法找到), 不做任何处理. 因为Delete方法中删除元素只从items中删除了该元素, 该元素对应的key仍然还在queue中.

Resync

同步, 就是让items和queue中的数据保持一致.

func (f *FIFO) Resync() error {
    f.lock.Lock()
    defer f.lock.Unlock()

    inQueue := sets.NewString()
    for _, id := range f.queue {
        inQueue.Insert(id)
    }
    for id := range f.items {
        // 如果items里面有 queue里面没有
        if !inQueue.Has(id) {
            f.queue = append(f.queue, id)
        }
    }
    if len(f.queue) > 0 {
        f.cond.Broadcast()
    }
    return nil
}

整体的具体操作是把items里面有但是queue里面没有的元素加入到queue中.

HasSynced

判断是否sync.

func (f *FIFO) HasSynced() bool {
    f.lock.Lock()
    defer f.lock.Unlock()
    return f.populated && f.initialPopulationCount == 0
}

假设此时FIFQ刚刚初始化.
1. 如果啥方法都没有调用, 那么HasSynced返回false, 因为populated=false.
2. 如果先调用Add/Update/AddIfNotPresent/Delete后(后面调用什么函数都不用管了), 那么HasSynced返回true, 因为populated=true并且initialPopulationCount == 0.
3. 如果先调用Replace(后面调用什么函数都不用管了), 那么必须要等待该replace方法加入元素的个数全部pop之后, HasSynced才会返回true, 因为只有全部pop完了之后initialPopulationCount才减为0.