快捷導(dǎo)航

go語(yǔ)言中的map如何解決散列性能下降

更新時(shí)間：2024年03月26日 09:13:10 作者：shark_chili

近期對(duì)go語(yǔ)言的map進(jìn)行深入了解和探究,其中關(guān)于map解決大量沖突的擴(kuò)容操作設(shè)計(jì)的十分巧妙,所以筆者特地整理了這篇文章來(lái)探討一下go語(yǔ)言中map如何解決散列性能下降,文中有相關(guān)的代碼示例供大家參考,需要的朋友可以參考下

寫(xiě)在文章開(kāi)頭

近期對(duì)go語(yǔ)言的map進(jìn)行深入了解和探究，其中關(guān)于map解決大量沖突的擴(kuò)容操作設(shè)計(jì)的十分巧妙，所以筆者特地整理了這篇文章來(lái)探討問(wèn)題。

hmap擴(kuò)容詳解

為什么需要擴(kuò)容

go語(yǔ)言中map是由無(wú)數(shù)個(gè)bucket構(gòu)成，假設(shè)某個(gè)哈希對(duì)應(yīng)的bucket空間已滿，則需要?jiǎng)?chuàng)建一個(gè)新的bmap存儲(chǔ)鍵值對(duì)，無(wú)數(shù)個(gè)bmap通過(guò)overflow指針進(jìn)行關(guān)聯(lián)。以下圖為例，假設(shè)我們需要查找key-111元素，就需要經(jīng)過(guò)以下幾個(gè)步驟:

通過(guò)哈希運(yùn)算定位到bucket[1]。
基于哈希值高位得到一個(gè)值tophash。
最終遍歷當(dāng)前bucket[1]中的所有數(shù)組，終于在第二個(gè)溢出桶找到key-111。

很明顯，如果極端情況下因?yàn)橛邢薜耐皩?dǎo)致大量的沖突就很可能使map元素定位的時(shí)間復(fù)雜度退化為O(n)，所以我們需要重新計(jì)算哈希值以及對(duì)桶進(jìn)行擴(kuò)容，從而解決極端的哈希沖突場(chǎng)景。

hmap擴(kuò)容過(guò)程

默認(rèn)情況下，map的進(jìn)行擴(kuò)容需要符合以下兩大條件之一：

未處于擴(kuò)容且鍵值對(duì)數(shù)超過(guò)bucket數(shù)以及當(dāng)前負(fù)載系統(tǒng)超過(guò)6.5。
未發(fā)生擴(kuò)容且溢出桶數(shù)量大于bucket數(shù)。

假設(shè)我們當(dāng)前hmap如下，所有key的算法都是通過(guò)哈希值的低3位和B(011)進(jìn)行取模運(yùn)算，因?yàn)榉仙鲜瞿硞€(gè)條件之一觸發(fā)了，需要進(jìn)行擴(kuò)容。

一般情況擴(kuò)容都是以原有bucket數(shù)*2，所以新的bucket數(shù)組長(zhǎng)度為16，創(chuàng)建新的數(shù)組空間后，hmap的bucket指針該指向這個(gè)新的數(shù)組，而原有bucket則交由oldbucket管理。也因?yàn)閿?shù)組長(zhǎng)度有8變?yōu)?code>16，所以記錄底數(shù)的變量B也由原來(lái)的3變?yōu)?(2^3變?yōu)?^4)，因?yàn)樵械臄?shù)組還有兩個(gè)空閑的溢出桶，所以新的數(shù)組也會(huì)創(chuàng)建同等數(shù)量的溢出交由管理空閑一處的指針extra.nextOverflow管理。

自此我們完成了擴(kuò)容操作，go語(yǔ)言中的map為了避免擴(kuò)容后大量遷移導(dǎo)致的計(jì)算耗時(shí)，對(duì)于舊有的bucket元素采用漸進(jìn)式再哈希的方式進(jìn)行遷移。所以后續(xù)我們?cè)谶M(jìn)行相同的寫(xiě)入操作時(shí)，若發(fā)現(xiàn)這個(gè)桶正處于擴(kuò)容狀態(tài)，那么它就會(huì)計(jì)算當(dāng)前桶中每個(gè)元素的新位置，然后一次性進(jìn)行驅(qū)逐。

