关于golang:聊聊dapr的consistent-hash

序

本文次要钻研一下 dapr 的 consistent hash

consistent_hash

dapr/pkg/placement/hashing/consistent_hash.go

var replicationFactor int

// ErrNoHosts is an error for no hosts
var ErrNoHosts = errors.New("no hosts added")

// ConsistentHashTables is a table holding a map of consistent hashes with a given version
type ConsistentHashTables struct {
    Version string
    Entries map[string]*Consistent
}

// Host represents a host of stateful entities with a given name, id, port and load
type Host struct {
    Name  string
    Port  int64
    Load  int64
    AppID string
}

// Consistent represents a data structure for consistent hashing
type Consistent struct {hosts     map[uint64]string
    sortedSet []uint64
    loadMap   map[string]*Host
    totalLoad int64

    sync.RWMutex
}

ConsistentHashTables 定义了 Version、Entries 属性，Entries 是个 map，value 为 Consistent；Consistent 定义了 hosts、sortedSet、loadMap、totalLoad、sync.RWMutex 属性

GetInternals

dapr/pkg/placement/hashing/consistent_hash.go

// GetInternals returns the internal data structure of the consistent hash
func (c *Consistent) GetInternals() (map[uint64]string, []uint64, map[string]*Host, int64) {c.RLock()
    defer c.RUnlock()

    return c.hosts, c.sortedSet, c.loadMap, c.totalLoad
}

GetInternals 办法返回 hosts、sortedSet、loadMap、totalLoad 属性

Add

dapr/pkg/placement/hashing/consistent_hash.go

// Add adds a host with port to the table
func (c *Consistent) Add(host, id string, port int64) bool {c.Lock()
    defer c.Unlock()

    if _, ok := c.loadMap[host]; ok {return true}

    c.loadMap[host] = &Host{Name: host, AppID: id, Load: 0, Port: port}
    for i := 0; i < replicationFactor; i++ {h := c.hash(fmt.Sprintf("%s%d", host, i))
        c.hosts[h] = host
        c.sortedSet = append(c.sortedSet, h)
    }
    // sort hashes ascendingly
    sort.Slice(c.sortedSet, func(i int, j int) bool {return c.sortedSet[i] < c.sortedSet[j]
    })

    return false
}

Add 办法创立 Host 并增加到 loadMap 中，之后依据 replicationFactor 次数对 host 进行 hash 并增加到 hosts 及 sortedSet 中，最初对 sortedSet 进行排序

Get

dapr/pkg/placement/hashing/consistent_hash.go

// Get returns the host that owns `key`.
//
// As described in https://en.wikipedia.org/wiki/Consistent_hashing
//
// It returns ErrNoHosts if the ring has no hosts in it.
func (c *Consistent) Get(key string) (string, error) {c.RLock()
    defer c.RUnlock()

    if len(c.hosts) == 0 {return "", ErrNoHosts}

    h := c.hash(key)
    idx := c.search(h)
    return c.hosts[c.sortedSet[idx]], nil
}

Get 办法先对 key 进行 hash，而后通过 search 查找 idx，最初找到 idx 在 sortedSet 中对应的 host，最初从 hosts 中返回对应 host

GetHost

dapr/pkg/placement/hashing/consistent_hash.go

// GetHost gets a host
func (c *Consistent) GetHost(key string) (*Host, error) {h, err := c.Get(key)
    if err != nil {return nil, err}

    return c.loadMap[h], nil
}

GetHost 办法先 get key，之后再去 loadMap 获取 host

GetLeast

dapr/pkg/placement/hashing/consistent_hash.go

// GetLeast uses Consistent Hashing With Bounded loads
//
// https://research.googleblog.com/2017/04/consistent-hashing-with-bounded-loads.html
//
// to pick the least loaded host that can serve the key
//
// It returns ErrNoHosts if the ring has no hosts in it.
//
func (c *Consistent) GetLeast(key string) (string, error) {c.RLock()
    defer c.RUnlock()

    if len(c.hosts) == 0 {return "", ErrNoHosts}

    h := c.hash(key)
    idx := c.search(h)

    i := idx
    for {host := c.hosts[c.sortedSet[i]]
        if c.loadOK(host) {return host, nil}
        i++
        if i >= len(c.hosts) {i = 0}
    }
}

GetLeast 办法先对 key 进行 hash，而后通过 search 获取 idx，之后通过 loadOK 来获取 least loaded host

search

dapr/pkg/placement/hashing/consistent_hash.go

func (c *Consistent) search(key uint64) int {idx := sort.Search(len(c.sortedSet), func(i int) bool {return c.sortedSet[i] >= key
    })

    if idx >= len(c.sortedSet) {idx = 0}
    return idx
}

search 办法通过 sort.Search 依据 key 获取 idx

小结

dapr 的 Consistent 定义了 hosts、sortedSet、loadMap、totalLoad、sync.RWMutex 属性；它定义了 GetInternals、Add、Get、GetHost、GetLeast、search 等办法。

doc

• dapr