Skip to content

map

go 與其它語言不同的是,映射表的支持是由map關鍵字提供的,而非將其封裝為標准庫。映射表是一種使用場景非常多的數據結構,底層有著許多的實現方式,最常見的兩種方式就是紅黑樹和哈希表,go 采用的是哈希表實現方式。

TIP

map 的實現中涉及到了大量的指針移動操作,所以閱讀本文需要unsafe標准庫的知識。

內部結構

runtime.hmap結構體就是代表著 go 中的map,與切片一樣map的內部實現也是結構體。

go
// A header for a Go map.
type hmap struct {
  // Note: the format of the hmap is also encoded in cmd/compile/internal/reflectdata/reflect.go.
  // Make sure this stays in sync with the compiler's definition.
  count     int // # live cells == size of map.  Must be first (used by len() builtin)
  flags     uint8
  B         uint8  // log_2 of # of buckets (can hold up to loadFactor * 2^B items)
  noverflow uint16 // approximate number of overflow buckets; see incrnoverflow for details
  hash0     uint32 // hash seed

  buckets    unsafe.Pointer // array of 2^B Buckets. may be nil if count==0.
  oldbuckets unsafe.Pointer // previous bucket array of half the size, non-nil only when growing
  nevacuate  uintptr        // progress counter for evacuation (buckets less than this have been evacuated)

  extra *mapextra // optional fields
}

英文注釋已經說明的很清晰了,下面對比較重要的字段進行一些簡單的解釋

  • count,表示 hamp 中的元素數量,結果等同於len(map)

  • flags,hmap 的標志位,用於表示 hmap 處於什麼狀態,有以下幾種可能。

    go
    const (
        iterator     = 1 // 迭代器正在使用桶
        oldIterator  = 2 // 迭代器正在使用舊桶
        hashWriting  = 4 // 一個協程正在寫入hmap
        sameSizeGrow = 8 // 正在等量擴容
    )
  • B,hmap 中的哈希桶的數量為1 << B

  • noverflow,hmap 中溢出桶的大致數量。

  • hash0,哈希種子,在 hmap 被創建時指定,用於計算哈希值。

  • buckets,存放哈希桶數組的指針。

  • oldbuckets,存放 hmap 在擴容前哈希桶數組的指針。

  • extra,存放著 hmap 中的溢出桶,溢出桶指的是就是當前桶已經滿了,創建新的桶來存放元素,新創建的桶就是溢出桶。

hamp 中的buckets也就是桶切片指針,在 go 中對應的結構為runtime.bmap,如下所示

go
// A bucket for a Go map.
type bmap struct {
  tophash [bucketCnt]uint8
}

從上面可以看到它只有一個tophash的字段,該字段是用於存放每個鍵的高八位,不過實際上來說,bmap的字段不止這些,這是因為map可以存儲各種類型的鍵值對,所以需要在編譯時根據類型來推導佔用的內存空間,在cmd/compile/internal/reflectdata/reflect.go中的MapBucketType函數的功能就是在編譯時構造 bmap,它會進行一系列檢查工作,比如 key 的類型是否comparable

go
// MapBucketType makes the map bucket type given the type of the map.
func MapBucketType(t *types.Type) *types.Type

所以實際上,bmap的結構如下,不過這些字段對我們是不可見的,go 在實際操作中是通過移動 unsafe 指針來進行訪問

go
type bmap struct {
  tophash [BUCKETSIZE]uint8
  keys [BUCKETSIZE]keyType
  elems [BUCKETSIZE]elemType
  overflow *bucket
}

其中的一些解釋如下

  • tophash,存放每一個鍵的高八位值,對於一個 tophash 的元素而言,有下面幾種特殊的值

    go
    const (
        emptyRest      = 0 // 當前元素是空的,並且該元素後面也沒有可用的鍵值了
        emptyOne       = 1 // 當前元素是空的,但是該元素後面有可用的鍵值。
        evacuatedX     = 2 // 擴容時出現,只能出現在oldbuckets中,表示當前元素被搬遷到了新哈希桶數組的上半區
        evacuatedY     = 3 // 擴容時出現只能出現在oldbuckets中,表示當前元素被搬遷到了新哈希桶數組的下半區
        evacuatedEmpty = 4 // 擴容時出現,元素本身就是空的,在搬遷時被標記
        minTopHash     = 5 // 對於一個正常的鍵值來說tophash的最小值
    )

    只要是tophash[i]的值大於minTophash的值,就說明對應下標存在正常的鍵值。

  • keys,存放指定類型鍵的數組。

  • elems,存放指定類型值的數組。

  • overflow,指向溢出桶的指針。

既然鍵值無法通過結構體字段來直接訪問,為此 go 事先聲明了一個常量dataoffset,它代表的是數據在bmap中的內存偏移量。

go
const dataOffset = unsafe.Offsetof(struct {
    b bmap
    v int64
}{}.v)

實際上,鍵值是存放在一個連續的內存地址中,類似於下面這種結構,這樣做是為了避免內存對齊帶來的空間浪費。

k1,k2,k3,k4,k5,k6...v1,v2,v3,v4,v5,v6...

所以對於一個 bmap 而言,指針移動dataoffset後,移動i*sizeof(keyType)就是第 i 個 key 的地址

go
k := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.KeySize))

獲取第 i 個 value 的地址也是同理

go
e := add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.KeySize)+i*uintptr(t.ValueSize))

hamp中的buckets指針,就是指向的第一個哈希桶的地址,如果想要獲取第 i 個哈希桶的地址,偏移量就是i*sizeof(bucket)

go
b := (*bmap)(add(h.buckets, i*uintptr(t.BucketSize)))

在後續的內容中,這些操作會非常頻繁的出現。

哈希

沖突

在 hmap 中,有一個字段extra專門用來存放溢出桶的信息,它會指向存放溢出桶的切片,其結構如下。

go
type mapextra struct {
  // 溢出桶的指針切片
  overflow    *[]*bmap
  // 擴容前舊的溢出桶的指針切片
  oldoverflow *[]*bmap
  // 指向下一個空閒的溢出桶的指針
  nextOverflow *bmap
}

TIP

在上圖中,藍色部分是哈希桶數組,橙色部分是溢出哈希桶數組,溢出哈希桶下面統稱為溢出桶。

上圖就可以比較好的展示 hmap 的大致結構,buckets指向哈希桶數組,extra指向溢出桶數組,桶bucket0指向溢出桶overflow0,兩種不同的桶分別存放在兩個切片中,兩種桶的內存都是連續的。當兩個鍵通過哈希後被分配到了同一個 bucket,這種情況就是發生了哈希沖突。go 中解決哈希沖突的方式就是拉鏈法,當發生沖突的鍵的數量大於桶的容量後,一般是 8 個,其值取決於internal/abi.MapBucketCount。然後就會創建一個新的桶來存放這些鍵,而這個桶就叫溢出桶,意為原來的桶裝不下了,元素溢出到這個新桶裡來了,創建完畢後,哈希桶會有一個指針指向新的溢出桶,這些桶的指針連起來就形成了一個bmap鏈表。