以下圖為例，假設(shè)此時(shí)我們要修改key-111的鍵值對(duì)的值，在進(jìn)行修改的過(guò)程中，map就會(huì)通過(guò)哈希定位到舊的bucket的key-111的值，然后進(jìn)行修改，完成后基于全新的哈希算法(由%3改為%4)

源碼印證

我們現(xiàn)在通過(guò)一段代碼調(diào)試來(lái)了解一下go語(yǔ)言中的map時(shí)機(jī)的工作流程:

func main() {
 m := make(map[int]string)
 for i := 0; i < 9; i++ {
  m[i] = "xiaoming"
 }
 
}

上述這段代碼經(jīng)過(guò)編譯之后，實(shí)際調(diào)用的是mapassign_fast64這個(gè)方法：

0x0092 00146 (F:\github\awesomeProject\main.go:10)      CALL    runtime.mapassign_fast64(SB)

這里筆者就貼出這段代碼的核心片段，從核心代碼可以看出當(dāng)map符合進(jìn)行擴(kuò)容的條件時(shí)會(huì)調(diào)用hashGrow進(jìn)行擴(kuò)容，再通過(guò)goto again到again對(duì)應(yīng)代碼段進(jìn)行驅(qū)逐操作：

func mapassign_fast64(t *maptype, h *hmap, key uint64) unsafe.Pointer {
 // ......

again:
 bucket := hash & bucketMask(h.B)
 if h.growing() {
  growWork_fast64(t, h, bucket)
 }
 b := (*bmap)(add(h.buckets, bucket*uintptr(t.bucketsize)))

 var insertb *bmap
 var inserti uintptr
 var insertk unsafe.Pointer

 // ......

 //如果map未進(jìn)行擴(kuò)容，且當(dāng)前map負(fù)載超過(guò)最大值(默認(rèn)6.5)或者溢出桶數(shù)量超過(guò)了bucket數(shù)量則進(jìn)行擴(kuò)容
 if !h.growing() && (overLoadFactor(h.count+1, h.B) || tooManyOverflowBuckets(h.noverflow, h.B)) {
  hashGrow(t, h)
  goto again // Growing the table invalidates everything, so try again
 }

 // ......
}

hashGrow的擴(kuò)容邏輯和上圖的流程差不多，讀者可以基于筆者給出的核心注釋進(jìn)行了解：

func hashGrow(t *maptype, h *hmap) {
 //默認(rèn)B的值是+1
 bigger := uint8(1)
 
 //oldbuckets 記錄現(xiàn)在的bucket
 oldbuckets := h.buckets
 //創(chuàng)建一個(gè)新的bucket和溢出桶
 newbuckets, nextOverflow := makeBucketArray(t, h.B+bigger, nil)

 //更新b、oldbuckets 指針指向原有bucket、buckets 指向新創(chuàng)建的bucket
 h.B += bigger
 h.flags = flags
 h.oldbuckets = oldbuckets
 h.buckets = newbuckets
 h.nevacuate = 0
 h.noverflow = 0
 
 //如果原有的bucket還有空閑溢出桶，則記錄到h.extra.oldoverflow指針上
 if h.extra != nil && h.extra.overflow != nil {
  // Promote current overflow buckets to the old generation.
  if h.extra.oldoverflow != nil {
   throw("oldoverflow is not nil")
  }
  h.extra.oldoverflow = h.extra.overflow
  h.extra.overflow = nil
 }
 //如果剛剛有新創(chuàng)建的溢出桶，則用h.extra.nextOverflow指針進(jìn)行管理
 if nextOverflow != nil {
  if h.extra == nil {
   h.extra = new(mapextra)
  }
  h.extra.nextOverflow = nextOverflow
 }

 
}

進(jìn)行修改操作時(shí)會(huì)觸發(fā)的again代碼段的growWork_fast64方法其內(nèi)部就涉及了驅(qū)逐操作方法evacuate_fast64方法，因?yàn)橛辛松衔牡膱D解，所以對(duì)于這段代碼的解讀就很容易了，我們直接查看evacuate_fast64的核心注釋即可了解驅(qū)逐操作的含義：

func growWork_fast64(t *maptype, h *hmap, bucket uintptr) {
 //將對(duì)應(yīng)哈希的oldbucket的元素驅(qū)逐到新bucket上
 evacuate_fast64(t, h, bucket&h.oldbucketmask())

 //......
}



