Golang: Sync.Pool 源碼解析

Sync.Pool

Sync.Pool

需要提前了解GMPhttps://www.kancloud.cn/aceld/golang/1958305#2GolangGMP_2

簡單來說就是Goroutine(Go協(xié)程): Thread(線程): Process(調(diào)度器)

不在詳細(xì)展開了, 只針對Pool做一個簡單的分析

使用

package main

import "sync"

type Instance struct {
  Id string
}

func main() {
  // new a pool and Type is *Instance
  pool := sync.Pool{
    New: func() interface{} {
      return &Instance{Id: ""}
    },
  }
  
  // get from empty Pool
  ins := pool.Get().(*Instance)
  ins.Id = "1"
  
  // after usage, put back to pool
  pool.Put(ins)
  
  // check if same with var ins
  print(pool.Get().(*Instance).Id)
}

結(jié)構(gòu)體

Pool

type Pool struct {
    noCopy noCopy

    local     unsafe.Pointer // local fixed-size per-P pool, actual type is [P]poolLocal
    localSize uintptr        // size of the local array

    victim     unsafe.Pointer // local from previous cycle
    victimSize uintptr        // size of victims array

    // New optionally specifies a function to generate
    // a value when Get would otherwise return nil.
    // It may not be changed concurrently with calls to Get.
    New func() any
}

Pool是我們實(shí)際接觸的數(shù)據(jù), 其中包含了

  • local 是個數(shù)組，長度為 P 的個數(shù)。其元素類型是 poolLocal。這里面存儲著各個 P 對應(yīng)的本地對象池?？梢越频目醋?[P]poolLocal。
  • localSize。代表 local 數(shù)組的長度。因為 P 可以在運(yùn)行時通過調(diào)用 runtime.GOMAXPROCS 進(jìn)行修改, 因此我們還是得通過 localSize 來對應(yīng) local 數(shù)組的長度。
  • New 就是用戶提供的創(chuàng)建對象的函數(shù)。這個選項也不是必需。當(dāng)不填的時候，Get 有可能返回 nil。

poolLocal

// Local per-P Pool appendix.
type poolLocalInternal struct {
    private any       // Can be used only by the respective P.
    shared  poolChain // Local P can pushHead/popHead; any P can popTail.
}

type poolLocal struct {
    poolLocalInternal

    // Prevents false sharing on widespread platforms with
    // 128 mod (cache line size) = 0 .
    pad [128 - unsafe.Sizeof(poolLocalInternal{})%128]byte
}

poolLocal是和每個P綁定的一個存儲

  • private是為了能夠快速處理, 尤其是類似于經(jīng)常性的出現(xiàn)Get->Put->Get->Put時, 減少請求雙鏈表存儲的次數(shù).
  • shared有兩個作用, 第一是作為一個大容量的存儲, 第二是其他的P竊取.

這里稍微聊一下poolLocal的pad主要是為了加速CPU的Cache Line,使用的是緩存行Padding (謹(jǐn)慎使用)

不要忘記了, poolLocal是一個Core獨(dú)占的, 那么這個時候, 防止其他的碎片數(shù)據(jù)一起塞入緩存行就有作用了, 即不會因為碎片數(shù)據(jù)的頻繁更新而刷新Cache Line, 導(dǎo)致出現(xiàn)不命中導(dǎo)致的其他問題.

另外需要注意, 這里的poolChain是結(jié)構(gòu)體, 而不是指針, 原因是poolChain是一個非常短小的結(jié)構(gòu)體

type poolChain struct {
    head *poolChainElt
    tail *poolChainElt
}

方法

Put

func (p *Pool) Put(x any) {
    // 檢查x是否為空
    if x == nil {
        return
    }

/*  
    這部分不用看, race.Enabled是用來指示是否啟用了競態(tài)檢測器。當(dāng)你使用 -race 標(biāo)志編譯你的Go程序時，競態(tài)檢測器會被啟用，而這個變量會被設(shè)置為 true。
    通常只在調(diào)試或測試時使用，而不在生產(chǎn)環(huán)境下使用。

    if race.Enabled {
        if fastrandn(4) == 0 {
            // Randomly drop x on floor.
            return
        }
        race.ReleaseMerge(poolRaceAddr(x))
        race.Disable()
    }
*/
    // pin函數(shù)的作用我們稍后再聊, 簡單來說就是獲取和`P`綁定的poolLocal
    l, _ := p.pin()
    // 因為是在同一個線程下執(zhí)行的
    // 所以沒有一些互斥鎖等等

    // 如果能夠直接放入private, 則直接對private賦值
    if l.private == nil {
        l.private = x
    } else {
        // 塞入緩存隊列中.
        l.shared.pushHead(x)
    }
    // unpin操作
    runtime_procUnpin()
    if race.Enabled {
        race.Enable()
    }
}