對於拉鏈法而言,使用負載因子用於衡量哈希表的沖突情況,其計算公式如下

go
loadfactor := len(elems) / len(buckets)

當負載因子越大時,說明哈希沖突越多,也就是溢出桶的數量越多,那麼在讀寫哈希表時,就需要遍歷更多的溢出桶鏈表,才能找到指定的位置,所以性能就越差。為了改善這種情況,應該增加buckets桶的數量,也就是擴容,對於 hmap 而言,有兩種情況會觸發擴容

  • 負載因子超過閾值bucketCnt*13/16,其值至少是 6.5。
  • 溢出桶數量過多

當負載因子越小時,說明 hmap 的內存利用率低,佔用的內存就越大。go 中用於計算負載因子的函數是runtime.overLoadFactor,代碼如下

go
func overLoadFactor(count int, B uint8) bool {
  return count > bucketCnt && uintptr(count) > loadFactorNum*(bucketShift(B)/loadFactorDen)
}

其中loadFactorNumloadFactorDen都是一個常數,bucketshift是計算1 << B,並且已知

go
loadFactorNum = (bucketCnt * 13 / 16) * loadFactorDen

所以化簡一下就能得到

go
count > bucketCnt && uintptr(count) / 1 << B > (bucketCnt * 13 / 16)

其中(bucketCnt * 13 / 16)值為 6.5,1 << B就是哈希桶的數量,所以該函數的作用就是計算元素數量除以桶的數量值是否大於負載因子 6.5。

計算

go 內部計算哈希的函數位於runtime/alg.go文件中的f32hash,如下所示,包含了對 NaN 和 0 兩種情況的處理。

go
func f32hash(p unsafe.Pointer, h uintptr) uintptr {
  f := *(*float32)(p)
  switch {
  case f == 0:
    return c1 * (c0 ^ h) // +0, -0
  case f != f:
    return c1 * (c0 ^ h ^ uintptr(fastrand())) // any kind of NaN
  default:
    return memhash(p, h, 4)
  }
}

可以看到的是,map 哈希計算方法並不是基於類型,而是基於內存,最終會走到memhash函數,該函數由匯編實現,邏輯位於runtime/asm*.s中。基於內存的哈希值不應該被持久化保存,因為它應該只在運行時被使用,不可能保證每一次運行內存分布都完全一致。

在該文件中還有一個名為typehash的函數,該函數會根據不同類型來計算哈希值,不過 map 並不會使用這個函數來計算哈希。

func typehash(t *_type, p unsafe.Pointer, h uintptr) uintptr

這種實現相比上面那種更慢,但是更通用些,主要用於反射以及編譯時的函數生成,比如下面這個函數。

go
//go:linkname reflect_typehash reflect.typehash
func reflect_typehash(t *_type, p unsafe.Pointer, h uintptr) uintptr {
  return typehash(t, p, h)
}

創建

map 的初始化有兩種方式,這一點已經在語言入門中闡述過了,一種是使用關鍵字map直接創建,另一種是使用make函數,不管用何種方式初始化,最後都是由runtime.makemap來創建 map,該函數簽名如下

go
func makemap(t *maptype, hint int, h *hmap) *hmap

其中的參數

  • t,指的是 map 的類型,不同的類型所需的內存佔用不同
  • hint,指的是make函數的第二個參數,map 預計元素的容量。
  • h,指的是hmap的指針,可以為nil

返回值就是初始化完畢的hmap指針。該函數在初始化過程中有幾個主要的工作。首先就是計算預計分配的內存是否會超出最大分配內存,對應如下代碼

go
// 將預計容量與桶類型的內存大小相乘
mem, overflow := math.MulUintptr(uintptr(hint), t.Bucket.Size_)
// 數值溢出或者超出了最大分配內存
if overflow || mem > maxAlloc {
    hint = 0
}

在先前的內部結構中已經提到過,hmap 內部是由桶組成的,在內存利用率最低的情況下,一個桶只有一個元素,佔用的內存最多,所以預計的最大內存佔用就是元素數量乘以對應類型的內存佔用大小。當計算結果數值溢出了,或者超出了最大能分配的內存,就將 hint 置為 0,因為後續需要用 hint 來計算桶數組的容量。

第二步初始化 hmap,並計算出一個隨機的哈希種子,對應如下代碼

go
// 初始化
if h == nil {
    h = new(hmap)
}
// 獲取一個隨機的哈希種子
h.hash0 = fastrand()

再根據 hint 的值計算出哈希桶的容量,對應的代碼如下

go
B := uint8(0)
// 不斷循環直到 hint / 1 << B < 6.5
for overLoadFactor(hint, B) {
    B++
}
// 賦值給hmap
h.B = B

通過不斷循環找到第一個滿足(hint / 1 << B) < 6.5的 B 值,將其賦值給 hmap,在知曉了哈希桶的容量後,最後就是為哈希桶分配內存

go
if h.B != 0 {
    var nextOverflow *bmap
    // 分配好的哈希桶,和預先分配的空閒溢出桶
    h.buckets, nextOverflow = makeBucketArray(t, h.B, nil)
    // 如果預先分配了空閒溢出桶,就指向該溢出桶
    if nextOverflow != nil {
        h.extra = new(mapextra)
        h.extra.nextOverflow = nextOverflow
    }
}

makeBucketArray函數會根據 B 的值,為哈希桶分配對應大小的內存,以及預先分配空閒的溢出桶,當 B 小於 4 時,就不會創建溢出桶,如果大於 4 那麼就會創建2^B-4個溢出桶。對應runtime.makeBucketArray函數中的如下代碼

go
base := bucketShift(b)
nbuckets := base
// 小於4就不會創建溢出桶
if b >= 4 {
    // 預計桶的數量加上1 << (b-4)
    nbuckets += bucketShift(b - 4)
    // 溢出桶所需的內存
    sz := t.Bucket.Size_ * nbuckets
    // 將內存空間向上取整
    up := roundupsize(sz)
    if up != sz {
        // 不相等就采用up重新計算
        nbuckets = up / t.Bucket.Size_
    }
}

base指的是預計分配桶的數量,nbuckets指的是實際分配桶的數量,它加上了溢出桶的數量。

go
if base != nbuckets {
    // 第一個可用的溢出桶
    nextOverflow = (*bmap)(add(buckets, base*uintptr(t.BucketSize)))
    // 為了減少跟蹤溢出桶的開銷,將最後一個可用溢出桶的溢出指針指向哈希桶的頭部
    last := (*bmap)(add(buckets, (nbuckets-1)*uintptr(t.BucketSize)))
    // 最後一個溢出桶指向哈希桶
    last.setoverflow(t, (*bmap)(buckets))
}