func evacuate_fast64(t *maptype, h *hmap, oldbucket uintptr) {
 //獲取舊的bucket的指針
 b := (*bmap)(add(h.oldbuckets, oldbucket*uintptr(t.bucketsize)))
 //......

 //通過(guò)[2]evacDst數(shù)組的0索引記錄oldbucket的bucket、keys、values指針，1記錄new buckets的bucket、keys、values指針
 if !evacuated(b) {
  
  var xy [2]evacDst
  x := &xy[0]
  //記錄oldbucket的bucket、keys、values指針
  x.b = (*bmap)(add(h.buckets, oldbucket*uintptr(t.bucketsize)))
  x.k = add(unsafe.Pointer(x.b), dataOffset)
  x.e = add(x.k, bucketCnt*8)

  //new buckets的bucket、keys、values指針
  if !h.sameSizeGrow() {
   // Only calculate y pointers if we're growing bigger.
   // Otherwise GC can see bad pointers.
   y := &xy[1]
   y.b = (*bmap)(add(h.buckets, (oldbucket+newbit)*uintptr(t.bucketsize)))
   y.k = add(unsafe.Pointer(y.b), dataOffset)
   y.e = add(y.k, bucketCnt*8)
  }
  //循環(huán)遍歷非空的tophash進(jìn)行驅(qū)逐操作
  for ; b != nil; b = b.overflow(t) {
   //獲取在old bucket上的地址
   k := add(unsafe.Pointer(b), dataOffset)
   e := add(k, bucketCnt*8)
   
   for i := 0; i < bucketCnt; i, k, e = i+1, add(k, 8), add(e, uintptr(t.elemsize)) {
    //計(jì)算在舊的bucket的tophash以定位鍵值對(duì)
    top := b.tophash[i]
    if isEmpty(top) {
     b.tophash[i] = evacuatedEmpty
     continue
    }
    if top < minTopHash {
     throw("bad map state")
    }
    //計(jì)算在新的bucket上的hash值
    var useY uint8
    if !h.sameSizeGrow() {
     // Compute hash to make our evacuation decision (whether we need
     // to send this key/elem to bucket x or bucket y).
     hash := t.hasher(k, uintptr(h.hash0))
     if hash&newbit != 0 {
      useY = 1
     }
    }

    b.tophash[i] = evacuatedX + useY // evacuatedX + 1 == evacuatedY, enforced in makemap
    //獲取新的bucket的指針地址
    dst := &xy[useY]                 // evacuation destination

   //到新bucke的tophash數(shù)組上的位置記錄這個(gè)tophash
    dst.b.tophash[dst.i&(bucketCnt-1)] = top // mask dst.i as an optimization, to avoid a bounds check

    //將舊的bucket的key復(fù)制到新bucket上
    if t.key.ptrdata != 0 && writeBarrier.enabled {
     if goarch.PtrSize == 8 {
      // 通過(guò)指針操作，將舊的key復(fù)制到新的bucket的key上
      *(*unsafe.Pointer)(dst.k) = *(*unsafe.Pointer)(k)
     } else {
      // There are three ways to squeeze at least one 32 bit pointer into 64 bits.
      // Give up and call typedmemmove.
      typedmemmove(t.key, dst.k, k)
     }
    } else {
     *(*uint64)(dst.k) = *(*uint64)(k)
    }
    //復(fù)制value到新bucket的位置上
    typedmemmove(t.elem, dst.e, e)
    //....
   }
  }
  // 完成后刪除舊的bucket的鍵值對(duì)，輔助GC
  if h.flags&oldIterator == 0 && t.bucket.ptrdata != 0 {
   b := add(h.oldbuckets, oldbucket*uintptr(t.bucketsize))
   // Preserve b.tophash because the evacuation
   // state is maintained there.
   ptr := add(b, dataOffset)
   n := uintptr(t.bucketsize) - dataOffset
   memclrHasPointers(ptr, n)
  }
 }

 if oldbucket == h.nevacuate {
  advanceEvacuationMark(h, t, newbit)
 }
}

經(jīng)過(guò)這段代碼之后，我們的舊的bucket上的桶的數(shù)據(jù)也就都驅(qū)逐完成了，隨后跳出這些代碼段，再次回到mapassign_fast64的賦值操作的代碼找到合適key的位置設(shè)置鍵值對(duì)。