Put主要有這么幾步

  1. 綁定G -> P
  2. 獲取P的id
  3. 根據(jù)pid獲取p對應(yīng)的poolLocal
  4. 優(yōu)先直接放入private, 其次放入緩存列表中

Get

func (p *Pool) Get() any {
/*
    if race.Enabled {
        race.Disable()
    }
*/
    // 綁定G -> P, 并返回P的poolLocal
    l, pid := p.pin()
    // 優(yōu)先使用private
    x := l.private
    // 不管最后有沒有用到, private一定是nil
    l.private = nil

    // 如果為nil
    if x == nil {
        // Try to pop the head of the local shard. We prefer
        // the head over the tail for temporal locality of
        // reuse.
        // 嘗試從緩存列表中得到一個
        // 嘗試從head彈出元素，而不是尾部，這種偏好是基于時間局部性原理。
        x, _ = l.shared.popHead()
        // 如果得不到
        if x == nil {
            // 嘗試從其他P中獲取, 如果沒法獲取, 則嘗試從victim中獲取
            x = p.getSlow(pid)
        }
    }
    runtime_procUnpin()
/*
    if race.Enabled {
        race.Enable()
        if x != nil {
            race.Acquire(poolRaceAddr(x))
        }
    }
*/
    // 如果沒有緩存中沒有任何對象, 但是有New函數(shù), 那么嘗試直接New一個
    if x == nil && p.New != nil {
        x = p.New()
    }
    // 返回x, 需要注意x存在為nil的可能性
    return x
}

總結(jié)

Get主要有這么幾步

  1. 綁定G -> P
  2. 獲取P的id
  3. 根據(jù)pid獲取p對應(yīng)的poolLocal
  4. 優(yōu)先使用private
  5. 其次使用自己的緩存列表
  6. 再次使用嘗試從其他的P里的poolLocal中獲取一個
  7. 還不行就從vivtim中復(fù)用一個
  8. 如果Pool里沒有任何可用對象, New

Other

pin

pin將當(dāng)前goroutine綁定到P，禁用搶占，并返回P的poolLocal池和P的id。
調(diào)用者在使用完池后必須調(diào)用runtime_procUnpin()。

<span style="font-weight: bold;" class="bold">實(shí)現(xiàn)</span>

func (p *Pool) pin() (*poolLocal, int) {
    // 調(diào)用runtime_procPin函數(shù)將當(dāng)前的goroutine固定到一個P上，并獲取該P(yáng)的id。
    pid := runtime_procPin()

    // 加載p.localSize和p.local
    s := runtime_LoadAcquintptr(&p.localSize) // load-acquire
    l := p.local                              // load-consume

    // 檢查pid是否小于p.localSize。
    if uintptr(pid) < s {
        // 如果是，則返回對應(yīng)的poolLocal和pid。
        return indexLocal(l, pid), pid
    }
    // 如果是，則返回對應(yīng)的poolLocal和pid。
    return p.pinSlow()
}

這里做一些解釋, pid通常是 0 - GOMAXPROCS的一個值, 用來標(biāo)記是哪一個線程

localSize的值一般來說就等于GOMAXPROCS

如果uintptr(pid) < s, 就代表著此時的poolLocal已經(jīng)被初始化過, 那么此時就可以直接返回.

反之就必須要做初始化的工作.

indexLocal

這個函數(shù)的作用比較簡單, 主要是根據(jù)原始指針 + pid偏移, 計算出真正的poolLocal

func indexLocal(l unsafe.Pointer, i int) *poolLocal {
    lp := unsafe.Pointer(uintptr(l) + uintptr(i)*unsafe.Sizeof(poolLocal{}))
    return (*poolLocal)(lp)
}

pinSlow