當兩者不相等時,就說明分配了額外的溢出桶,nextoverflow指針就是指向的第一個可用的溢出桶。由此可見,哈希桶與溢出桶其實是在同一塊連續的內存中,這也是為什麼在圖中哈希桶數組與溢出桶數組是相鄰的。

訪問

在語法入門當中講到過,訪問 map 總共有三種方式,如下所示

go
# 直接訪問值
val := dict[key]
# 訪問值以及該鍵是否存在
val, exist := dict[key]
# 遍歷map
for key, val := range dict{

}

這三種方式所用到的函數都不相同,其中for range遍歷 map 最為復雜。

鍵值

對於前兩種方式而言,對應著兩個函數,分別是runtime.mapaccess1runtime.mapaccess2,函數簽名如下

go
func mapaccess1(t *maptype, h *hmap, key unsafe.Pointer) unsafe.Pointer

func mapaccess2(t *maptype, h *hmap, key unsafe.Pointer) (unsafe.Pointer, bool)

其中的 key 是指向訪問 map 的鍵的指針,返回的時候也只會返回指針。在訪問時,首先需要計算 key 的哈希值,定位 key 在哪個哈希桶,對應代碼如下

go
// 邊界處理
if h == nil || h.count == 0 {
    if t.HashMightPanic() {
        t.Hasher(key, 0) // see issue 23734
    }
    return unsafe.Pointer(&zeroVal[0])
}
// 防止並發讀寫
if h.flags&hashWriting != 0 {
    fatal("concurrent map read and map write")
}
// 使用指定類型的hasher計算哈希值
hash := t.Hasher(key, uintptr(h.hash0))
// (1 << B) - 1
m := bucketMask(h.B)
// 通過移動指針定位key所在的哈希桶
b := (*bmap)(add(h.buckets, (hash&m)*uintptr(t.BucketSize)))

在訪問的一開始先進行邊界情況處理,並防止 map 並發讀寫,當 map 處於並發讀寫狀態時,就會發生 panic。再然後計算哈希值,bucketMask函數所干的事就是計算(1 << B) - 1 hash & m就等於hash & (1 << B) - 1 ,這是二進制取余操作,等價於hash % (1 << B),使用位運算的好處就是更快。最後三行代碼干的事就是將 key 計算得到的哈希值與當前 map 中的桶的數量取余,得到哈希桶的序號,然後根據序號移動指針獲取 key 所在的哈希桶指針。

在知曉了 key 在哪個哈希桶後,就可以展開查找了,這部分對應代碼如下

go
  // 獲取哈希值的高八位
  top := tophash(hash)
bucketloop:
  // 遍歷bmap鏈表
  for ; b != nil; b = b.overflow(t) {
        // bmap中的元素
    for i := uintptr(0); i < bucketCnt; i++ {
            //  將計算得出的top與tophash中的元素進行對比
      if b.tophash[i] != top {
                // 後續都是空的,沒有了。
        if b.tophash[i] == emptyRest {
          break bucketloop
        }
                // 不相等就繼續遍歷溢出桶
        continue
      }
            // 根據i移動指針獲取對應下標的鍵
      k := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.KeySize))
            // 處理下指針
      if t.IndirectKey() {
        k = *((*unsafe.Pointer)(k))
      }
            // 比對兩個鍵是否相等
      if t.Key.Equal(key, k) {
                // 如果相等的話,就移動指針返回k對應下標的元素
                // 從這行代碼就能看出來鍵值的內存地址是連續的
        e := add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.KeySize)+i*uintptr(t.ValueSize))
        if t.IndirectElem() {
          e = *((*unsafe.Pointer)(e))
        }

        return e
      }
    }
  }
// 沒找到,返回零值。
return unsafe.Pointer(&zeroVal[0])

在定位哈希桶時,是通過取余來定位的,所以 key 在哪個哈希桶取決於哈希值的低位,至於到底是低幾位這取決於 B 的大小,而找到了哈希桶後,其中的 tophash 存放的是哈希值的高八位,因為低位取余值都是相同的,這樣就不需要去一個個對比 key 是否相等,只對比哈希值高八位就夠了。根據先前計算得到的哈希值獲取其高八位,在 bmap 中的 tophash 數組一個個對比,如果高八位相等的話,再對比鍵是否相等,如果鍵也相等的話就說明找到了元素,不相等就繼續遍歷 tophash 數組,還找不到就繼續遍歷溢出桶 bmap 鏈表,直到 bmap 的tophash[i]emptyRest退出循環,最後返回對應類型的零值。

mapaccess2 函數與mapaccess1函數邏輯完全一致,僅僅多了一個布爾返回值,用於表示元素是否存在。

遍歷

遍歷 map 的語法如下

go
for key, val := range dict {
  // do somthing...
}

在實際進行遍歷時,go 使用了hiter結構體來存放遍歷信息,hiter就是hmap interator的簡寫,意為哈希表迭代器,結構如下所示。

go
// A hash iteration structure.
// If you modify hiter, also change cmd/compile/internal/reflectdata/reflect.go
// and reflect/value.go to match the layout of this structure.
type hiter struct {
  key         unsafe.Pointer // Must be in first position.  Write nil to indicate iteration end (see cmd/compile/internal/walk/range.go).
  elem        unsafe.Pointer // Must be in second position (see cmd/compile/internal/walk/range.go).
  t           *maptype
  h           *hmap
  buckets     unsafe.Pointer // bucket ptr at hash_iter initialization time
  bptr        *bmap          // current bucket
  overflow    *[]*bmap       // keeps overflow buckets of hmap.buckets alive
  oldoverflow *[]*bmap       // keeps overflow buckets of hmap.oldbuckets alive
  startBucket uintptr        // bucket iteration started at
  offset      uint8          // intra-bucket offset to start from during iteration (should be big enough to hold bucketCnt-1)
  wrapped     bool           // already wrapped around from end of bucket array to beginning
  B           uint8
  i           uint8
  bucket      uintptr
  checkBucket uintptr
}

下面對它的一些字段做一些簡單的解釋

  • keyelem就是for range遍歷時獲取到的鍵值
  • buckets,在初始化迭代器時指定,指向哈希桶的頭部
  • bptr,當前正在遍歷的 bmap
  • startBucket,迭代開始時的起始桶序號
  • offset,桶內偏移量,范圍[0, bucketCnt-1]
  • B,就是 hmap 的 B 值,
  • i,桶內元素下標
  • wrapped,是否從哈希桶數組末尾回到了頭部

在遍歷開始前,go 通過runtime.mapiterinit函數來初始化遍歷器,然後再通過函數runtime.mapinternext對 map 進行遍歷,兩者都需要用到hiter結構體,兩個函數簽名如下。

go
func mapiterinit(t *maptype, h *hmap, it *hiter)

func mapiternext(it *hiter)

對於迭代器初始化而言,首先要獲取 map 當前的一個快照,對應如下的代碼。