如下所示，可以看到核心步驟大致是:

for循環(huán)定位到哈希算法得出的tophash對(duì)應(yīng)key的位置。
如果找到空位置則跳出循環(huán)進(jìn)行鍵值對(duì)設(shè)置，并更新count值。
若找到和當(dāng)前key相等的位置，直接跳到done代碼段設(shè)置value。

func mapassign_fast64(t *maptype, h *hmap, key uint64) unsafe.Pointer {

//......

//定位到key的指針
bucketloop:
 for {
  for i := uintptr(0); i < bucketCnt; i++ {
   //若定位到的bucket為nil,則說(shuō)明這個(gè)位置沒(méi)有被使用過(guò)，則進(jìn)入該分支定位指針
   if isEmpty(b.tophash[i]) {
    if insertb == nil {
     insertb = b
     inserti = i
    }
    //如果找到當(dāng)前tophash為0值，說(shuō)明這個(gè)位置沒(méi)有被用過(guò)，直接退出循環(huán)，到外部直接賦值
    if b.tophash[i] == emptyRest {
     break bucketloop
    }
    continue
   }
   //獲取k判斷和當(dāng)前key值是否一致，若不一致的continue，反之記錄直接到done代碼段，直接設(shè)置value的值
   k := *((*uint64)(add(unsafe.Pointer(b), dataOffset+i*8)))
   if k != key {
    continue
   }
   insertb = b
   inserti = i
   goto done
  }
  //.....
 }

 
 //定位到key位置設(shè)置key
 insertk = add(unsafe.Pointer(insertb), dataOffset+inserti*8)
 // store new key at insert position
 *(*uint64)(insertk) = key
 
 //元素?cái)?shù)量+1
 h.count++

done:
 //設(shè)置value
 elem := add(unsafe.Pointer(insertb), dataOffset+bucketCnt*8+inserti*uintptr(t.elemsize))
 if h.flags&hashWriting == 0 {
  fatal("concurrent map writes")
 }
 h.flags &^= hashWriting
 return elem

}

擴(kuò)容未完成時(shí)如何讀

因?yàn)閙ap是在操作時(shí)進(jìn)行驅(qū)逐操作的，所以在讀取時(shí)需要會(huì)按照以下步驟執(zhí)行：

定位到當(dāng)前hash對(duì)應(yīng)bucket設(shè)置給指針b。
查看當(dāng)前hash對(duì)應(yīng)oldbucket是否發(fā)生驅(qū)逐操作，若未發(fā)生則說(shuō)明當(dāng)前的值在oldbucket上，將指針b指向oldbucket。
到b指針的bucket上查到對(duì)應(yīng)的key，若找到則返回value。

對(duì)應(yīng)的核心代碼如下，讀者可參照注釋了解大體流程：

func mapaccess1(t *maptype, h *hmap, key unsafe.Pointer) unsafe.Pointer {
 //......
 //定位hash值
 hash := t.hasher(key, uintptr(h.hash0))
 //先定位到bucket指針
 m := bucketMask(h.B)
 b := (*bmap)(add(h.buckets, (hash&m)*uintptr(t.bucketsize)))
 
 //在基于hash到oldbucket定位舊的bucket指針，若未發(fā)生驅(qū)逐則上b指針指向oldbucket
 if c := h.oldbuckets; c != nil {
  if !h.sameSizeGrow() {
   // There used to be half as many buckets; mask down one more power of two.
   m >>= 1
  }
  oldb := (*bmap)(add(c, (hash&m)*uintptr(t.bucketsize)))
  //若未發(fā)生驅(qū)逐則上b指針指向oldbucket
  if !evacuated(oldb) {
   b = oldb
  }
 }
 top := tophash(hash)
bucketloop:
 //循環(huán)定位鍵值對(duì)
 for ; b != nil; b = b.overflow(t) {
  for i := uintptr(0); i < bucketCnt; i++ {
   //......
   
   //如果找到的key和我們要查詢(xún)的key相等則直接返回
   if t.key.equal(key, k) {
    e := add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.elemsize))
    if t.indirectelem() {
     e = *((*unsafe.Pointer)(e))
    }
    return e
   }
  }
 }
 return unsafe.Pointer(&zeroVal[0])
}