func (p *Pool) pinSlow() (*poolLocal, int) {
    // 調(diào)用pinSlow前, 必定調(diào)用了pin, 暫時釋放, 因為在pin的過程中, 無法對互斥鎖進(jìn)行操作
    runtime_procUnpin()
    // 初始化local, 禁止并發(fā)訪問
    allPoolsMu.Lock()
    defer allPoolsMu.Unlock()
    // 重新pin
    pid := runtime_procPin()
    // poolCleanup 在pin的過程中, GC不會調(diào)用
    // double check
    s := p.localSize
    l := p.local
    if uintptr(pid) < s {
        return indexLocal(l, pid), pid
    }
    // 為了GC時能夠管理所有的pool, 會將p放入管理隊列中
    if p.local == nil {
        allPools = append(allPools, p)
    }
    // If GOMAXPROCS changes between GCs, we re-allocate the array and lose the old one.
    // 獲取GOMAXPROCS, 表示最多有多少`P`
    size := runtime.GOMAXPROCS(0)
    // 初始化 pool
    local := make([]poolLocal, size)
    // 之所以只用atomic, 可能是為了防止在多核環(huán)境下出現(xiàn)cache不一致問題.
    atomic.StorePointer(&p.local, unsafe.Pointer(&local[0])) // store-release
    runtime_StoreReluintptr(&p.localSize, uintptr(size))     // store-release
    // 返回該`P`所對應(yīng)的poolLocal, 以及pid
    return &local[pid], pid
}

getSlow

func (p *Pool) getSlow(pid int) any {
    // 獲取p的數(shù)量
    size := runtime_LoadAcquintptr(&p.localSize) // load-acquire
    locals := p.local                            // load-consume
    // Try to steal one element from other procs.
    // 嘗試從其他的P的緩存中偷取一個
    for i := 0; i < int(size); i++ {
        l := indexLocal(locals, (pid+i+1)%int(size))
        if x, _ := l.shared.popTail(); x != nil {
            return x
        }
    }

    // Try the victim cache. We do this after attempting to steal
    // from all primary caches because we want objects in the
    // victim cache to age out if at all possible.
    // 嘗試從victim中復(fù)用一個
    size = atomic.LoadUintptr(&p.victimSize)
    // 如果victim中也沒有, 直接返回
    if uintptr(pid) >= size {
        return nil
    }
    locals = p.victim
    l := indexLocal(locals, pid)
    if x := l.private; x != nil {
        l.private = nil
        return x
    }
    for i := 0; i < int(size); i++ {
        l := indexLocal(locals, (pid+i)%int(size))
        if x, _ := l.shared.popTail(); x != nil {
            return x
        }
    }

    // Mark the victim cache as empty for future gets don't bother
    // with it.
    // 如果victim中也沒有, 直接將其標(biāo)記為0, 防止其他的P又一次的遍歷vivtim
    atomic.StoreUintptr(&p.victimSize, 0)

    return nil
}

poolCleanup

func poolCleanup() {
    // This function is called with the world stopped, at the beginning of a garbage collection.
    // It must not allocate and probably should not call any runtime functions.

    // Because the world is stopped, no pool user can be in a
    // pinned section (in effect, this has all Ps pinned).

    // Drop victim caches from all pools.
    for _, p := range oldPools {
        p.victim = nil
        p.victimSize = 0
    }

    // Move primary cache to victim cache.
    for _, p := range allPools {
        p.victim = p.local
        p.victimSize = p.localSize
        p.local = nil
        p.localSize = 0
    }

    // The pools with non-empty primary caches now have non-empty
    // victim caches and no pools have primary caches.
    oldPools, allPools = allPools, nil
}

這個函數(shù)會在GC是調(diào)用, 主要的作用就是將當(dāng)前的local刷到victim中, 之所以不直接清理掉, 是因為可能出現(xiàn)大量的New的情況.

可以參考: https://en.wikipedia.org/wiki/Victim_cache

我沒太明白, 但是在這里的使用, 我理解主要是用作一個緩沖和平滑作用.

拓展閱讀 poolChain

結(jié)構(gòu)體

poolChain

type poolChain struct {
    // head is the poolDequeue to push to. This is only accessed
    // by the producer, so doesn't need to be synchronized.
    // 只有生產(chǎn)者會操作head, 所以不存在并發(fā)競爭
    head *poolChainElt

    // tail is the poolDequeue to popTail from. This is accessed
    // by consumers, so reads and writes must be atomic.
    // 可能存在多個消費(fèi)者使用tail, 因此必須保證原子性的處理tail
    tail *poolChainElt
}

poolDequeue