go
it.t = t
it.h = h
// 記錄hmap當前狀態的快照,只需要保存B值。
it.B = h.B
it.buckets = h.buckets
if t.Bucket.PtrBytes == 0 {
    h.createOverflow()
    it.overflow = h.extra.overflow
    it.oldoverflow = h.extra.oldoverflow
}

後續在迭代時,實際上遍歷的是 map 的一個快照,而非實際的 map,所以在遍歷過程中添加的元素和桶都不會被遍歷到,並且同時並發遍歷寫入元素,有可能觸發fatal

go
if h.flags&hashWriting != 0 {
    fatal("concurrent map iteration and map write")
}

第二步再決定遍歷的兩個起始位置,第一個是起始桶的位置,第二個桶內的起始位置,這兩個都是隨機選的,對應如下代碼

go
// r是一個隨機數
var r uintptr
if h.B > 31-bucketCntBits {
    r = uintptr(fastrand64())
} else {
    r = uintptr(fastrand())
}

// r % (1 << B) 得到起始桶的位置
it.startBucket = r & bucketMask(h.B)
// r >> B % 8 得到起始桶內的元素起始位置
it.offset = uint8(r >> h.B & (bucketCnt - 1))

// 記錄當前正在遍歷的桶序號
it.bucket = it.startBucket

通過fastrand()fastrand64()獲取一個隨機數,兩次取模運算得到桶起始位置和桶內起始位置。

TIP

map 雖然不允許同時並發讀寫,但是允許同時並發遍歷。

接下來才真正開始迭代 map,如何對桶進行遍歷,以及退出的策略,這部分對應下面的代碼

go
// hmap
h := it.h
// maptype
t := it.t
// 待遍歷的桶的位置
bucket := it.bucket
// 待遍歷的bmap
b := it.bptr
// 桶內序號i
i := it.i

next:
  if b == nil {
        // 如果當前桶的位置與起始位置相等,說明是繞了一圈回來,後面的已經遍歷過了
        // 遍歷結束,可以退出了
    if bucket == it.startBucket && it.wrapped {
      it.key = nil
      it.elem = nil
      return
    }
        // 桶下標後移
    bucket++
        // bucket == 1 << B,就是走到哈希桶數組的末尾了
    if bucket == bucketShift(it.B) {
            // 從頭開始
      bucket = 0
      it.wrapped = true
    }
    i = 0
  }

哈希桶的起始位置選取是隨機的,在遍歷時,從起始位置向桶切片的末尾逐個迭代,走到1 << B時,然後再從頭開始,當再次回到起始位置時,說明已經遍歷完畢,然後退出。上面的代碼是關於如何在 map 中遍歷桶,而下面的代碼描述的就是如何在桶內進行遍歷。

go
for ; i < bucketCnt; i++ {
    // (i + offset) % 8
    offi := (i + it.offset) & (bucketCnt - 1)
    // 如果當前元素是空的就跳過
    if isEmpty(b.tophash[offi]) || b.tophash[offi] == evacuatedEmpty {
      continue
    }
    // 移動指針獲取鍵
    k := add(unsafe.Pointer(b), dataOffset+uintptr(offi)*uintptr(t.KeySize))
    if t.IndirectKey() {
      k = *((*unsafe.Pointer)(k))
    }
      // 移動指針獲取值
    e := add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.KeySize)+uintptr(offi)*uintptr(t.ValueSize))

    // 處理等量擴容的情況,當鍵值被疏散到了其它位置後,需要重新去尋找鍵值。
    if (b.tophash[offi] != evacuatedX && b.tophash[offi] != evacuatedY) ||
      !(t.ReflexiveKey() || t.Key.Equal(k, k)) {
      it.key = k
      if t.IndirectElem() {
        e = *((*unsafe.Pointer)(e))
      }
      it.elem = e
    } else {
      rk, re := mapaccessK(t, h, k)
      if rk == nil {
        continue
      }
      it.key = rk
      it.elem = re
    }
    it.bucket = bucket
    it.i = i + 1
    return
  }
  // 沒找到就去溢出桶裡面找
  b = b.overflow(t)
  i = 0
  goto next

TIP

在上面的擴容判斷條件中,有一個表達式可能會讓人感到困惑,如下

go
t.Key.Equal(k, k)

之所以要去判斷k是否與自身相等,是為了過濾鍵為Nan的情況,如果一個元素的鍵是Nan,那麼就會無法正常訪問該元素,無論是遍歷還是直接訪問,或者是刪除都無法正常進行,因為Nan != Nan永遠成立,也就永遠無法找到這個鍵。

首先根據i值和offset值取模運算得到待遍歷的桶內下標,通過移動指針獲取鍵值,由於在 map 遍歷期間,會有其它的寫入操作觸發了 map 的擴容,所以實際的鍵值可能已經不在原來的位置了,在這種情況下就需要使用mapaccessK函數去重新獲取實際的鍵值,該函數簽名如下

go
func mapaccessK(t *maptype, h *hmap, key unsafe.Pointer) (unsafe.Pointer, unsafe.Pointer)

它的功能與mapaccess1函數完全一致,區別在於mapaccessK函數會同時返回 key 值和 value 值。最終獲取到了鍵值以後,將其賦值給迭代器的keyelem,然後更新迭代器的下標,這樣就完成了一次迭代,代碼執行就回到了for range的代碼塊中。如果在桶內沒有找到,就再去溢出桶裡面找,繼續重復上面的步驟,直到溢出桶鏈表遍歷完畢後,再繼續迭代下一個哈希桶。

修改

修改 map 的語法如下

go
dict[key] = val

在 go 中,對於修改 map 的操作,由runtime.mapassign函數來完成,該函數簽名如下

go
func mapassign(t *maptype, h *hmap, key unsafe.Pointer) unsafe.Pointer

其訪問過程的邏輯與mapaccess1相同,不過 key 不存在時會為其分配一個位置,如果存在就更新,最後返回元素的指針。在開始時,需要做一些准備工作,這部分對應的代碼如下

go
// 不允許寫入為nil的map
if h == nil {
    panic(plainError("assignment to entry in nil map"))
}
// 禁止同時並發寫
if h.flags&hashWriting != 0 {
    fatal("concurrent map writes")
}
// 計算key哈希值
hash := t.Hasher(key, uintptr(h.hash0))

// 修改hmap狀態
h.flags ^= hashWriting

// 初始化哈希桶
if h.buckets == nil {
    h.buckets = newobject(t.Bucket) // newarray(t.Bucket, 1)
}

上面的代碼主要做了以下幾件事

  • hmap 狀態檢查
  • 計算 key 的哈希值
  • 檢查哈希桶是否需要初始化

再之後通過哈希值取模運算得到哈希桶位置,以及 key 的 tophash,代碼對應如下

go
again:
  // hash % (1 << B)
  bucket := hash & bucketMask(h.B)
  // 移動指針獲取指定位置的bmap
  b := (*bmap)(add(h.buckets, bucket*uintptr(t.BucketSize)))
  // 計算tophash
  top := tophash(hash)

