1. 前言

轉(zhuǎn)載請(qǐng)說明原文出處, 尊重他人勞動(dòng)成果!

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

本文將分析tools/cache包中的fifo. 主要會(huì)涉及到fifo.go, 該類在kube-scheduler中的scheduling_queue在沒有開啟pod優(yōu)先級(jí)的時(shí)候會(huì)使用FIFIO.

2. 整體接口與實(shí)現(xiàn)類

architecture.png

可以看到Queue繼承Store接口, 由于Queue是一個(gè)隊(duì)列, 所以增加了Pop方法, 另外FIFO結(jié)構(gòu)體是Queue接口的一個(gè)實(shí)現(xiàn).

type FIFO struct {
    // 用于并發(fā)控制
    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 并且有出隊(duì)列的順序
    // items里面存的是key與obj之間的對(duì)應(yīng)關(guān)系 根據(jù)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()
    // 第一次調(diào)用replace時(shí)候 加入到queue中的items的個(gè)數(shù)
    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, 這兩個(gè)參數(shù)在實(shí)現(xiàn)HasSynced()方法中會(huì)用到, 具體意義在那塊進(jìn)行說明.

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是一樣的操作, 但是需要注意的是如果是更新操作, 也就是說該元素已經(jīng)存在了, 此時(shí)只會(huì)更新item里面的obj, 而不會(huì)動(dòng)該obj在queue中的位置.
2. AddIfNotPresent如果已經(jīng)存在了, 就直接返回.
3. Delete方法可以看到只是從items中刪除, 并沒有從queue中刪除該obj的key., 不過這不會(huì)有影響, 在pop方法的時(shí)候, 如果從queue里面出來的key在items中找不到, 就認(rèn)為該obj已經(jīng)刪除了, 就不做處理了. 所以items里面的數(shù)據(jù)是安全的, queue里面的數(shù)據(jù)有可能是已經(jīng)被刪除了的.
4. 另外需要注意的是這些方法里面全部都直接設(shè)置了populated為true.

Replace

Replace的功能是刪除所有的元素, 然后把list的元素全部加入到該FIFO的對(duì)象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的時(shí)候才會(huì)設(shè)置populated和initialPopulationCount
    // 1. 如果Add/Update/AddIfNotPresent/Delete比Replace先調(diào)用 不會(huì)進(jìn)入到這里
    // 2. 如果Replace比Add/Update/AddIfNotPresent/Delete比Replace先調(diào)用 并且是第一次調(diào)用 會(huì)進(jìn)入此代碼塊 
    //    后續(xù)再次Replace不會(huì)進(jìn)入該代碼塊
    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
}

這里主要需要注意關(guān)于populated的操作.
f.populated為false的時(shí)候才會(huì)設(shè)置populated和initialPopulationCount.
1. 如果Add/Update/AddIfNotPresent/Delete比Replace先調(diào)用 不會(huì)進(jìn)入到代碼塊, 因?yàn)檫@種情況下populated已經(jīng)被設(shè)置為true了.
2. 如果Replace比Add/Update/AddIfNotPresent/Delete先調(diào)用, 并且是第一次調(diào)用, 會(huì)進(jìn)入此代碼塊,那么initialPopulationCount為第一次調(diào)用replace時(shí)加入的元素的個(gè)數(shù). 如果后續(xù)再次Replace不會(huì)進(jìn)入該代碼塊, 因?yàn)?code>populated已經(jīng)被設(shè)置為true, 沒有別的地方會(huì)把populated設(shè)置為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().
            // 如果隊(duì)列已經(jīng)關(guān)閉 則直接返回錯(cuò)誤
            if f.IsClosed() {
                return nil, ErrFIFOClosed
            }
            // 等待 有元素了之后會(huì)通知
            f.cond.Wait()
        }
        id := f.queue[0]
        f.queue = f.queue[1:]
        // 如果initialPopulationCount > 0 表明Replace是比Add/Update/AddIfNotPresent/Delete先調(diào)用 然后設(shè)置了initialPopulationCount
        if f.initialPopulationCount > 0 {
            f.initialPopulationCount--
        }
        item, ok := f.items[id]
        if !ok {
            // 如果已經(jīng)刪除了 不做處理
            // Item may have been deleted subsequently.
            continue
        }
        // 從items中刪除id
        delete(f.items, id)
        // 調(diào)用用戶自己的處理邏輯
        err := process(item)
        if e, ok := err.(ErrRequeue); ok {
            // 如果用戶處理邏輯返回錯(cuò)誤是ErrRequeue
            // 那么表明需要重新加回到queue里面去
            f.addIfNotPresent(id, item)
            err = e.Err
        }
        return item, err
    }
}

這里需要注意:
1. 如果initialPopulationCount > 0, 表明Replace是比Add/Update/AddIfNotPresent/Delete先調(diào)用 然后設(shè)置了initialPopulationCount就是第一次調(diào)用Replace中加入的元素個(gè)數(shù), 那在pop中對(duì)于initialPopulationCount--做的操作就是每出來一個(gè)元素就減少一個(gè), 等到initialPopulationCount=0的時(shí)候, 也就表明第一次調(diào)用replace加入的元素已經(jīng)全部出隊(duì)列了.
2. 從隊(duì)列出來的元素有可能已經(jīng)被刪除了(也就是在items中無法找到), 不做任何處理. 因?yàn)?code>Delete方法中刪除元素只從items中刪除了該元素, 該元素對(duì)應(yīng)的key仍然還在queue中.

Resync

同步, 就是讓items和queue中的數(shù)據(jù)保持一致.

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
}

假設(shè)此時(shí)FIFQ剛剛初始化.
1. 如果啥方法都沒有調(diào)用, 那么HasSynced返回false, 因?yàn)?code>populated=false.
2. 如果先調(diào)用Add/Update/AddIfNotPresent/Delete后(后面調(diào)用什么函數(shù)都不用管了), 那么HasSynced返回true, 因?yàn)?code>populated=true并且initialPopulationCount == 0.
3. 如果先調(diào)用Replace(后面調(diào)用什么函數(shù)都不用管了), 那么必須要等待該replace方法加入元素的個(gè)數(shù)全部pop之后, HasSynced才會(huì)返回true, 因?yàn)橹挥腥?code>pop完了之后initialPopulationCount才減為0.