type poolDequeue struct {
    // headTail packs together a 32-bit head index and a 32-bit
    // tail index. Both are indexes into vals modulo len(vals)-1.
    //
    // tail = index of oldest data in queue
    // head = index of next slot to fill
    //
    // Slots in the range [tail, head) are owned by consumers.
    // A consumer continues to own a slot outside this range until
    // it nils the slot, at which point ownership passes to the
    // producer.
    //
    // The head index is stored in the most-significant bits so
    // that we can atomically add to it and the overflow is
    // harmless.
    // 將head 和 tail 封裝成一個, 可以更好的保證原子性
    headTail uint64

    // vals is a ring buffer of interface{} values stored in this
    // dequeue. The size of this must be a power of 2.
    // vals[i].typ is nil if the slot is empty and non-nil
    // otherwise. A slot is still in use until *both* the tail
    // index has moved beyond it and typ has been set to nil. This
    // is set to nil atomically by the consumer and read
    // atomically by the producer.
    // 實(shí)際存儲的示例 (指針)
    vals []eface
}

poolChainElt

type poolChainElt struct {
    poolDequeue

    // next and prev link to the adjacent poolChainElts in this
    // poolChain.
    //
    // next is written atomically by the producer and read
    // atomically by the consumer. It only transitions from nil to
    // non-nil.
    //
    // prev is written atomically by the consumer and read
    // atomically by the producer. It only transitions from
    // non-nil to nil.
    next, prev *poolChainElt
}

eface

type eface struct {
    typ, val unsafe.Pointer
}

<span style="font-weight: bold;" class="bold">注意</span>

在 Go 語言中，interface{} 是一個空接口，它可以接受任何類型的值。

然而，當(dāng)我們需要存儲一個 interface{} 類型的值時，實(shí)際上我們需要存儲兩個信息：值的類型和值本身。

這是因為 Go 語言的接口是靜態(tài)類型的，即使是空接口也需要知道值的具體類型。

eface 結(jié)構(gòu)體被用來表示一個空接口的值。它有兩個字段：typ 和 val，分別用來存儲值的類型和值本身。

這樣做的好處是可以顯式地管理這兩個信息，而不是依賴 Go 語言的運(yùn)行時系統(tǒng)來管理。

這對于實(shí)現(xiàn)低級別的并發(fā)數(shù)據(jù)結(jié)構(gòu)（如poolChain的無鎖隊列）是非常有用的，因為這樣可以更精細(xì)地控制內(nèi)存的使用和同步。

此外，使用 unsafe.Pointer 可以避免 Go 語言的垃圾收集器誤判這些值仍然在使用。

當(dāng)一個值被從隊列中移除時，它的 typ 字段會被設(shè)置為 nil，這樣 Go 語言的垃圾收集器就知道這個值不再被使用，可以安全地回收它。

總的來說，使用 eface 結(jié)構(gòu)體而不是直接使用 interface{} 可以提供更精細(xì)的控制，這對于實(shí)現(xiàn)高效的并發(fā)數(shù)據(jù)結(jié)構(gòu)是非常重要的。



type poolChainElt struct {
    poolDequeue

    // next and prev link to the adjacent poolChainElts in this
    // poolChain.
    //
    // next is written atomically by the producer and read
    // atomically by the consumer. It only transitions from nil to
    // non-nil.
    //
    // prev is written atomically by the consumer and read
    // atomically by the producer. It only transitions from
    // non-nil to nil.
    next, prev *poolChainElt
}

type poolDequeue struct {
    // headTail packs together a 32-bit head index and a 32-bit
    // tail index. Both are indexes into vals modulo len(vals)-1.
    //
    // tail = index of oldest data in queue
    // head = index of next slot to fill
    //
    // Slots in the range [tail, head) are owned by consumers.
    // A consumer continues to own a slot outside this range until
    // it nils the slot, at which point ownership passes to the
    // producer.
    //
    // The head index is stored in the most-significant bits so
    // that we can atomically add to it and the overflow is
    // harmless.
    headTail uint64

    // vals is a ring buffer of interface{} values stored in this
    // dequeue. The size of this must be a power of 2.
    //
    // vals[i].typ is nil if the slot is empty and non-nil
    // otherwise. A slot is still in use until *both* the tail
    // index has moved beyond it and typ has been set to nil. This
    // is set to nil atomically by the consumer and read
    // atomically by the producer.
    vals []eface
}

方法

poolChain.pushHead