現在確定了桶的位置,bmap,tophash,就可以開始查找元素了,這部分代碼對應如下

go
// 待插入的tophash
var inserti *uint8
// 待插入的key值指針
var insertk unsafe.Pointer
// 待插入的value值指針
var elem unsafe.Pointer

bucketloop:
  for {
        // 遍歷桶內的tophash數組
    for i := uintptr(0); i < bucketCnt; i++ {
            // tophash不相等
      if b.tophash[i] != top {
                // 如果當前桶內下標是空的,且還沒有插入元素,就選取該位置插入
        if isEmpty(b.tophash[i]) && inserti == nil {
                    // 找到了一個合適的位置分配給key
          inserti = &b.tophash[i]
          insertk = add(unsafe.Pointer(b), dataOffset+i*uintptr(t.KeySize))
          elem = add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.KeySize)+i*uintptr(t.ValueSize))
        }
                // 遍歷完了就退出循環
        if b.tophash[i] == emptyRest {
          break bucketloop
        }
        continue
      }
            // 如果tophash相等的話,則說明key可能已經存在了
      k := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.KeySize))
      if t.IndirectKey() {
        k = *((*unsafe.Pointer)(k))
      }
            // 判斷key是否相等
      if !t.Key.Equal(key, k) {
        continue
      }
      // 如果需要更新key值的話,就將key的內存直接復制到k處
      if t.NeedKeyUpdate() {
        typedmemmove(t.Key, k, key)
      }
            // 得到了元素的指針
      elem = add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.KeySize)+i*uintptr(t.ValueSize))
            // 更新完畢,返回元素指針
      goto done
    }
        // 執行到這裡說明沒找到key,遍歷溢出桶鏈表,繼續找
    ovf := b.overflow(t)
    if ovf == nil {
      break
    }
    b = ovf
  }

  // 執行到這裡說明key沒有存在於map中,但可能已經找到了一個合適的位置分配給key,也可能沒有

  // 沒有找到一個合適的位置分配給key
  if inserti == nil {
        // 說明當前的哈希桶以及它的溢出桶都滿了,那就再分配一個溢出桶
    newb := h.newoverflow(t, b)
        // 在溢出桶中分配一個位置給key
    inserti = &newb.tophash[0]
    insertk = add(unsafe.Pointer(newb), dataOffset)
    elem = add(insertk, bucketCnt*uintptr(t.KeySize))
  }

  // 如果存放的是key指針的話
  if t.IndirectKey() {
        // 新分配內存,返回的是一個unsafe指針
    kmem := newobject(t.Key)
    *(*unsafe.Pointer)(insertk) = kmem
        // 賦值給insertk,方便後面進行key的內存復制
    insertk = kmem
  }

  // 如果存放的是元素指針
  if t.IndirectElem() {
        // 分配內存
    vmem := newobject(t.Elem)
        // 讓指針指向vmem
    *(*unsafe.Pointer)(elem) = vmem
  }
  // 將key的內存直接復制到insertk的位置
  typedmemmove(t.Key, insertk, key)
  *inserti = top
  // 數量加一
  h.count++

done:
  // 執行到這裡說明修改過程已經完成了
  if h.flags&hashWriting == 0 {
    fatal("concurrent map writes")
  }
  h.flags &^= hashWriting
  if t.IndirectElem() {
    elem = *((*unsafe.Pointer)(elem))
  }
  // 返回元素指針
  return elem

在上述一大段代碼中,首先進入 for 循環嘗試在哈希桶和溢出桶中去查找,查找的邏輯與mapaccess完全一致,此時有三種可能

  • 第一種,在 map 中找到了已經存在的 key,直接跳到done代碼塊,返回元素指針
  • 第二種,在 map 中找到了一個位置分配給 key 使用,將 key 復制到指定位置,並返回元素指針
  • 第三種,所有桶都找完了,沒有在 map 中找到可以分配給 key 的位置,創建一個新的溢出桶,並將 key 分配到桶中,然後將 key 復制到指定位置,並返回元素指針

最終得到了元素指針後,就可以賦值了,不過這部分操作並不由mapassign函數來完成,它只負責返回元素指針,賦值操作是在編譯器期間插入的,在代碼裡面看不見,但是編譯後的代碼可以看見,假設有下面的代碼

go
func main() {
  dict := make(map[string]string, 100)
  dict["hello"] = "world"
}

通過命令如下命令得到匯編代碼

go tool compile -S -N -l main.go

關鍵的地方就在這部分

0x004e 00078     LEAQ    type:map[string]string(SB), AX
0x0055 00085     CALL    runtime.mapassign_faststr(SB)
0x005a 00090     MOVQ    AX, main..autotmp_2+40(SP)
0x0083 00131     LEAQ    go:string."world"(SB), CX
0x008a 00138     MOVQ    CX, (AX)

可以看到調用了runtime.mapassign_faststr,其邏輯與mapassign完全相似,LEAQ go:string."world"(SB), CX就是將字符串的地址存儲到CX上,MOVQ CX, (AX)再將其存儲到了AX上,於是就完成了元素的賦值。

刪除

在 go 中,要想刪除一個 map 的元素,只能使用內置函數delete,如下

go
delete(dict, "abc")

其內部調用的是runtime.mapdelete函數,該函數簽名如下

go
func mapdelete(t *maptype, h *hmap, key unsafe.Pointer)

它會刪除 map 中指定 key 的元素,不管元素是否存在於 map 中,它都不會有任何反應。函數開頭做准備工作的邏輯都是類似的,無非就是下面幾件事

  • hmap 狀態檢查
  • 計算 key 的哈希值
  • 定位哈希桶
  • 計算 tophash

前面有很多重復的內容,就不再贅述了,這裡只關注它刪除的邏輯,對應的部分代碼如下。

go
bOrig := b
search:
  for ; b != nil; b = b.overflow(t) {
    for i := uintptr(0); i < bucketCnt; i++ {
      if b.tophash[i] != top {
        // 沒找到就退出循環
        if b.tophash[i] == emptyRest {
          break search
        }
        continue
      }
      // 移動指針找到key的位置
      k := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.KeySize))
      k2 := k
      if t.IndirectKey() {
        k2 = *((*unsafe.Pointer)(k2))
      }
      if !t.Key.Equal(key, k2) {
        continue
      }

      // 刪除key
      if t.IndirectKey() {
        *(*unsafe.Pointer)(k) = nil
      } else if t.Key.PtrBytes != 0 {
        memclrHasPointers(k, t.Key.Size_)
      }
       // 移動指針找到元素的位置
      e := add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.KeySize)+i*uintptr(t.ValueSize))

      // 刪除元素
      if t.IndirectElem() {
        *(*unsafe.Pointer)(e) = nil
      } else if t.Elem.PtrBytes != 0 {
        memclrHasPointers(e, t.Elem.Size_)
      } else {
        memclrNoHeapPointers(e, t.Elem.Size_)
      }

    notLast:
      // 數量減一
      h.count--
      // 重置哈希種子,減小沖突發生概率
      if h.count == 0 {
        h.hash0 = fastrand()
      }
      break search
    }
  }

通過上面的代碼可以看到,查找的邏輯跟前面的操作是幾乎完全一致的,找到元素,然後刪除,hmap 記錄的數量減一,當數量減為 0 時,就重置哈希種子。

另一個要注意的點就是,在刪除完元素後,需要修改當前下標的 tophash 狀態,這部分對應的代碼如下。當 i 在桶末尾時,判斷根據下一個溢出桶來判斷當前元素是否是最後一個可用元素,否則的話就直接查看相鄰元素的哈希狀態。如果當前元素不是最後一個可用的,就將狀態置為emptyOne

go
// 將當前tophash標記為空
b.tophash[i] = emptyOne

// 如果在tophash末尾
if i == bucketCnt-1 {
    // 溢出桶不為空,且溢出桶內有元素,說明當前元素不是最後一個
    if b.overflow(t) != nil && b.overflow(t).tophash[0] != emptyRest {
        goto notLast
    }
} else {
    // 相鄰元素不為空
    if b.tophash[i+1] != emptyRest {
        goto notLast
    }
}

如果元素確實是最後一個元素的話,就需要修正一下桶鏈表中部分桶的 tophash 數組的值,否則的話後續遍歷時會導致無法在正確的位置退出。在 map 創建溢出桶的時候講述過,go 為了減少追蹤溢出桶的成本,最後一個溢出桶的overflow指針就是指向頭部的哈希桶,所以它實際上是一個單向環形鏈表,鏈表的「頭部」就是哈希桶。而這裡的b是查找過後的b,很可能是鏈表中間的某一個溢出桶,逆序遍歷環形鏈表查找最後一個存在的元素。盡管代碼寫的是正序遍歷,由於鏈表是一個環,它不斷正序遍歷直到當前溢出桶的前一個為止,從結果上來說確實是逆序的。再然後逆序遍歷桶中的 tophash 數組,將狀態為emptyOne的元素更新為emptyRest,直到找到最後一個存在的元素。為避免無限陷入在環中,當回到了最開始的桶時,也就是bOrig,說明此時鏈表內已經沒有可用的元素了,就可以退出循環了。

go
// 執行到這裡說明當前元素後面沒有元素了
// 不斷的倒序遍歷bmap鏈表,倒序遍歷桶內的tophash
// 將狀態為emptyOne的更新為emptyRest
for {
    b.tophash[i] = emptyRest
    if i == 0 {
        if b == bOrig {
            break
        }

        c := b
    // 找到當前bmap鏈表的前一個
        for b = bOrig; b.overflow(t) != c; b = b.overflow(t) {
        }
        i = bucketCnt - 1
    } else {
        i--
    }
    if b.tophash[i] != emptyOne {
        break
    }
}

清空

go1.21版本中,新增了內置函數clear,可以用於清空 map 中的所有元素,語法如下

go
clear(dict)

其內部調用了runtime.mapclear函數,它負責刪除 map 中的所有元素,其函數簽名如下

go
func mapclear(t *maptype, h *hmap)

該函數的邏輯並不復雜,首先需要將整個 map 標記為空,對應的代碼如下。

go
// 遍歷每一個桶以及溢出桶,將所有的tophash元素置為emptyRest
markBucketsEmpty := func(bucket unsafe.Pointer, mask uintptr) {
    for i := uintptr(0); i <= mask; i++ {
        b := (*bmap)(add(bucket, i*uintptr(t.BucketSize)))
        for ; b != nil; b = b.overflow(t) {
            for i := uintptr(0); i < bucketCnt; i++ {
                b.tophash[i] = emptyRest
            }
        }
    }
}
markBucketsEmpty(h.buckets, bucketMask(h.B))

上面代碼做的事情就是遍歷每一個桶,將桶中的 tophash 數組的元素都置為emptyRest,將 map 標記為空,這樣就能阻止迭代器繼續迭代,然後再清空 map,對應的代碼如下。

go
// 重置哈希種子
h.hash0 = fastrand()

// 重置extra結構體
if h.extra != nil {
    *h.extra = mapextra{}
}

// 這個操作會清除原來buckets的內存,並重新分配新的桶
_, nextOverflow := makeBucketArray(t, h.B, h.buckets)
if nextOverflow != nil {
  // 分配新的空閒溢出桶
    h.extra.nextOverflow = nextOverflow
}

通過makeBucketArray清除之前的桶的內存,然後新分配一個,這樣一來就完成了桶的清除,除此之外還有一些細節,比如說將count置 0,還有其它的一些操作就不過多贅述。

擴容

在之前的所有操作中,為了更關注其本身的邏輯,所以屏蔽了很多跟擴容有關內容,這樣會簡單很多。擴容的邏輯其實比較復雜,放在最後就是不希望產生干擾,那麼現在就來看看 go 是如何對 map 進行擴容的,前面已經提到過,觸發擴容有兩個條件:

  • 負載因子超過 6.5
  • 溢出桶的數量過多

判斷負載因子是否超過閾值的函數是runtime.overLoadFactor函數,在哈希沖突部分已經闡述過,而判斷溢出桶的數量是否過多的函數是runtime.tooManyOverflowBuckets,其工作原理也很簡單,代碼如下

go
func tooManyOverflowBuckets(noverflow uint16, B uint8) bool {
  if B > 15 {
    B = 15
  }
  return noverflow >= uint16(1)<<(B&15)
}

上面的代碼可以簡化成如下表達式,一眼就能看懂。

overflow >= 1 << (min(15,B) % 16)

在這裡,go 對於 too many 的定義是:溢出桶的數量跟哈希桶的數量差不多,如果閾值低了,就會做多余的工作,如果閾值高了,那麼在擴容的時候就會佔用大量的內存。在修改和刪除元素時就有可能觸發擴容,判斷是否需要擴容的代碼如下。

go
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
}

可以看到的就是這三個條件限制

  1. 當前不能正在擴容
  2. 負載因子小於 6.5
  3. 溢出桶數量不能過多

負責擴容的函數自然就是runtime.hashGrow,其函數簽名如下

go
func hashGrow(t *maptype, h *hmap)

實際上,擴容也分種類,根據不同條件觸發的擴容其類型也不同,分為以下兩種

  • 增量擴容
  • 等量擴容

增量擴容