func (c *poolChain) pushHead(val any) {
    d := c.head
    // 如果為初始化
    if d == nil {
        // Initialize the chain.
        const initSize = 8 // Must be a power of 2
        d = new(poolChainElt)
        d.vals = make([]eface, initSize)
        c.head = d
        // 初始化一個, 并且原子性的設(shè)置head以及tail
        storePoolChainElt(&c.tail, d)
    }
    // 如果能夠?qū)懭氘?dāng)前的dequeue中, 則直接返回
    if d.pushHead(val) {
        return
    }

    // The current dequeue is full. Allocate a new one of twice
    // the size.
    // 如果dequeue已經(jīng)滿了
    // 第二層的dequeue直接翻倍
    newSize := len(d.vals) * 2
    // 一層最大有1 << 30個
    if newSize >= dequeueLimit {
        // Can't make it any bigger.
        newSize = dequeueLimit
    }

    // 初始化下一層
    d2 := &poolChainElt{prev: d}
    d2.vals = make([]eface, newSize)
    c.head = d2
    storePoolChainElt(&d.next, d2)
    // 剛初始化, 必定可以直接放入
    d2.pushHead(val)
}

當(dāng)head已經(jīng)填滿了, 就會生成的head, 同時修改當(dāng)前的head為創(chuàng)建的, 并將新創(chuàng)建的head的prev設(shè)為之前的head

poolChain.popHead

func (c *poolChain) popHead() (any, bool) {
    // 獲取當(dāng)前的head poolChainElt
    d := c.head
    for d != nil {
        // 嘗試從當(dāng)前的head的dequeue中pop一個
        if val, ok := d.popHead(); ok {
            return val, ok
        }
        // 如果沒法成功, 則嘗試
        // There may still be unconsumed elements in the
        // previous dequeue, so try backing up.
        d = loadPoolChainElt(&d.prev)
    }
    return nil, false
}

需要注意的是目前的實(shí)現(xiàn)會有一個問題, 那就是在批量push, 又批量pop時, 可能會頻繁的調(diào)用loadPoolChainElt(&d.prev)

另外一個問題是, 可能會導(dǎo)致c.head.prev中出現(xiàn)大量的空poolDequeue

對于一些比較大量的使用pool的程序來說, 可能會引入一些問題.

一個可能的做法是

func (c *poolChain) popHead() (any, bool) {
    d := c.head
    for d != nil {
        if val, ok := d.popHead(); ok {
            return val, ok
        }
        // There may still be unconsumed elements in the
        // previous dequeue, so try backing up.
        prev := loadPoolChainElt(&d.prev)

        // when try to load d.prev, that means the current dequeue is empty
        // try remove the current dequeue from the chain
        // and try to load the previous dequeue
        // atomic do this
        {
            d.prev.next = d.next
            d.next.prev = d.prev
            d.next = nil
            d.prev = nil
        }

        d = prev
    }
    return nil, false
}

但是很難在無鎖的情況下實(shí)現(xiàn), 而且可能引入更加復(fù)雜的處理.

popTail

func (c *poolChain) popTail() (any, bool) {
    // 獲取tail
    d := loadPoolChainElt(&c.tail)
    // 如果沒有tail, 即沒有初始化過, 直接返回
    if d == nil {
        return nil, false
    }

    for {
        // It's important that we load the next pointer
        // *before* popping the tail. In general, d may be
        // transiently empty, but if next is non-nil before
        // the pop and the pop fails, then d is permanently
        // empty, which is the only condition under which it's
        // safe to drop d from the chain.
        // 獲取d 的 next
        d2 := loadPoolChainElt(&d.next)
        // 嘗試popTail
        if val, ok := d.popTail(); ok {
            return val, ok
        }
        // 如果tail poolDequeue已經(jīng)空了, 則返回false
        if d2 == nil {
            // This is the only dequeue. It's empty right
            // now, but could be pushed to in the future.
            return nil, false
        }

        // The tail of the chain has been drained, so move on
        // to the next dequeue. Try to drop it from the chain
        // so the next pop doesn't have to look at the empty
        // dequeue again.
        // 
        if atomic.CompareAndSwapPointer((*unsafe.Pointer)(unsafe.Pointer(&c.tail)), unsafe.Pointer(d), unsafe.Pointer(d2)) {
            // We won the race. Clear the prev pointer so
            // the garbage collector can collect the empty
            // dequeue and so popHead doesn't back up
            // further than necessary.
            storePoolChainElt(&d2.prev, nil)
        }
        d = d2
    }
}

poolChain.popTail 是 poolChain 結(jié)構(gòu)體的一個方法，它的主要作用是從隊列的尾部移除并返回一個元素。如果隊列為空，它將返回 false。這個方法可以被任意數(shù)量的消費(fèi)者調(diào)用。