當負載因子過大時,即元素數量較大於哈希桶的數量,當哈希沖突比較嚴重的時候,會形成很多溢出桶鏈表,這樣會導致 map 讀寫性能下降,因為需要查找一個元素遍歷更多的溢出桶鏈表,而遍歷的時間復雜度是 O(n),哈希表查找的時間復雜度主要取決於哈希值的計算時間和遍歷的時間,當遍歷的時間遠小於計算哈希的時間時,查找的時間復雜度才能稱為 O(1)。倘若哈希沖突比較頻繁,過多 key 都被分配到了同一個哈希桶,溢出桶鏈表過長導致遍歷時間增大,就會導致查找時間增大,而增刪改操作都需要先進行查找操作,這樣一來的話就會導致整個 map 的性能嚴重下降。

像圖中這種比較極端的情況,查找的時間復雜度基本上跟 O(n)沒啥區別了。面對這種情況,解決辦法就是新增更多的哈希桶,避免形成過長的溢出桶鏈表,這種方法也被稱為增量擴容。

在 go 中,增量擴容每次都會將 B 加一,也就是哈希桶的規模每次擴大一倍。擴容後,需要將舊數據搬遷到新的 map 中,倘若 map 中的元素數量以千萬甚至億計,一次性全部搬遷完的話耗時會很久,所以 go 采用的是逐步搬遷的策略,為此,go 會將 hamp 中的oldBuckets指向原來的哈希桶數組,表示這是舊數據,然後創建更大容量的哈希桶數組,讓hmap中的buckets指向新的哈希桶數組,然後在每一次修改和刪除元素時,將部分元素搬遷從舊桶搬到新桶,直到搬遷完畢,舊桶的內存才會被釋放掉。

go
func hashGrow(t *maptype, h *hmap) {
  // 差值
  bigger := uint8(1)
  // 舊桶
  oldbuckets := h.buckets
  // 新的哈希桶
  newbuckets, nextOverflow := makeBucketArray(t, h.B+bigger, nil)

  flags := h.flags &^ (iterator | oldIterator)
  if h.flags&iterator != 0 {
    flags |= oldIterator
  }
  // B+bigger
  h.B += bigger
  h.flags = flags
  h.oldbuckets = oldbuckets
  h.buckets = newbuckets
  h.nevacuate = 0
  h.noverflow = 0

  // 溢出桶也要指定舊桶和新桶
  if h.extra != nil && h.extra.overflow != nil {
    if h.extra.oldoverflow != nil {
      throw("oldoverflow is not nil")
    }
    h.extra.oldoverflow = h.extra.overflow
    h.extra.overflow = nil
  }
  if nextOverflow != nil {
    if h.extra == nil {
      h.extra = new(mapextra)
    }
    h.extra.nextOverflow = nextOverflow
  }

}

上面的代碼做的事情就是創建大一倍容量的新哈希桶,然後指定哈希新舊桶的引用,以及溢出新舊桶的引用,而實際的搬遷工作並不由hashGrow函數來完成,它只負責指定新舊桶,並更新hmap的一些狀態。這些工作實際上是由runtime.growWork函數完成的,其函數簽名如下

go
func growWork(t *maptype, h *hmap, bucket uintptr)

它會在mapassign函數和mapdelete函數中,以如下的形式被調用,其作用就是如果當前 map 正在擴容中,就進行一次部分搬遷工作。

go
if h.growing() {
    growWork(t, h, bucket)
}

在進行修改和刪除操作時,需要判斷當前是否處於擴容中,這主要由hmap.growing方法來完成,內容很簡單,就是判斷oldbuckets是否為不為空,對應代碼如下。

go
func (h *hmap) growing() bool {
  return h.oldbuckets != nil
}

來看看growWork函數都干了些什麼。

go
func growWork(t *maptype, h *hmap, bucket uintptr) {
    // bucket % 1 << (B-1)
  evacuate(t, h, bucket&h.oldbucketmask())
  if h.growing() {
    evacuate(t, h, h.nevacuate)
  }
}

其中bucket&h.oldbucketmask()操作是計算得到舊桶的位置,確保將要搬遷的是當前桶的舊桶。從函數中可以看到真正負責搬遷工作的是runtime.evacuate函數,其中用到了evaDst結構體用來表示搬遷目的地,主要作用是在搬遷的過程中迭代新桶,它的結構如下。

go
type evacDst struct {
  b *bmap          // 搬遷目的地的新桶
  i int            // 桶內下標
  k unsafe.Pointer // 指向新鍵目的地的指針
  e unsafe.Pointer // 指向新值目的地的指針
}

在搬遷開始之前,go 會分配兩個evacDst結構體,一個指向新哈希桶的上半區,另一個指向新哈希桶的下半區,對應的代碼如下

go
// 定位舊桶中指定的哈希桶
b := (*bmap)(add(h.oldbuckets, oldbucket*uintptr(t.BucketSize)))
// 舊桶的長度 = 1 << (B - 1)
newbit := h.noldbuckets()

var xy [2]evacDst
x := &xy[0]
// 指向新桶的上半區
x.b = (*bmap)(add(h.buckets, oldbucket*uintptr(t.BucketSize)))
x.k = add(unsafe.Pointer(x.b), dataOffset)
x.e = add(x.k, bucketCnt*uintptr(t.KeySize))

// 判斷是不是等量擴容
if !h.sameSizeGrow() {
    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*uintptr(t.KeySize))
}

舊桶的搬遷目的地是兩個新桶,搬遷後,桶中的部分數據會在上半區,另一部分的數據會在下半區,這樣做是希望擴容後的數據能夠更加均勻的分布,分布的越均勻,map 的查找性能就會越好。go 如何將數據搬遷到兩個新桶中,對應著下面的代碼。

go
// 遍歷溢出桶鏈表
for ; b != nil; b = b.overflow(t) {
    // 拿到每個桶的第一鍵值對
    k := add(unsafe.Pointer(b), dataOffset)
    e := add(k, bucketCnt*uintptr(t.KeySize))
    // 遍歷桶中的每一個鍵值對
    for i := 0; i < bucketCnt; i, k, e = i+1, add(k, uintptr(t.KeySize)), add(e, uintptr(t.ValueSize)) {
        top := b.tophash[i]
        // 如果是空的就跳過
        if isEmpty(top) {
            b.tophash[i] = evacuatedEmpty
            continue
        }
        // 新哈希桶不應該處於搬遷狀態
        // 否則的話肯定是出問題了
        if top < minTopHash {
            throw("bad map state")
        }
        k2 := k
        if t.IndirectKey() {
            k2 = *((*unsafe.Pointer)(k2))
        }
        // 該變量決定了當前鍵值對被搬遷到上半區還是下半區
        // 它的值只能是0或1
        var useY uint8
        if !h.sameSizeGrow() {
            // 重新計算哈希值
            hash := t.Hasher(k2, uintptr(h.hash0))
            // 處理k2 != k2的特殊情況,
            if h.flags&iterator != 0 && !t.ReflexiveKey() && !t.Key.Equal(k2, k2) {
                useY = top & 1
                top = tophash(hash)
            } else {
                // hash % 1 << (B - 1)
                if hash&newbit != 0 {
                    useY = 1
                }
            }
        }
        // 檢查常量的值
        if evacuatedX+1 != evacuatedY || evacuatedX^1 != evacuatedY {
            throw("bad evacuatedN")
        }

        // 更新舊桶的tophash,表示當前元素已被搬遷
        b.tophash[i] = evacuatedX + useY // evacuatedX + 1 == evacuatedY
        // 指定搬遷目的地
        dst := &xy[useY]                 // evacuation destination

        // 新桶容量不夠用了建個溢出桶
        if dst.i == bucketCnt {
            dst.b = h.newoverflow(t, dst.b)
            dst.i = 0
            dst.k = add(unsafe.Pointer(dst.b), dataOffset)
            dst.e = add(dst.k, bucketCnt*uintptr(t.KeySize))
        }
        dst.b.tophash[dst.i&(bucketCnt-1)] = top // mask dst.i as an optimization, to avoid a bounds check
        // 復制鍵
        if t.IndirectKey() {
            *(*unsafe.Pointer)(dst.k) = k2 // copy pointer
        } else {
            typedmemmove(t.Key, dst.k, k) // copy elem
        }
        // 復制值
        if t.IndirectElem() {
            *(*unsafe.Pointer)(dst.e) = *(*unsafe.Pointer)(e)
        } else {
            typedmemmove(t.Elem, dst.e, e)
        }
        // 後移新桶目的指針,為下一個鍵值做准備
        dst.i++
        dst.k = add(dst.k, uintptr(t.KeySize))
        dst.e = add(dst.e, uintptr(t.ValueSize))
    }
}

從上面的代碼可以看到,go 會遍歷舊桶鏈表中的每一個桶中的每一個元素,將其中的數據搬到新桶中,決定數據到底是去上半區還是下半區取決於重新計算後的哈希值

go
// hash % 1 << (B - 1)
if hash&newbit != 0 {
    useY = 1
}

在搬遷後,會將當前元素的 tophash 置為evacuatedXevacuated,如果在擴容的過程中嘗試查找數據,通過此狀態就可以得知數據已經被搬遷,就知道要去新桶裡面找對應的數據。這部分邏輯對應runtime.mapaccess1函數中的如下代碼。

go
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)))
    // 如果舊桶已經被搬遷了就不找了
    if !evacuated(oldb) {
        b = oldb
    }
}

在訪問元素時,如果當前正處於擴容狀態,會先去嘗試去舊桶裡面查找,如果舊桶已經被搬遷了就不去舊桶裡面找。回到搬遷這塊,此時已經確定了搬遷的目的地,接下來要做就是將數據復制到新桶中,然後讓evacDst結構體指向下一個目的地。

go
dst := &xy[useY]
dst.b.tophash[dst.i&(bucketCnt-1)] = top
typedmemmove(t.Key, dst.k, k)
typedmemmove(t.Elem, dst.e, e)

if h.flags&oldIterator == 0 && t.Bucket.PtrBytes != 0 {
    b := add(h.oldbuckets, oldbucket*uintptr(t.BucketSize))
    ptr := add(b, dataOffset)
    n := uintptr(t.BucketSize) - dataOffset
    memclrHasPointers(ptr, n)
}

這樣操作直到當前桶全部搬遷完畢,然後 go 就會將當前舊桶鍵值數據全部內存清除,只留下一個哈希桶 tophash 數組(留下是因為後續要靠 tophash 數組判斷搬遷狀態),舊桶中的溢出桶內存由於不再被引用後續會被 GC 回收掉。hmap中有一個字段nevacuate用來記錄搬遷進度,每搬完一個舊的哈希桶,就會加一,當它的值與舊桶的數量相等時,就說明整個 map 的擴容已經完成了,接下來就由runtime.advanceEvacuationMark函數進行擴容的收尾工作。

go
func advanceEvacuationMark(h *hmap, t *maptype, newbit uintptr) {
  h.nevacuate++
  stop := h.nevacuate + 1024
  if stop > newbit {
    stop = newbit
  }
  for h.nevacuate != stop && bucketEvacuated(t, h, h.nevacuate) {
    h.nevacuate++
  }
    if h.nevacuate == newbit { // newbit = len(oldbuckets)
    h.oldbuckets = nil
    if h.extra != nil {
      h.extra.oldoverflow = nil
    }
    h.flags &^= sameSizeGrow
  }
}

它會統計已搬遷數量並確認是否與舊桶數量相等,相等的話就清除對於所有舊桶和舊溢出桶的引用,至此擴容完畢。

growWork函數中,總共調用了兩次evacuate函數,第一次是搬當前正在訪問桶的舊桶,第二次是搬h.nevacuate所指向的舊桶,總共搬了兩次,說明逐步搬遷時,每一次都會搬兩個桶。

等量擴容

前面提到過,等量擴容的觸發條件是溢出桶數量過多,假如 map 先是添加了大量的元素,然後又大量刪除元素,這樣一來可能很多桶都是空的,可能有些桶有很多的元素,數據分布十分不均勻,有相當多的溢出桶都是空的,佔用了不少的內存。為了解決這類問題,就需要創建一個同等容量的新 map,重新分配一次哈希桶,這個過程就叫等量擴容。所以其實並不是擴容操作,只是將所有元素二次分配使數據分布更加均勻,等量擴容操作是糅合在增量擴容操作中的,其邏輯與增量擴容完全一致,新舊 map 容量完全相等。

hashGrow函數中,如果負載因子沒有超過閾值,進行的就是等量擴容,go 更新h.flags的狀態為sameSizeGrowh.B也不會加一,所以新創建的哈希桶容量也不會有變化,對應代碼如下。

go
bigger := uint8(1)
if !overLoadFactor(h.count+1, h.B) {
    bigger = 0
    h.flags |= sameSizeGrow
}
oldbuckets := h.buckets
newbuckets, nextOverflow := makeBucketArray(t, h.B+bigger, nil)

evacuate函數中,剛開始創建eavcDst結構體時,如果是等量擴容的話就只會創建一個結構體指向新桶,對應代碼如下。

go
if !h.sameSizeGrow() {
    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*uintptr(t.KeySize))
}

並且在搬遷元素的時候,等量擴容不會重新計算哈希值,也沒有上下半區的選擇

go
if !h.sameSizeGrow() {
    // 重新計算哈希值
    hash := t.Hasher(k2, uintptr(h.hash0))
    // 處理k2 != k2的特殊情況,
    if h.flags&iterator != 0 && !t.ReflexiveKey() && !t.Key.Equal(k2, k2) {
        useY = top & 1
        top = tophash(hash)
    } else {
        // hash % 1 << (B - 1)
        if hash&newbit != 0 {
            useY = 1
        }
    }
}

除了這些,其它的邏輯與增量擴容完全一致。經過等量擴容哈希桶重新分配過後,溢出桶的數量就會減少,舊的溢出桶都會被 GC 回收掉。

Golang學習網由www.golangdev.cn整理維護