以下是 poolChain.popTail 方法的詳細(xì)步驟：

  1. 首先，它通過 loadPoolChainElt(&c.tail) 獲取隊列的尾部元素 d。如果 d 為 nil，則表示隊列為空，直接返回 nil, false。
  2. 然后，它進(jìn)入一個無限循環(huán)，嘗試從隊列的尾部彈出一個元素。在每次循環(huán)中，它首先在彈出尾部元素之前加載下一個指針 d2。這是因為在一般情況下，d 可能會暫時為空，但如果在彈出操作之前 next 是非 nil 的，并且彈出操作失敗，那么 d 就會永久性地為空。這是從鏈中刪除 d 的唯一條件。
  3. 接著，它嘗試通過 d.popTail() 方法從 d 的尾部彈出一個元素。如果成功，那么它將返回彈出的元素和 true。
  4. 如果 d2 為 nil，那么這就是唯一的隊列。雖然現(xiàn)在它是空的，但是未來可能會有元素被推入，所以返回 nil, false。
  5. 如果隊列的尾部已經(jīng)被清空，那么就移動到下一個隊列 d2。嘗試從鏈中刪除它，這樣下一次彈出操作就不必再看到空的隊列。這是通過 atomic.CompareAndSwapPointer 實(shí)現(xiàn)的。
  6. 如果我們贏得了比賽，那么就清除 d2.prev 指針，這樣垃圾收集器就可以收集空的隊列，而且 popHead 不必再回退更多。
  7. 最后，將 d 設(shè)置為 d2，然后進(jìn)入下一次循環(huán)。
image

注意

poolDequeue 是一個無鎖的固定大小的單生產(chǎn)者，多消費(fèi)者隊列。它有三個主要的方法：pushHead，popHead 和 popTail。

  1. pushHead(val any) bool 方法：

這個方法將一個元素添加到隊列的頭部。如果隊列已滿，它將返回 false。這個方法只能由單個生產(chǎn)者調(diào)用。

它首先獲取當(dāng)前的頭部和尾部索引，然后檢查隊列是否已滿。如果隊列已滿，它將返回 false。否則，它將獲取頭部索引對應(yīng)的槽位，并檢查該槽位是否已被 popTail 方法釋放。如果該槽位還未被釋放，那么隊列實(shí)際上仍然是滿的，因此返回 false。

如果頭部槽位是空的，那么生產(chǎn)者就擁有了這個槽位，并將值存入該槽位。然后，它將頭部索引加一，這將把槽位的所有權(quán)傳遞給 popTail 方法，并作為寫入槽位的存儲屏障。

  1. popHead() (any, bool) 方法：

這個方法從隊列的頭部移除并返回一個元素。如果隊列為空，它將返回 false。這個方法只能由單個生產(chǎn)者調(diào)用。

它首先獲取當(dāng)前的頭部和尾部索引，然后檢查隊列是否為空。如果隊列為空，它將返回 nil, false。否則，它將頭部索引減一，并嘗試獲取頭部索引對應(yīng)的槽位。如果成功，那么它就擁有了這個槽位，并從槽位中讀取出值。

然后，它將槽位清零。由于這個方法不與 pushHead 方法競爭，所以這里不需要特別小心。

  1. popTail() (any, bool) 方法：

這個方法從隊列的尾部移除并返回一個元素。如果隊列為空，它將返回 false。這個方法可以被任意數(shù)量的消費(fèi)者調(diào)用。

它首先獲取當(dāng)前的頭部和尾部索引，然后檢查隊列是否為空。如果隊列為空，它將返回 nil, false。否則，它將尾部索引加一，并嘗試獲取尾部索引對應(yīng)的槽位。如果成功，那么它就擁有了這個槽位，并從槽位中讀取出值。

然后，它將槽位的值字段清零，并將類型字段設(shè)置為 nil，這將把槽位的所有權(quán)傳遞給 pushHead 方法，并告訴 pushHead 方法這個槽位已經(jīng)不再使用，可以安全地回收它。

需要注意的是, popTail -> 完全釋放slot并不是一個原子性操作.

所以pushHead需要做兩個操作:

  1. 查看是否能夠獲取該槽
  2. 查看popTail是不是已經(jīng)釋放了該槽
  3. pushHead和popHead在同一時間只會有一個占用, 所以可以不考慮并發(fā)問題
最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容